rs_server_common/fastapi_app.md

<< Back to index

Init the FastAPI application.

flatten_routes(routes)

Recursively resolve fastapi.routing._IncludedRouter wrappers into concrete routes.

Since fastapi 0.141 (starlette 1.x), app.include_router() no longer flattens the included router's routes eagerly into the parent router: it stores a lazy _IncludedRouter wrapper instead. Without this, routes nested behind an included router (e.g. /auxip/queryables) are silently skipped below, and the authenticate dependency never gets attached to them.

Source code in docs/rs-server/services/common/rs_server_common/fastapi_app.py

def flatten_routes(routes: list[BaseRoute]) -> list[BaseRoute]:
    """
    Recursively resolve fastapi.routing._IncludedRouter wrappers into concrete routes.

    Since fastapi 0.141 (starlette 1.x), app.include_router() no longer flattens the included
    router's routes eagerly into the parent router: it stores a lazy _IncludedRouter wrapper
    instead. Without this, routes nested behind an included router (e.g. /auxip/queryables) are
    silently skipped below, and the authenticate dependency never gets attached to them.
    """
    flat: list[BaseRoute] = []
    for route in routes:
        if isinstance(route, _IncludedRouter):
            flat.extend(flatten_routes(route.original_router.routes))
        else:
            flat.append(route)
    return flat

init_app(api_version, routers, router_prefix='')

Init the FastAPI application. See: https://praciano.com.br/fastapi-and-async-sqlalchemy-20-with-pytest-done-right.html

Parameters:

Name	Type	Description	Default
api_version	str	version of our application (not the version of the OpenAPI specification	required
routers	list[APIRouter]	list of FastAPI routers to add to the application.	required
router_prefix	str	used by stac_fastapi	''

Source code in docs/rs-server/services/common/rs_server_common/fastapi_app.py

@typing.no_type_check
def init_app(  # pylint: disable=too-many-locals, too-many-statements
    api_version: str,
    routers: list[APIRouter],
    router_prefix: str = "",
) -> FastAPI:  # pylint: disable=too-many-arguments
    """
    Init the FastAPI application.
    See: https://praciano.com.br/fastapi-and-async-sqlalchemy-20-with-pytest-done-right.html

    Args:
        api_version (str): version of our application (not the version of the OpenAPI specification
        nor the version of FastAPI being used)
        routers (list[APIRouter]): list of FastAPI routers to add to the application.
        router_prefix (str): used by stac_fastapi
    """

    @asynccontextmanager
    async def lifespan(*_):
        """Automatically executed when starting and stopping the FastAPI server."""

        ###########
        # STARTUP #
        ###########

        # Init objects for dependency injection
        settings.set_http_client(httpx.AsyncClient(timeout=DEFAULT_TIMEOUT_CONFIG))

        yield

        ############
        # SHUTDOWN #
        ############

        # Close objects for dependency injection
        await settings.del_http_client()

    # Init the FastAPI application
    app = FastAPI(title="RS-Server", version=api_version, lifespan=lifespan, **docs_params(router_prefix))

    # Configure OpenTelemetry
    init_opentelemetry.init_traces(app, settings.SERVICE_NAME)

    # Init a pgstac client for adgs and cadip.
    # TODO: remove this when adgs and cadip switch to a stac_fastapi application.
    # Example taken from: https://github.com/stac-utils/stac-fastapi-pgstac/blob/main/tests/api/test_api.py
    app.state.router_prefix = router_prefix  # NOTE: maybe we should keep this one
    sortables_client = RSSortablesClient(router_prefix)
    extensions = [  # no transactions because we don't update the database
        # TransactionExtension(client=TransactionsClient(), settings=api_settings),
        QueryExtension(),
        SortExtension(client=sortables_client),
        FieldsExtension(),
        FilterExtension(client=FiltersClient()),
        PaginationExtension(),
        # BulkTransactionExtension(client=BulkTransactionsClient()),
    ]
    search_post_request_model = create_post_request_model(extensions, base_model=PgstacSearch)
    app.state.pgstac_client = CoreCrudClient(pgstac_search_model=search_post_request_model)

    # patch the pgstac_client.landing_page method to add "rel": "child" link for each collection
    async def patched_landing_page(self, request, **kwargs):
        # Call the original method
        original = await CoreCrudClient.landing_page(self, request=request, **kwargs)

        # Get base from 'self' link
        base = next((link["href"] for link in original["links"] if link.get("rel") == "self"), "").rstrip("/") + "/"

        # Fetch collections
        collections = (await self.all_collections(request=request)).get("collections", [])

        # Add Sortables link
        original["links"].append(
            {
                "rel": "http://www.opengis.net/def/rel/ogc/1.0/sortables",
                "type": "application/schema+json",
                "title": "Sortables",
                "href": urljoin(base, "sortables"),
            },
        )

        # Append rel="child" links
        original["links"] += [
            {
                "rel": "child",
                "type": "application/json",
                "title": collection.get("title") or collection["id"],
                "href": urljoin(base, f"collections/{collection['id']}"),
            }
            for collection in collections
        ]

        return original

    # Monkey patch the pgstac_client.landing_page method
    app.state.pgstac_client.landing_page = MethodType(patched_landing_page, app.state.pgstac_client)

    # TODO: remove this when adgs and cadip switch to a stac_fastapi application.
    app.state.pgstac_client.extensions = extensions
    for ext in extensions:
        ext.register(app)
    app.state.pgstac_client.title = app.title
    app.state.pgstac_client.description = app.description
    # Implement the /search endpoints by simulating a StacApi object, TODO remove this also
    app.settings = State()
    app.settings.enable_response_models = False
    app.settings.use_api_hydrate = False
    app.state.settings = app.settings
    app.client = app.state.pgstac_client
    app.search_get_request_model = create_get_request_model(extensions)
    app.search_post_request_model = search_post_request_model
    app.router.prefix = router_prefix  # TODO should be used by other endpoints ?
    StacApi.register_get_search(app)
    StacApi.register_post_search(app)
    app.router.prefix = ""
    if settings.CLUSTER_MODE:
        scopes = []  # One scope for each Router path and method
        for route in flatten_routes(app.router.routes):
            if not isinstance(route, APIRoute):
                continue
            for method_ in route.methods:
                scopes.append({"path": route.path, "method": method_})
        add_route_dependencies(app.router.routes, scopes=scopes, dependencies=[Depends(authenticate)])
    # TODO: title and description must be set using the env vars
    # CATALOG_METADATA_TITLE and CATALOG_METADATA_DESCRIPTION
    service = router_prefix.strip("/").title()
    app.state.pgstac_client.title = f"RS-PYTHON {service} collections"
    app.state.pgstac_client.description = f"{service} collections of Copernicus Reference System Python"
    # By default FastAPI will return 422 status codes for invalid requests
    # But the STAC api spec suggests returning a 400 in this case
    # TODO remove this also
    add_exception_handlers(app, {})

    dependencies = []
    if settings.CLUSTER_MODE:

        # Apply middlewares and authentication routes to the FastAPI application.
        # This also adds the SessionMiddleware
        apply_middlewares(app)

        # Add the api key / oauth2 security: the user must provide
        # an api key (generated from the apikey manager) or authenticate to the
        # oauth2 service (keycloak) to access the endpoints
        dependencies.append(Depends(authenticate))

    # Add all the input routers (and not the oauth2 nor technical routers) to a single bigger router
    # to which we add the authentication dependency.
    need_auth_router = APIRouter(dependencies=dependencies)
    for router in routers:
        need_auth_router.include_router(router)

    # Add routers to the FastAPI app
    app.include_router(need_auth_router)
    app.include_router(technical_router)

    # Add middlewares. When sending a request, the middleware order must be:
    # Health -> CORS -> HandleExceptions -> Session -> [any other middlewares ...]
    # Then after processing the request, the response is sent in the opposite order:
    # [any other middlewares ...] -> Session -> HandleExceptions -> CORS -> Health

    # This middleware allows to have consistant http/https protocol in stac links
    app.add_middleware(ProxyHeaderMiddleware)

    # Middleware for implementing first and last buttons in STAC Browser
    app.add_middleware(PaginationLinksMiddleware)

    # Middleware used to update links with title
    app.add_middleware(StacLinksTitleMiddleware, title="My STAC Title")

    # Catch all exceptions and return a JSONResponse
    app.add_middleware(HandleExceptionsMiddleware)
    HandleExceptionsMiddleware.disable_default_exception_handler(app)

    # Add CORS requests from the STAC browser
    if settings.CORS_ORIGINS:
        app.add_middleware(
            CORSMiddleware,
            allow_origins=settings.CORS_ORIGINS,
            allow_methods=["*"],
            allow_headers=["*"],
            allow_credentials=True,
        )

    # More responsive /health and /ping endpoints
    app.add_middleware(HealthMiddleware)

    logger.debug(f"Middlewares: {app.user_middleware}")

    # Finally, apply stac-fastapi openapi patch to comply with the STAC API spec
    return update_openapi(app)