app

stac_fastapi.api.app

Fastapi app creation.

StacApi

StacApi factory.

Factory for creating a STAC-compliant FastAPI application. After instantation, the application is accessible from the StacApi.app attribute.

Attributes:

• settings (ApiSettings) –

  API settings and configuration, potentially using environment variables. See pydantic-docs.helpmanual.io/usage/settings/.

• client (Union[AsyncBaseCoreClient, BaseCoreClient]) –

  A subclass of stac_api.clients.BaseCoreClient. Defines the application logic which is injected into the API.

• extensions (List[ApiExtension]) –

  API extensions to include with the application. This may include official STAC extensions as well as third-party add ons.

• exceptions (Dict[Type[Exception], int]) –

  Defines a global mapping between exceptions and status codes, allowing configuration of response behavior on certain exceptions (fastapi.tiangolo.com/tutorial/handling-errors/#install-custom-exception-handlers).

• app (FastAPI) –

  The FastAPI application, defaults to a fresh application.

• route_dependencies (List[Tuple[List[Scope], List[Depends]]]) –

  List of tuples of route scope dicts (eg {'path': '/collections', 'method': 'POST'}) and list of dependencies (e.g. [Depends(oauth2_scheme)])). Applies specified dependencies to specified routes. This is useful for applying custom auth requirements to routes defined elsewhere in the application.

• health_check (Union[Callable[[], Dict], Callable[[], Awaitable[Dict]]]) –

  A Callable which return application's health information. Defaults to def health: return {"status": "UP"}

Source code in stac_fastapi/api/app.py

@attr.s
class StacApi:
    """StacApi factory.

    Factory for creating a STAC-compliant FastAPI application.  After
    instantation, the application is accessible from the `StacApi.app` attribute.

    Attributes:
        settings:
            API settings and configuration, potentially using environment
            variables. See https://pydantic-docs.helpmanual.io/usage/settings/.
        client:
            A subclass of `stac_api.clients.BaseCoreClient`.  Defines the
            application logic which is injected into the API.
        extensions:
            API extensions to include with the application.  This may include
            official STAC extensions as well as third-party add ons.
        exceptions:
            Defines a global mapping between exceptions and status codes,
            allowing configuration of response behavior on certain exceptions
            (https://fastapi.tiangolo.com/tutorial/handling-errors/#install-custom-exception-handlers).
        app:
            The FastAPI application, defaults to a fresh application.
        route_dependencies:
            List of tuples of route scope dicts (eg `{'path':
            '/collections', 'method': 'POST'}`) and list of dependencies (e.g.
            `[Depends(oauth2_scheme)]`)).  Applies specified dependencies to
            specified routes. This is useful
            for applying custom auth requirements to routes defined elsewhere in
            the application.
        health_check:
            A Callable which return application's `health` information.
            Defaults to `def health: return {"status": "UP"}`

    """

    settings: ApiSettings = attr.ib()
    client: Union[AsyncBaseCoreClient, BaseCoreClient] = attr.ib()
    extensions: List[ApiExtension] = attr.ib(default=attr.Factory(list))
    exceptions: Dict[Type[Exception], int] = attr.ib(
        default=attr.Factory(lambda: DEFAULT_STATUS_CODES)
    )
    title: str = attr.ib(
        default=attr.Factory(
            lambda self: self.settings.stac_fastapi_title, takes_self=True
        )
    )
    api_version: str = attr.ib(
        default=attr.Factory(
            lambda self: self.settings.stac_fastapi_version, takes_self=True
        )
    )
    stac_version: str = attr.ib(default=STAC_VERSION)
    description: str = attr.ib(
        default=attr.Factory(
            lambda self: self.settings.stac_fastapi_description, takes_self=True
        )
    )
    app: FastAPI = attr.ib(
        default=attr.Factory(
            lambda self: FastAPI(
                openapi_url=self.settings.openapi_url,
                docs_url=self.settings.docs_url,
                redoc_url=None,
                root_path=self.settings.root_path,
                title=self.title,
                version=self.api_version,
                description=self.description,
            ),
            takes_self=True,
        ),
        converter=update_openapi,
    )
    router: APIRouter = attr.ib(default=attr.Factory(APIRouter))
    search_get_request_model: Type[BaseSearchGetRequest] = attr.ib(
        default=BaseSearchGetRequest
    )
    search_post_request_model: Type[BaseSearchPostRequest] = attr.ib(
        default=BaseSearchPostRequest
    )
    collections_get_request_model: Type[APIRequest] = attr.ib(default=EmptyRequest)
    collection_get_request_model: Type[APIRequest] = attr.ib(default=CollectionUri)
    items_get_request_model: Type[APIRequest] = attr.ib(default=ItemCollectionUri)
    item_get_request_model: Type[APIRequest] = attr.ib(default=ItemUri)
    response_class: Type[Response] = attr.ib(default=JSONResponse)
    middlewares: List[Middleware] = attr.ib(
        default=attr.Factory(
            lambda: [
                Middleware(BrotliMiddleware),
                Middleware(CORSMiddleware),
                Middleware(ProxyHeaderMiddleware),
            ]
        )
    )
    route_dependencies: List[Tuple[List[Scope], List[Depends]]] = attr.ib(default=[])
    health_check: Union[Callable[[], Dict], Callable[[], Awaitable[Dict]]] = attr.ib(
        default=lambda: {"status": "UP"}
    )

    def get_extension(self, extension: Type[ApiExtension]) -> Optional[ApiExtension]:
        """Get an extension.

        Args:
            extension: extension to check for.

        Returns:
            The extension instance, if it exists.
        """
        for ext in self.extensions:
            if isinstance(ext, extension):
                return ext
        return None

    def register_landing_page(self) -> None:
        """Register landing page (GET /)."""
        self.router.add_api_route(
            name="Landing Page",
            path="/",
            response_model=(
                api.LandingPage if self.settings.enable_response_models else None
            ),
            responses={
                200: {
                    "content": {
                        MimeTypes.json.value: {},
                    },
                    "model": api.LandingPage,
                },
            },
            response_class=self.response_class,
            response_model_exclude_unset=False,
            response_model_exclude_none=True,
            methods=["GET"],
            endpoint=create_async_endpoint(self.client.landing_page, EmptyRequest),
        )

    def register_conformance_classes(self) -> None:
        """Register conformance classes (GET /conformance)."""
        self.router.add_api_route(
            name="Conformance Classes",
            path="/conformance",
            response_model=(
                api.Conformance if self.settings.enable_response_models else None
            ),
            responses={
                200: {
                    "content": {
                        MimeTypes.json.value: {},
                    },
                    "model": api.Conformance,
                },
            },
            response_class=self.response_class,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["GET"],
            endpoint=create_async_endpoint(self.client.conformance, EmptyRequest),
        )

    def register_get_item(self) -> None:
        """Register get item endpoint (GET /collections/{collection_id}/items/{item_id})."""  # noqa: E501
        self.router.add_api_route(
            name="Get Item",
            path="/collections/{collection_id}/items/{item_id}",
            response_model=api.Item if self.settings.enable_response_models else None,
            responses={
                200: {
                    "content": {
                        MimeTypes.geojson.value: {},
                    },
                    "model": api.Item,
                },
            },
            response_class=GeoJSONResponse,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["GET"],
            endpoint=create_async_endpoint(
                self.client.get_item, self.item_get_request_model
            ),
        )

    def register_post_search(self) -> None:
        """Register search endpoint (POST /search)."""
        self.router.add_api_route(
            name="Search",
            path="/search",
            response_model=api.ItemCollection
            if self.settings.enable_response_models
            else None,
            responses={
                200: {
                    "content": {
                        MimeTypes.geojson.value: {},
                    },
                    "model": api.ItemCollection,
                },
            },
            response_class=GeoJSONResponse,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["POST"],
            endpoint=create_async_endpoint(
                self.client.post_search, self.search_post_request_model
            ),
        )

    def register_get_search(self) -> None:
        """Register search endpoint (GET /search)."""
        self.router.add_api_route(
            name="Search",
            path="/search",
            response_model=api.ItemCollection
            if self.settings.enable_response_models
            else None,
            responses={
                200: {
                    "content": {
                        MimeTypes.geojson.value: {},
                    },
                    "model": api.ItemCollection,
                },
            },
            response_class=GeoJSONResponse,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["GET"],
            endpoint=create_async_endpoint(
                self.client.get_search, self.search_get_request_model
            ),
        )

    def register_get_collections(self) -> None:
        """Register get collections endpoint (GET /collections)."""
        self.router.add_api_route(
            name="Get Collections",
            path="/collections",
            response_model=(
                Collections if self.settings.enable_response_models else None
            ),
            responses={
                200: {
                    "content": {
                        MimeTypes.json.value: {},
                    },
                    "model": Collections,
                },
            },
            response_class=self.response_class,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["GET"],
            endpoint=create_async_endpoint(
                self.client.all_collections, self.collections_get_request_model
            ),
        )

    def register_get_collection(self) -> None:
        """Register get collection endpoint (GET /collection/{collection_id})."""
        self.router.add_api_route(
            name="Get Collection",
            path="/collections/{collection_id}",
            response_model=api.Collection
            if self.settings.enable_response_models
            else None,
            responses={
                200: {
                    "content": {
                        MimeTypes.json.value: {},
                    },
                    "model": api.Collection,
                },
            },
            response_class=self.response_class,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["GET"],
            endpoint=create_async_endpoint(
                self.client.get_collection, self.collection_get_request_model
            ),
        )

    def register_get_item_collection(self) -> None:
        """Register get item collection endpoint (GET /collection/{collection_id}/items)."""  # noqa: E501
        self.router.add_api_route(
            name="Get ItemCollection",
            path="/collections/{collection_id}/items",
            response_model=(
                api.ItemCollection if self.settings.enable_response_models else None
            ),
            responses={
                200: {
                    "content": {
                        MimeTypes.geojson.value: {},
                    },
                    "model": api.ItemCollection,
                },
            },
            response_class=GeoJSONResponse,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["GET"],
            endpoint=create_async_endpoint(
                self.client.item_collection, self.items_get_request_model
            ),
        )

    def register_core(self) -> None:
        """Register core STAC endpoints.

            GET /
            GET /conformance
            GET /collections
            GET /collections/{collection_id}
            GET /collections/{collection_id}/items
            GET /collection/{collection_id}/items/{item_id}
            GET /search
            POST /search

        Injects application logic (StacApi.client) into the API layer.

        """
        self.register_landing_page()
        self.register_conformance_classes()
        self.register_get_item()
        self.register_post_search()
        self.register_get_search()
        self.register_get_collections()
        self.register_get_collection()
        self.register_get_item_collection()

    def add_health_check(self) -> None:
        """Add a health check."""

        mgmt_router = APIRouter(prefix=self.app.state.router_prefix)

        async def ping():
            """Liveliness probe."""
            return {"message": "PONG"}

        mgmt_router.add_api_route(
            name="Ping",
            path="/_mgmt/ping",
            response_model=Dict,
            responses={
                200: {
                    "content": {
                        MimeTypes.json.value: {},
                    },
                },
            },
            response_class=self.response_class,
            methods=["GET"],
            endpoint=ping,
        )

        mgmt_router.add_api_route(
            name="Health",
            path="/_mgmt/health",
            response_model=Dict,
            responses={
                200: {
                    "content": {
                        MimeTypes.json.value: {},
                    },
                },
            },
            response_class=self.response_class,
            methods=["GET"],
            endpoint=self.health_check,
        )
        self.app.include_router(mgmt_router, tags=["Liveliness/Readiness"])

    def add_route_dependencies(
        self, scopes: List[Scope], dependencies: List[Depends]
    ) -> None:
        """Add custom dependencies to routes.

        Args:
            scopes: list of Scope.
                Each scope should be a dict with a `path` and `method` property.
            dependencies: list of fastapi.Depends
                [FastAPI dependencies](https://fastapi.tiangolo.com/tutorial/dependencies/)
                to apply to each scope.

        """
        return add_route_dependencies(self.app.router.routes, scopes, dependencies)

    def __attrs_post_init__(self) -> None:
        """Post-init hook.

        Responsible for setting up the application upon instantiation of the class.

        """
        # inject settings
        self.client.extensions = self.extensions
        self.client.stac_version = self.stac_version
        self.client.title = self.title
        self.client.description = self.description

        Settings.set(self.settings)
        self.app.state.settings = self.settings

        # Register core STAC endpoints
        self.register_core()
        self.app.include_router(self.router)

        # keep link to the router prefix value
        router_prefix = self.router.prefix
        self.app.state.router_prefix = router_prefix if router_prefix else ""

        # register extensions
        for ext in self.extensions:
            ext.register(self.app)

        # add health check
        self.add_health_check()

        # register exception handlers
        add_exception_handlers(self.app, status_codes=self.exceptions)

        # add middlewares
        if self.middlewares and self.app.middleware_stack is not None:
            raise RuntimeError("Cannot add middleware after an application has started")

        for middleware in self.middlewares:
            self.app.user_middleware.insert(0, middleware)

        # customize route dependencies
        for scopes, dependencies in self.route_dependencies:
            self.add_route_dependencies(scopes=scopes, dependencies=dependencies)

        if self.app.state.settings.enable_direct_response:
            add_direct_response(self.app)

__attrs_post_init__

__attrs_post_init__() -> None

Post-init hook.

Responsible for setting up the application upon instantiation of the class.

Source code in stac_fastapi/api/app.py

def __attrs_post_init__(self) -> None:
    """Post-init hook.

    Responsible for setting up the application upon instantiation of the class.

    """
    # inject settings
    self.client.extensions = self.extensions
    self.client.stac_version = self.stac_version
    self.client.title = self.title
    self.client.description = self.description

    Settings.set(self.settings)
    self.app.state.settings = self.settings

    # Register core STAC endpoints
    self.register_core()
    self.app.include_router(self.router)

    # keep link to the router prefix value
    router_prefix = self.router.prefix
    self.app.state.router_prefix = router_prefix if router_prefix else ""

    # register extensions
    for ext in self.extensions:
        ext.register(self.app)

    # add health check
    self.add_health_check()

    # register exception handlers
    add_exception_handlers(self.app, status_codes=self.exceptions)

    # add middlewares
    if self.middlewares and self.app.middleware_stack is not None:
        raise RuntimeError("Cannot add middleware after an application has started")

    for middleware in self.middlewares:
        self.app.user_middleware.insert(0, middleware)

    # customize route dependencies
    for scopes, dependencies in self.route_dependencies:
        self.add_route_dependencies(scopes=scopes, dependencies=dependencies)

    if self.app.state.settings.enable_direct_response:
        add_direct_response(self.app)

add_health_check

add_health_check() -> None

Add a health check.

Source code in stac_fastapi/api/app.py

def add_health_check(self) -> None:
    """Add a health check."""

    mgmt_router = APIRouter(prefix=self.app.state.router_prefix)

    async def ping():
        """Liveliness probe."""
        return {"message": "PONG"}

    mgmt_router.add_api_route(
        name="Ping",
        path="/_mgmt/ping",
        response_model=Dict,
        responses={
            200: {
                "content": {
                    MimeTypes.json.value: {},
                },
            },
        },
        response_class=self.response_class,
        methods=["GET"],
        endpoint=ping,
    )

    mgmt_router.add_api_route(
        name="Health",
        path="/_mgmt/health",
        response_model=Dict,
        responses={
            200: {
                "content": {
                    MimeTypes.json.value: {},
                },
            },
        },
        response_class=self.response_class,
        methods=["GET"],
        endpoint=self.health_check,
    )
    self.app.include_router(mgmt_router, tags=["Liveliness/Readiness"])

add_route_dependencies

add_route_dependencies(scopes: List[Scope], dependencies: List[Depends]) -> None

Add custom dependencies to routes.

Parameters:

• scopes (List[Scope]) –

  list of Scope. Each scope should be a dict with a path and method property.

• dependencies (List[Depends]) –

  list of fastapi.Depends FastAPI dependencies to apply to each scope.

Source code in stac_fastapi/api/app.py

def add_route_dependencies(
    self, scopes: List[Scope], dependencies: List[Depends]
) -> None:
    """Add custom dependencies to routes.

    Args:
        scopes: list of Scope.
            Each scope should be a dict with a `path` and `method` property.
        dependencies: list of fastapi.Depends
            [FastAPI dependencies](https://fastapi.tiangolo.com/tutorial/dependencies/)
            to apply to each scope.

    """
    return add_route_dependencies(self.app.router.routes, scopes, dependencies)

get_extension

get_extension(extension: Type[ApiExtension]) -> Optional[ApiExtension]

Get an extension.

Parameters:

• extension (Type[ApiExtension]) –

  extension to check for.

Returns:

• Optional[ApiExtension] –

  The extension instance, if it exists.

Source code in stac_fastapi/api/app.py

def get_extension(self, extension: Type[ApiExtension]) -> Optional[ApiExtension]:
    """Get an extension.

    Args:
        extension: extension to check for.

    Returns:
        The extension instance, if it exists.
    """
    for ext in self.extensions:
        if isinstance(ext, extension):
            return ext
    return None

register_conformance_classes

register_conformance_classes() -> None

Register conformance classes (GET /conformance).

Source code in stac_fastapi/api/app.py

def register_conformance_classes(self) -> None:
    """Register conformance classes (GET /conformance)."""
    self.router.add_api_route(
        name="Conformance Classes",
        path="/conformance",
        response_model=(
            api.Conformance if self.settings.enable_response_models else None
        ),
        responses={
            200: {
                "content": {
                    MimeTypes.json.value: {},
                },
                "model": api.Conformance,
            },
        },
        response_class=self.response_class,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["GET"],
        endpoint=create_async_endpoint(self.client.conformance, EmptyRequest),
    )

register_core

register_core() -> None

Register core STAC endpoints.

GET /
GET /conformance
GET /collections
GET /collections/{collection_id}
GET /collections/{collection_id}/items
GET /collection/{collection_id}/items/{item_id}
GET /search
POST /search

Injects application logic (StacApi.client) into the API layer.

Source code in stac_fastapi/api/app.py

def register_core(self) -> None:
    """Register core STAC endpoints.

        GET /
        GET /conformance
        GET /collections
        GET /collections/{collection_id}
        GET /collections/{collection_id}/items
        GET /collection/{collection_id}/items/{item_id}
        GET /search
        POST /search

    Injects application logic (StacApi.client) into the API layer.

    """
    self.register_landing_page()
    self.register_conformance_classes()
    self.register_get_item()
    self.register_post_search()
    self.register_get_search()
    self.register_get_collections()
    self.register_get_collection()
    self.register_get_item_collection()

register_get_collection

register_get_collection() -> None

Register get collection endpoint (GET /collection/{collection_id}).

Source code in stac_fastapi/api/app.py

def register_get_collection(self) -> None:
    """Register get collection endpoint (GET /collection/{collection_id})."""
    self.router.add_api_route(
        name="Get Collection",
        path="/collections/{collection_id}",
        response_model=api.Collection
        if self.settings.enable_response_models
        else None,
        responses={
            200: {
                "content": {
                    MimeTypes.json.value: {},
                },
                "model": api.Collection,
            },
        },
        response_class=self.response_class,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["GET"],
        endpoint=create_async_endpoint(
            self.client.get_collection, self.collection_get_request_model
        ),
    )

register_get_collections

register_get_collections() -> None

Register get collections endpoint (GET /collections).

Source code in stac_fastapi/api/app.py

def register_get_collections(self) -> None:
    """Register get collections endpoint (GET /collections)."""
    self.router.add_api_route(
        name="Get Collections",
        path="/collections",
        response_model=(
            Collections if self.settings.enable_response_models else None
        ),
        responses={
            200: {
                "content": {
                    MimeTypes.json.value: {},
                },
                "model": Collections,
            },
        },
        response_class=self.response_class,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["GET"],
        endpoint=create_async_endpoint(
            self.client.all_collections, self.collections_get_request_model
        ),
    )

register_get_item

register_get_item() -> None

Register get item endpoint (GET /collections/{collection_id}/items/{item_id}).

Source code in stac_fastapi/api/app.py

def register_get_item(self) -> None:
    """Register get item endpoint (GET /collections/{collection_id}/items/{item_id})."""  # noqa: E501
    self.router.add_api_route(
        name="Get Item",
        path="/collections/{collection_id}/items/{item_id}",
        response_model=api.Item if self.settings.enable_response_models else None,
        responses={
            200: {
                "content": {
                    MimeTypes.geojson.value: {},
                },
                "model": api.Item,
            },
        },
        response_class=GeoJSONResponse,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["GET"],
        endpoint=create_async_endpoint(
            self.client.get_item, self.item_get_request_model
        ),
    )

register_get_item_collection

register_get_item_collection() -> None

Register get item collection endpoint (GET /collection/{collection_id}/items).

Source code in stac_fastapi/api/app.py

def register_get_item_collection(self) -> None:
    """Register get item collection endpoint (GET /collection/{collection_id}/items)."""  # noqa: E501
    self.router.add_api_route(
        name="Get ItemCollection",
        path="/collections/{collection_id}/items",
        response_model=(
            api.ItemCollection if self.settings.enable_response_models else None
        ),
        responses={
            200: {
                "content": {
                    MimeTypes.geojson.value: {},
                },
                "model": api.ItemCollection,
            },
        },
        response_class=GeoJSONResponse,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["GET"],
        endpoint=create_async_endpoint(
            self.client.item_collection, self.items_get_request_model
        ),
    )

register_get_search

register_get_search() -> None

Register search endpoint (GET /search).

Source code in stac_fastapi/api/app.py

def register_get_search(self) -> None:
    """Register search endpoint (GET /search)."""
    self.router.add_api_route(
        name="Search",
        path="/search",
        response_model=api.ItemCollection
        if self.settings.enable_response_models
        else None,
        responses={
            200: {
                "content": {
                    MimeTypes.geojson.value: {},
                },
                "model": api.ItemCollection,
            },
        },
        response_class=GeoJSONResponse,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["GET"],
        endpoint=create_async_endpoint(
            self.client.get_search, self.search_get_request_model
        ),
    )

register_landing_page

register_landing_page() -> None

Register landing page (GET /).

Source code in stac_fastapi/api/app.py

def register_landing_page(self) -> None:
    """Register landing page (GET /)."""
    self.router.add_api_route(
        name="Landing Page",
        path="/",
        response_model=(
            api.LandingPage if self.settings.enable_response_models else None
        ),
        responses={
            200: {
                "content": {
                    MimeTypes.json.value: {},
                },
                "model": api.LandingPage,
            },
        },
        response_class=self.response_class,
        response_model_exclude_unset=False,
        response_model_exclude_none=True,
        methods=["GET"],
        endpoint=create_async_endpoint(self.client.landing_page, EmptyRequest),
    )

register_post_search

register_post_search() -> None

Register search endpoint (POST /search).

Source code in stac_fastapi/api/app.py

def register_post_search(self) -> None:
    """Register search endpoint (POST /search)."""
    self.router.add_api_route(
        name="Search",
        path="/search",
        response_model=api.ItemCollection
        if self.settings.enable_response_models
        else None,
        responses={
            200: {
                "content": {
                    MimeTypes.geojson.value: {},
                },
                "model": api.ItemCollection,
            },
        },
        response_class=GeoJSONResponse,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["POST"],
        endpoint=create_async_endpoint(
            self.client.post_search, self.search_post_request_model
        ),
    )