Skip to content

symfonic.platform.transport_graph_paths

transport_graph_paths

The native graph path HTTP mapper, sharing the transport scope resolver.

mount_graph_path_route

mount_graph_path_route(router: Any, *, graph: Any, resolve: Any, working: Any) -> None

Mount the optional graph path reader without constructing a turn host.

Source code in src/symfonic/platform/transport_graph_paths.py
def mount_graph_path_route(router: Any, *, graph: Any, resolve: Any, working: Any) -> None:
    """Mount the optional graph path reader without constructing a turn host."""
    from fastapi import Header, HTTPException, Request

    from symfonic.capabilities.memory.graph_paths import GraphPathUnavailable

    @router.get("/graph/paths")
    async def find_path(
        start: str,
        end: str,
        request: Request,
        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)
        try:
            return await graph.shortest_path(
                memory_scope(scope), start, end, admit=guard.admission("graph"),
            )
        except LookupError as missing:
            raise HTTPException(404, "graph path endpoint not found") from missing
        except ValueError as invalid:
            raise HTTPException(400, "invalid graph path request") from invalid
        except GraphPathUnavailable as unavailable:
            raise HTTPException(503, "graph path unavailable") from unavailable