Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 56 additions & 19 deletions echo/server/dembrane/api/project.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,14 @@ class BffProjectsHomeResponse(BaseModel):


_HOME_FIELDS = [
"id", "name", "updated_at", "language", "pin_order", "count(conversations)",
"id",
"name",
"updated_at",
"language",
"pin_order",
"count(conversations)",
]
_HOME_FIELDS_WITHOUT_PIN_ORDER = [field for field in _HOME_FIELDS if field != "pin_order"]


def _build_project_summary(raw: dict) -> BffProjectSummary:
Expand Down Expand Up @@ -100,13 +106,17 @@ async def get_projects_home(
fields = list(_HOME_FIELDS)
if auth.is_admin:
fields.extend(["directus_user_id.first_name", "directus_user_id.email"])
fallback_fields = list(_HOME_FIELDS_WITHOUT_PIN_ORDER)
if auth.is_admin:
fallback_fields.extend(["directus_user_id.first_name", "directus_user_id.email"])

# Fetch pinned projects (always, regardless of search)
# Admins see only their own pins; non-admins see all (Directus permissions handle scoping)
pin_filter: dict[str, Any] = {"pin_order": {"_nnull": True}}
if auth.is_admin:
pin_filter["directus_user_id"] = {"_eq": auth.user_id}

supports_pin_order = True
pinned_raw = await run_in_thread_pool(
client.get_items,
"project",
Expand All @@ -122,10 +132,12 @@ async def get_projects_home(
if not isinstance(pinned_raw, list):
logger.warning("get_items returned non-list for pinned projects: %s", pinned_raw)
pinned_raw = []
supports_pin_order = False
pinned = [_build_project_summary(p) for p in pinned_raw]

# Parse owner: prefix from search string (admin only)
import re

owner_term: Optional[str] = None
text_search: Optional[str] = search
if search and auth.is_admin:
Expand All @@ -144,8 +156,9 @@ async def get_projects_home(
}

# Build query for paginated project list
list_fields = fields if supports_pin_order else fallback_fields
query: dict = {
"fields": fields,
"fields": list_fields,
"sort": ["-updated_at"],
"limit": limit + 1,
"offset": offset,
Expand All @@ -162,7 +175,17 @@ async def get_projects_home(
)
if not isinstance(projects_raw, list):
logger.warning("get_items returned non-list for projects: %s", projects_raw)
projects_raw = []
if supports_pin_order:
supports_pin_order = False
query["fields"] = fallback_fields
projects_raw = await run_in_thread_pool(
client.get_items,
"project",
{"query": query},
)
if not isinstance(projects_raw, list):
logger.warning("fallback get_items returned non-list for projects: %s", projects_raw)
projects_raw = []

has_more = len(projects_raw) > limit
projects = [_build_project_summary(p) for p in projects_raw[:limit]]
Expand Down Expand Up @@ -609,9 +632,13 @@ async def create_report(
if not is_scheduled:
# Dispatch background task immediately
task_create_report.send(project_id, report["id"], language, body.user_instructions or "")
logger.info(f"Report generation task dispatched for project {project_id}, report {report['id']}")
logger.info(
f"Report generation task dispatched for project {project_id}, report {report['id']}"
)
else:
logger.info(f"Report {report['id']} scheduled for {body.scheduled_at} for project {project_id}")
logger.info(
f"Report {report['id']} scheduled for {body.scheduled_at} for project {project_id}"
)

return report

Expand All @@ -621,6 +648,7 @@ def _extract_report_title(content: Optional[str]) -> Optional[str]:
if not content:
return None
import re

match = re.search(r"^#\s+(.+)$", content, re.MULTILINE)
return match.group(1).strip() if match else None

Expand All @@ -642,22 +670,32 @@ async def list_project_reports(
"project_id": {"_eq": project_id},
"status": {"_in": ["archived", "published", "scheduled", "draft"]},
},
"fields": ["id", "status", "date_created", "language", "user_instructions", "content", "scheduled_at"],
"fields": [
"id",
"status",
"date_created",
"language",
"user_instructions",
"content",
"scheduled_at",
],
"sort": ["-date_created"],
}
},
)
result = []
for r in (reports or []):
result.append({
"id": r["id"],
"status": r.get("status"),
"date_created": r.get("date_created"),
"language": r.get("language"),
"user_instructions": r.get("user_instructions"),
"scheduled_at": r.get("scheduled_at"),
"title": _extract_report_title(r.get("content")),
})
for r in reports or []:
result.append(
{
"id": r["id"],
"status": r.get("status"),
"date_created": r.get("date_created"),
"language": r.get("language"),
"user_instructions": r.get("user_instructions"),
"scheduled_at": r.get("scheduled_at"),
"title": _extract_report_title(r.get("content")),
}
)
return result


Expand Down Expand Up @@ -856,6 +894,7 @@ async def get_report_views(

# Recent views (last 10 minutes)
from datetime import datetime, timezone, timedelta

ten_mins_ago = (datetime.now(timezone.utc) - timedelta(minutes=10)).isoformat()
recent_metrics = await run_in_thread_pool(
directus.get_items,
Expand Down Expand Up @@ -976,9 +1015,7 @@ async def _generate_events() -> AsyncIterator[str]:
# Check if report is already done before subscribing
from dembrane.directus import directus

report = await run_in_thread_pool(
directus.get_item, "project_report", str(report_id)
)
report = await run_in_thread_pool(directus.get_item, "project_report", str(report_id))
if not report or str(report.get("project_id")) != project_id:
yield f"event: progress\ndata: {json.dumps({'type': 'failed', 'message': 'Report not found'})}\n\n"
return
Expand Down
83 changes: 83 additions & 0 deletions echo/server/tests/api/test_projects_home.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,83 @@
import os

import pytest

os.environ.setdefault("DIRECTUS_SECRET", "test-secret")
os.environ.setdefault("DIRECTUS_TOKEN", "test-token")
os.environ.setdefault("DATABASE_URL", "postgresql://localhost/test")
os.environ.setdefault("REDIS_URL", "redis://localhost:6379/0")
os.environ.setdefault("STORAGE_S3_BUCKET", "test-bucket")
os.environ.setdefault("STORAGE_S3_ENDPOINT", "https://example.com")
os.environ.setdefault("STORAGE_S3_KEY", "test-key")
os.environ.setdefault("STORAGE_S3_SECRET", "test-secret")

import dembrane.api.project as project_api
from dembrane.api.dependency_auth import DirectusSession


def _auth(client) -> DirectusSession:
return DirectusSession(
user_id="user-1",
is_admin=True,
access_token="token-1",
client=client,
)


@pytest.mark.asyncio
async def test_get_projects_home_falls_back_when_pin_order_is_unavailable(monkeypatch) -> None:
async def _fake_run_in_thread_pool(func, *args, **kwargs): # noqa: ANN001, ANN002, ANN003
return func(*args, **kwargs)

class _FakeClient:
def __init__(self) -> None:
self.calls: list[dict] = []

def get_items(self, collection_name: str, payload: dict) -> list[dict] | dict[str, str]:
assert collection_name == "project"
self.calls.append(payload)
query = payload["query"]

if "aggregate" in query:
return [{"count": {"id": "21"}}]

if "pin_order" in query.get("fields", []):
return {"error": 'You don\'t have permission to access field "pin_order"'}

return [
{
"id": "project-1",
"name": "Visible project",
"updated_at": "2026-03-19T17:00:00Z",
"language": "en",
"conversations_count": "2",
"directus_user_id": {
"first_name": "Admin",
"email": "admin@dembrane.com",
},
}
]

client = _FakeClient()
monkeypatch.setattr(project_api, "run_in_thread_pool", _fake_run_in_thread_pool)

response = await project_api.get_projects_home(
auth=_auth(client),
search=None,
offset=0,
limit=15,
)

assert response.is_admin is True
assert response.total_count == 21
assert response.has_more is False
assert response.pinned == []
assert len(response.projects) == 1
assert response.projects[0].id == "project-1"
assert response.projects[0].pin_order is None
assert any("pin_order" in call["query"].get("fields", []) for call in client.calls)
assert any(
"pin_order" not in call["query"].get("fields", [])
for call in client.calls
if "aggregate" not in call["query"]
)
Loading