Native HTTP mapper for the read-only graph audit.
mount_graph_audit_route(router: Any, *, audit: Any, resolve: Any, header: Any, http_exception: Any, request_type: Any, working: Any) -> None
Mount only when a graph audit service was explicitly supplied.
Source code in src/symfonic/platform/transport_graph_audit.py
| def mount_graph_audit_route(
router: Any, *, audit: Any, resolve: Any, header: Any, http_exception: Any,
request_type: Any, working: Any,
) -> None:
"""Mount only when a graph audit service was explicitly supplied."""
from symfonic.capabilities.memory.graph_owner_reads import OwnerReadUnavailable
@router.get("/memory/audit")
async def graph_audit(
request: request_type,
min_edge_weight: float = 1.5,
x_tenant_id: str | None = header(default=None), # noqa: B008
authorization: str | None = header(default=None), # noqa: B008
) -> dict[str, Any]:
scope, guard = await working(x_tenant_id, authorization, request=request)
v0_scope = memory_scope(scope) # outside the catch-all: 501 stays typed
try:
return await audit.audit(
v0_scope, min_edge_weight=min_edge_weight,
admit=guard.admission("graph"),
)
except OwnerReadUnavailable as unavailable:
raise http_exception(503, "owner-restricted graph audit unavailable") from unavailable
except http_exception:
raise
except Exception as unavailable: # noqa: BLE001 - keep storage details private
logger.error("Graph audit failed")
raise http_exception(500, "Internal server error") from unavailable
|