Skip to content
Draft
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
33 changes: 31 additions & 2 deletions apps/api/app/api/v1/routes/documents.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
from sqlalchemy.ext.asyncio import AsyncSession

from shared.core.database import get_db
from shared.core.exceptions.domain_exceptions import NotFoundException
from shared.core.exceptions.domain_exceptions import NotFoundException, ValidationException
from shared.models.schemas.retrieval_namespace import normalize_retrieval_namespace

router = APIRouter(tags=["Documents"])
Expand Down Expand Up @@ -44,12 +44,20 @@ async def _archive_document_response(
@router.get("")
async def list_documents(
namespace: str | None = Query(None, max_length=255),
all_namespaces: bool = Query(
False, description="List user-owned documents across all namespaces",
),
page: int = Query(1, ge=1, description="Page number"),
page_size: int = Query(50, ge=1, le=200, description="Items per page"),
current_user: CurrentUser = Depends(with_current_user),
db: AsyncSession = Depends(get_db),
):
effective_namespace = normalize_retrieval_namespace(namespace)
if all_namespaces and namespace is not None:
raise ValidationException(
user_message="namespace cannot be combined with all_namespaces",
violations=[{"field": "namespace", "description": "Omit when all_namespaces is true"}],
)
effective_namespace = None if all_namespaces else normalize_retrieval_namespace(namespace)
response = await _document_service.list_documents(
db,
user_id=current_user.user_id,
Expand All @@ -60,6 +68,27 @@ async def list_documents(
return response


@router.get("/{document_id}/jobs")
async def list_document_jobs(
document_id: str,
page: int = Query(1, ge=1, description="Page number"),
page_size: int = Query(50, ge=1, le=200, description="Items per page"),
current_user: CurrentUser = Depends(with_current_user),
db: AsyncSession = Depends(get_db),
):
"""Read processing history, including failed attempts and archived documents."""
response = await _document_service.list_document_jobs(
db, user_id=current_user.user_id, document_id=document_id,
page=page, page_size=page_size,
)
if response is None:
raise NotFoundException(
resource="Document", resource_id=document_id,
internal_message="Document not found",
)
return response


@router.get("/{document_id}")
async def get_document(
document_id: str,
Expand Down
62 changes: 57 additions & 5 deletions apps/api/app/repositories/document_repository.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
from datetime import datetime, timezone
from typing import Sequence, cast

from sqlalchemy import func, select, true
from sqlalchemy import and_, func, or_, select, true
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import noload

Expand All @@ -20,6 +20,27 @@

DocumentChunkRow = tuple[DocumentChunk, DocumentSection | None, JobResult]
DocumentJobRevisionRow = tuple[Document, JobResult, Job]
DocumentJobHistoryRow = tuple[Job, JobResult | None]


def _document_job_history_query(*, user_id: str, document_id: str):
# A published result owns the canonical link. Metadata preserves attempts
# that failed or are still processing, and legacy results without a link.
return (
select(Job, JobResult)
.outerjoin(JobResult, JobResult.job_id == Job.job_id)
.where(Job.user_id == user_id)
.where(
or_(
JobResult.document_id == document_id,
and_(
JobResult.document_id.is_(None),
JobResult.demo_document_id.is_(None),
Job.job_metadata["document_id"].as_string() == document_id,
),
)
)
)


class DocumentRepository:
Expand All @@ -28,15 +49,15 @@ async def list_by_user_namespace(
db: AsyncSession,
*,
user_id: str,
namespace: str,
namespace: str | None,
limit: int,
offset: int,
) -> Sequence[Document]:
corpusStorage: CorpusStorage = CorpusStorage.resolve_namespace(namespace)
result = await db.execute(
select(corpusStorage.Document)
.where(corpusStorage.Document.user_id == corpusStorage.resolve_owner(user_id))
.where(corpusStorage.Document.namespace == namespace)
.where(corpusStorage.Document.namespace == namespace if namespace is not None else true())
.where(corpusStorage.Document.status != "archived")
.order_by(corpusStorage.Document.updated_at.desc(), corpusStorage.Document.document_id.asc())
.limit(limit)
Expand All @@ -49,13 +70,13 @@ async def count_by_user_namespace(
db: AsyncSession,
*,
user_id: str,
namespace: str,
namespace: str | None,
) -> int:
corpusStorage: CorpusStorage = CorpusStorage.resolve_namespace(namespace)
result = await db.execute(
select(func.count(corpusStorage.Document.document_id))
.where(corpusStorage.Document.user_id == corpusStorage.resolve_owner(user_id))
.where(corpusStorage.Document.namespace == namespace)
.where(corpusStorage.Document.namespace == namespace if namespace is not None else true())
.where(corpusStorage.Document.status != "archived")
)
return int(result.scalar_one())
Expand All @@ -76,6 +97,37 @@ async def get_document(
)
return result.scalar_one_or_none()

async def count_document_jobs(
self,
db: AsyncSession,
*,
user_id: str,
document_id: str,
) -> int:
result = await db.execute(
_document_job_history_query(user_id=user_id, document_id=document_id)
.with_only_columns(func.count(Job.job_id), maintain_column_froms=True)
)
return int(result.scalar_one())

async def list_document_jobs(
self,
db: AsyncSession,
*,
user_id: str,
document_id: str,
limit: int,
offset: int,
) -> Sequence[DocumentJobHistoryRow]:
result = await db.execute(
_document_job_history_query(user_id=user_id, document_id=document_id)
.options(noload(JobResult.chunks), noload(JobResult.job))
.order_by(Job.created_at.desc(), Job.job_id.asc())
.limit(limit)
.offset(offset)
)
return cast(Sequence[DocumentJobHistoryRow], result.all())

async def get_current_document_job_revision(
self,
db: AsyncSession,
Expand Down
56 changes: 55 additions & 1 deletion apps/api/app/services/documents/lifecycle_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -265,7 +265,7 @@ async def list_documents(
db: AsyncSession,
*,
user_id: str,
namespace: str,
namespace: str | None,
page: int,
page_size: int,
) -> dict[str, Any]:
Expand All @@ -292,6 +292,60 @@ async def list_documents(
},
}

async def list_document_jobs(
self,
db: AsyncSession,
*,
user_id: str,
document_id: str,
page: int,
page_size: int,
) -> dict[str, Any] | None:
document = await self._repository.get_document(
db, user_id=user_id, document_id=document_id,
)
if document is None:
return None

# Shared demo documents do not expose their server-owned job ledger.
is_demo = CorpusStorage.resolve_document(document_id).is_demo
total = 0 if is_demo else await self._repository.count_document_jobs(
db, user_id=user_id, document_id=document_id,
)
rows = [] if is_demo else await self._repository.list_document_jobs(
db, user_id=user_id, document_id=document_id,
limit=page_size, offset=(page - 1) * page_size,
)
return {
"document_id": document.document_id,
"namespace": document.namespace,
"jobs": [
{
"job_id": job.job_id,
"job_type": job.job_type,
"status": job.status,
"source_type": job.source_type,
"error_code": job.error_code,
"page_count": job.page_count,
"credits_charged": job.credits_charged,
"billing_status": job.billing_status,
"created_at": _datetime_payload(job.created_at),
"updated_at": _datetime_payload(job.updated_at),
"job_result_id": job_result.id if job_result else None,
"is_current_revision": bool(
job_result and job_result.id == document.current_job_result_id
),
}
for job, job_result in rows
],
"pagination": {
"page": page,
"page_size": page_size,
"total": total,
"total_pages": math.ceil(total / page_size) if total else 0,
},
}

async def list_document_chunks(
self,
db: AsyncSession,
Expand Down
Loading