routes

stac_fastapi.api.routes

Route factories.

Scope

Bases: TypedDict

More strict version of Starlette's Scope.

Source code in stac_fastapi/api/stac_fastapi/api/routes.py

class Scope(TypedDict, total=False):
    """More strict version of Starlette's Scope."""

    # https://github.com/encode/starlette/blob/6af5c515e0a896cbf3f86ee043b88f6c24200bcf/starlette/types.py#L3
    path: str
    method: str
    type: str | None

add_direct_response

add_direct_response(app: FastAPI) -> None

Setup FastAPI application's endpoints to return Response Object directly, avoiding Pydantic validation and FastAPI (slow) serialization.

ref: gist.github.com/Zaczero/00f3a2679ebc0a25eb938ed82bc63553

Source code in stac_fastapi/api/stac_fastapi/api/routes.py

def add_direct_response(app: FastAPI) -> None:
    """
    Setup FastAPI application's endpoints to return Response Object directly, avoiding
    Pydantic validation and FastAPI (slow) serialization.

    ref: https://gist.github.com/Zaczero/00f3a2679ebc0a25eb938ed82bc63553
    """

    def wrap_endpoint(endpoint: Callable, cls: type[Response]):
        @functools.wraps(endpoint)
        async def wrapper(*args, **kwargs):
            content = await endpoint(*args, **kwargs)
            return content if isinstance(content, Response) else cls(content)

        return wrapper

    for route in app.routes:
        if not isinstance(route, APIRoute):
            continue

        response_class = route.response_class
        if isinstance(response_class, DefaultPlaceholder):
            response_class = response_class.value

        if issubclass(response_class, Response):
            route.endpoint = wrap_endpoint(route.endpoint, response_class)
            route.dependant = get_dependant(path=route.path_format, call=route.endpoint)
            route.app = request_response(route.get_route_handler())

add_route_dependencies

add_route_dependencies(
    routes: list[BaseRoute], scopes: list[Scope], dependencies: list[Depends]
) -> None

Add dependencies to routes.

Allows a developer to add dependencies to a route after the route has been defined.

"*" can be used for path or method to match all allowed routes.

Returns:

• None –

  None

Source code in stac_fastapi/api/stac_fastapi/api/routes.py

def add_route_dependencies(
    routes: list[BaseRoute], scopes: list[Scope], dependencies: list[params.Depends]
) -> None:
    """Add dependencies to routes.

    Allows a developer to add dependencies to a route after the route has been
    defined.

    "*" can be used for path or method to match all allowed routes.

    Returns:
        None
    """
    for route in routes:
        if hasattr(route, "original_router"):
            add_route_dependencies(route.original_router.routes, scopes, dependencies)
            continue

        if hasattr(route, "routes") and route.routes:
            add_route_dependencies(route.routes, scopes, dependencies)
            continue

        if not _is_endpoint_route(route):
            continue

        _apply_dependencies_to_route(route, scopes, dependencies)

create_async_endpoint

create_async_endpoint(
    func: Callable, request_model: type[APIRequest] | type[BaseModel] | dict
) -> Callable[[Any, Any], Awaitable[Any]]

Wrap a function in a coroutine which may be used to create a FastAPI endpoint.

Synchronous functions are executed asynchronously using a background thread.

Source code in stac_fastapi/api/stac_fastapi/api/routes.py

def create_async_endpoint(
    func: Callable,
    request_model: type[APIRequest] | type[BaseModel] | dict,
) -> Callable[[Any, Any], Awaitable[Any]]:
    """Wrap a function in a coroutine which may be used to create a FastAPI endpoint.

    Synchronous functions are executed asynchronously using a background thread.
    """

    if not inspect.iscoroutinefunction(func):
        func = sync_to_async(func)

    _endpoint: Callable[[Any, Any], Awaitable[Any]]

    if isinstance(request_model, dict):

        async def _endpoint(request: Request, request_data: dict[str, Any]):
            """Endpoint."""
            return _wrap_response(await func(request_data, request=request))

    elif issubclass(request_model, APIRequest):

        async def _endpoint(request: Request, request_data=Depends(request_model)):  # type: ignore
            """Endpoint."""
            return _wrap_response(await func(request=request, **request_data.kwargs()))

    elif issubclass(request_model, BaseModel):

        async def _endpoint(request: Request, request_data: request_model):  # type: ignore
            """Endpoint."""
            return _wrap_response(await func(request_data, request=request))

    else:
        raise ValueError(f"Unsupported type for request model {type(request_model)}")

    return _endpoint

sync_to_async

sync_to_async(func)

Run synchronous function asynchronously in a background thread.

Source code in stac_fastapi/api/stac_fastapi/api/routes.py

def sync_to_async(func):
    """Run synchronous function asynchronously in a background thread."""

    @functools.wraps(func)
    async def run(*args, **kwargs):
        return await run_in_threadpool(func, *args, **kwargs)

    return run