Files
govoplan-reporting/src/govoplan_reporting/backend/router.py
T

770 lines
26 KiB
Python

from __future__ import annotations
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
from govoplan_core.core.dataflows import dataflow_dataset_output
from govoplan_core.db.session import get_session
from govoplan_reporting.backend.definitions import (
READ_SCOPE,
WRITE_SCOPE,
ReportingDefinitionError,
create_definition,
definition_history,
get_definition,
list_definitions,
update_definition,
)
from govoplan_reporting.backend.drilldown import (
ReportingDrillError,
create_drill_context,
resolve_drill_context,
)
from govoplan_reporting.backend.execution import (
QUALITY_SCOPE,
RUN_SCOPE,
ReportingExecutionError,
ReportingExecutionFailure,
execute_report,
get_execution,
list_executions,
run_quality_plan,
)
from govoplan_reporting.backend.operations import (
IMPORT_SCOPE,
PUBLISH_SCOPE,
SCHEDULE_SCOPE,
ReportingOperationError,
assess_import,
delete_saved_view,
dispatch_due_schedules,
export_execution,
list_import_assessments,
list_publications,
list_saved_views,
list_schedules,
publish_execution,
upsert_saved_view,
upsert_schedule,
)
from govoplan_reporting.backend.provider_reports import (
ProviderReportError,
execute_provider_report,
export_provider_execution,
get_provider_execution,
list_provider_exports,
list_provider_reports,
provider_parameter_options,
)
from govoplan_reporting.backend.publication_targets import publication_target_catalog
from govoplan_reporting.backend.query_engine import ReportingQueryError
from govoplan_reporting.backend.schemas import (
DefinitionUpdateRequest,
DefinitionWriteRequest,
DrillContextCreateRequest,
ImportAssessmentRequest,
PublicationRequest,
ProviderReportExecutionRequest,
ProviderReportExportRequest,
QualityRunRequest,
ReportExecutionRequest,
SavedViewWriteRequest,
ScheduleWriteRequest,
)
def create_router(registry: object | None) -> APIRouter:
router = APIRouter(prefix="/reporting", tags=["reporting"])
@router.get("/provider-reports")
def api_list_provider_reports(
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, READ_SCOPE)
try:
return list_provider_reports(session, principal, registry=registry)
except (ProviderReportError, TypeError, ValueError) as exc:
raise _error(exc) from exc
@router.get(
"/provider-reports/{provider_id}/{report_id}/parameters/{parameter_key}/options"
)
def api_provider_parameter_options(
provider_id: str,
report_id: str,
parameter_key: str,
query: str = "",
limit: int = Query(default=100, ge=1, le=200),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, READ_SCOPE)
try:
options = provider_parameter_options(
session,
principal,
registry=registry,
provider_id=provider_id,
report_id=report_id,
parameter_key=parameter_key,
query=query,
limit=limit,
)
except (ProviderReportError, PermissionError, LookupError) as exc:
raise _error(exc) from exc
return {"options": [item.to_dict() for item in options]}
@router.post("/provider-reports/{provider_id}/{report_id}/executions")
def api_execute_provider_report(
provider_id: str,
report_id: str,
payload: ProviderReportExecutionRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, RUN_SCOPE)
try:
result = execute_provider_report(
session,
principal,
registry=registry,
provider_id=provider_id,
report_id=report_id,
**payload.model_dump(),
)
session.commit()
except (ProviderReportError, PermissionError, LookupError, ValueError) as exc:
session.rollback()
raise _error(exc) from exc
return result
@router.get("/provider-executions/{execution_id}")
def api_get_provider_execution(
execution_id: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, RUN_SCOPE)
try:
result = get_provider_execution(
session,
principal,
registry=registry,
execution_id=execution_id,
)
except (ProviderReportError, PermissionError, LookupError) as exc:
raise _error(exc) from exc
if result is None:
raise HTTPException(
status_code=404, detail="Provider report execution not found"
)
return result
@router.get("/provider-executions/{execution_id}/exports")
def api_list_provider_exports(
execution_id: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, RUN_SCOPE)
try:
return {
"exports": list(
list_provider_exports(
session,
principal,
registry=registry,
execution_id=execution_id,
)
)
}
except (ProviderReportError, PermissionError, LookupError) as exc:
raise _error(exc) from exc
@router.post("/provider-executions/{execution_id}/exports")
def api_export_provider_execution(
execution_id: str,
payload: ProviderReportExportRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> Response:
_require(principal, RUN_SCOPE)
try:
content, media_type, filename = export_provider_execution(
session,
principal,
registry=registry,
execution_id=execution_id,
**payload.model_dump(),
)
session.commit()
except (ProviderReportError, PermissionError, LookupError) as exc:
session.rollback()
raise _error(exc) from exc
return Response(
content=content,
media_type=media_type,
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
@router.get("/definitions")
def api_list_definitions(
definition_kind: list[str] | None = Query(default=None),
definition_status: list[str] | None = Query(default=None, alias="status"),
query: str = "",
offset: int = Query(default=0, ge=0),
limit: int = Query(default=100, ge=1, le=200),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, READ_SCOPE)
try:
records, total = list_definitions(
session,
principal,
definition_kinds=definition_kind,
statuses=definition_status,
query=query,
offset=offset,
limit=limit,
)
except ReportingDefinitionError as exc:
raise _error(exc) from exc
return {
"definitions": [item.to_dict() for item in records],
"total": total,
"offset": offset,
"limit": limit,
}
@router.post(
"/definitions",
status_code=status.HTTP_201_CREATED,
)
def api_create_definition(
payload: DefinitionWriteRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, Any]:
_require(principal, WRITE_SCOPE)
try:
record = create_definition(
session,
principal,
**payload.model_dump(),
)
session.commit()
except (ReportingDefinitionError, ValueError) as exc:
session.rollback()
raise _error(exc) from exc
return record.to_dict()
@router.get("/definitions/{definition_kind}/{definition_id}")
def api_get_definition(
definition_kind: str,
definition_id: str,
revision: int | None = Query(default=None, ge=1),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, Any]:
_require(principal, READ_SCOPE)
try:
record = get_definition(
session,
principal,
definition_kind=definition_kind,
definition_id=definition_id,
revision=revision,
)
except ReportingDefinitionError as exc:
raise _error(exc) from exc
if record is None:
raise HTTPException(
status_code=404, detail="Reporting definition not found"
)
return record.to_dict()
@router.patch("/definitions/{definition_kind}/{definition_id}")
def api_update_definition(
definition_kind: str,
definition_id: str,
payload: DefinitionUpdateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, Any]:
_require(principal, WRITE_SCOPE)
changes = payload.model_dump(
exclude={
"expected_revision",
"recorded_at",
"change_reason",
"idempotency_key",
},
exclude_unset=True,
)
try:
record = update_definition(
session,
principal,
definition_kind=definition_kind,
definition_id=definition_id,
expected_revision=payload.expected_revision,
recorded_at=payload.recorded_at,
change_reason=payload.change_reason,
idempotency_key=payload.idempotency_key,
changes=changes,
)
session.commit()
except (
ReportingDefinitionError,
PermissionError,
LookupError,
ValueError,
) as exc:
session.rollback()
raise _error(exc) from exc
return record.to_dict()
@router.get("/definitions/{definition_kind}/{definition_id}/history")
def api_definition_history(
definition_kind: str,
definition_id: str,
limit: int = Query(default=100, ge=1, le=200),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, READ_SCOPE)
records = definition_history(
session,
principal,
definition_kind=definition_kind,
definition_id=definition_id,
limit=limit,
)
if not records:
raise HTTPException(
status_code=404, detail="Reporting definition not found"
)
return {"revisions": [item.to_dict() for item in records]}
@router.get("/sources/dataflow")
def api_dataflow_sources(
query: str = "",
limit: int = Query(default=100, ge=1, le=200),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, WRITE_SCOPE)
provider = dataflow_dataset_output(registry)
if provider is None:
return {
"available": False,
"reason": "The Dataflow dataset provider is not enabled.",
"sources": [],
}
sources = provider.list_outputs(
session,
principal,
query=query,
limit=limit,
)
return {
"available": True,
"reason": None,
"sources": [
{
"pipeline_ref": item.pipeline_ref,
"name": item.name,
"description": item.description,
"revision": item.revision,
"definition_hash": item.definition_hash,
"status": item.status,
"updated_at": item.updated_at.isoformat()
if item.updated_at
else None,
"parameters": dict(item.parameters),
"provenance": dict(item.provenance),
}
for item in sources
],
}
@router.post("/reports/{report_id}/executions")
def api_execute_report(
report_id: str,
payload: ReportExecutionRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, RUN_SCOPE)
try:
result = execute_report(
session,
principal,
registry=registry,
report_id=report_id,
report_revision=payload.report_revision,
parameters=payload.parameters,
query=payload.query,
idempotency_key=payload.idempotency_key,
)
session.commit()
except ReportingExecutionFailure as exc:
session.commit()
raise HTTPException(
status_code=422,
detail={"message": str(exc), "execution_id": exc.execution_id},
) from exc
except (
ReportingExecutionError,
ReportingQueryError,
PermissionError,
LookupError,
ValueError,
) as exc:
session.rollback()
raise _error(exc) from exc
return result
@router.get("/reports/{report_id}/executions")
def api_list_executions(
report_id: str,
limit: int = Query(default=100, ge=1, le=200),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, RUN_SCOPE)
return {
"executions": list(
list_executions(
session,
principal,
report_id=report_id,
limit=limit,
registry=registry,
)
)
}
@router.get("/executions/{execution_id}")
def api_get_execution(
execution_id: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, RUN_SCOPE)
result = get_execution(
session,
principal,
execution_id=execution_id,
registry=registry,
)
if result is None:
raise HTTPException(status_code=404, detail="Reporting execution not found")
return result
@router.post("/executions/{execution_id}/drill-contexts", status_code=201)
def api_create_drill_context(
execution_id: str,
payload: DrillContextCreateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, RUN_SCOPE)
try:
result = create_drill_context(
session,
principal,
registry=registry,
execution_id=execution_id,
aggregate_row=payload.aggregate_row,
limit=payload.limit,
)
session.commit()
except (
ReportingDrillError,
ReportingExecutionError,
PermissionError,
LookupError,
) as exc:
session.rollback()
raise _error(exc) from exc
return result
@router.get("/drill-contexts/{token}")
def api_resolve_drill_context(
token: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, RUN_SCOPE)
try:
result = resolve_drill_context(
session,
principal,
registry=registry,
token=token,
)
session.commit()
except (
ReportingDrillError,
ReportingExecutionError,
PermissionError,
LookupError,
) as exc:
session.rollback()
raise _error(exc) from exc
return result
@router.get("/executions/{execution_id}/export")
def api_export_execution(
execution_id: str,
format: str = Query(default="csv", pattern="^(csv|json)$"),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> Response:
_require(principal, RUN_SCOPE)
try:
content, media_type, filename = export_execution(
session,
principal,
execution_id=execution_id,
format=format,
registry=registry,
)
except (ReportingOperationError, LookupError) as exc:
raise _error(exc) from exc
return Response(
content=content,
media_type=media_type,
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
@router.post("/executions/{execution_id}/publications")
def api_publish_execution(
execution_id: str,
payload: PublicationRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, PUBLISH_SCOPE)
try:
result = publish_execution(
session,
principal,
registry=registry,
execution_id=execution_id,
**payload.model_dump(),
)
session.commit()
except (ReportingOperationError, PermissionError, LookupError) as exc:
session.commit()
raise _error(exc) from exc
return result
@router.get("/publication-targets")
def api_publication_targets(
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, PUBLISH_SCOPE)
return {"targets": list(publication_target_catalog(registry))}
@router.get("/publications")
def api_list_publications(
execution_id: str | None = None,
limit: int = Query(default=100, ge=1, le=200),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, PUBLISH_SCOPE)
return {
"publications": list(
list_publications(
session,
principal,
execution_id=execution_id,
limit=limit,
registry=registry,
)
)
}
@router.get("/reports/{report_id}/saved-views")
def api_list_saved_views(
report_id: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, READ_SCOPE)
return {
"views": list(list_saved_views(session, principal, report_id=report_id))
}
@router.put("/reports/{report_id}/saved-views/{view_id}")
def api_upsert_saved_view(
report_id: str,
view_id: str,
payload: SavedViewWriteRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, READ_SCOPE)
if view_id != payload.view_id:
raise HTTPException(status_code=400, detail="Saved-view identifiers differ")
try:
result = upsert_saved_view(
session,
principal,
report_id=report_id,
**payload.model_dump(),
)
session.commit()
except (ReportingOperationError, PermissionError, LookupError) as exc:
session.rollback()
raise _error(exc) from exc
return result
@router.delete("/saved-views/{view_id}", status_code=204)
def api_delete_saved_view(
view_id: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> Response:
_require(principal, READ_SCOPE)
try:
deleted = delete_saved_view(session, principal, view_id=view_id)
session.commit()
except PermissionError as exc:
session.rollback()
raise _error(exc) from exc
if not deleted:
raise HTTPException(status_code=404, detail="Saved view not found")
return Response(status_code=204)
@router.get("/schedules")
def api_list_schedules(
report_id: str | None = None,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, SCHEDULE_SCOPE)
return {
"schedules": list(list_schedules(session, principal, report_id=report_id))
}
@router.put("/schedules/{schedule_id}")
def api_upsert_schedule(
schedule_id: str,
payload: ScheduleWriteRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, SCHEDULE_SCOPE)
if schedule_id != payload.schedule_id:
raise HTTPException(status_code=400, detail="Schedule identifiers differ")
try:
result = upsert_schedule(session, principal, **payload.model_dump())
session.commit()
except (ReportingOperationError, PermissionError, LookupError) as exc:
session.rollback()
raise _error(exc) from exc
return result
@router.post("/schedules/dispatch")
def api_dispatch_schedules(
limit: int = Query(default=20, ge=1, le=100),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, SCHEDULE_SCOPE)
result = dispatch_due_schedules(
session,
principal,
registry=registry,
now=None,
limit=limit,
)
session.commit()
return result
@router.post("/quality-plans/{quality_plan_id}/runs")
def api_run_quality_plan(
quality_plan_id: str,
payload: QualityRunRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, QUALITY_SCOPE)
try:
result = run_quality_plan(
session,
principal,
registry=registry,
quality_plan_id=quality_plan_id,
quality_plan_revision=payload.quality_plan_revision,
parameters=payload.parameters,
)
session.commit()
except (ReportingExecutionError, PermissionError, LookupError) as exc:
session.rollback()
raise _error(exc) from exc
return result
@router.post("/imports/assessments", status_code=201)
def api_assess_import(
payload: ImportAssessmentRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, IMPORT_SCOPE)
try:
result = assess_import(session, principal, **payload.model_dump())
session.commit()
except (ReportingOperationError, PermissionError) as exc:
session.rollback()
raise _error(exc) from exc
return result
@router.get("/imports/assessments")
def api_list_import_assessments(
limit: int = Query(default=100, ge=1, le=200),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, IMPORT_SCOPE)
return {
"assessments": list(
list_import_assessments(session, principal, limit=limit)
)
}
return router
def _require(principal: ApiPrincipal, scope: str) -> None:
if not has_scope(principal, scope):
raise HTTPException(status_code=403, detail=f"Missing scope: {scope}")
def _error(exc: Exception) -> HTTPException:
message = str(exc)
lowered = message.casefold()
if isinstance(exc, LookupError):
code = 404
elif isinstance(exc, PermissionError):
code = 403
elif any(word in lowered for word in ("conflict", "already", "stale")):
code = 409
elif "unavailable" in lowered or "not enabled" in lowered:
code = 503
else:
code = 400
return HTTPException(status_code=code, detail=message)
__all__ = ["create_router"]