main.py

"""titiler-openeo app."""

import logging

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from openeo_pg_parser_networkx.process_registry import (
    Process,  # type: ignore[import-untyped]
)
from starlette.middleware.cors import CORSMiddleware
from starlette_cramjam.middleware import CompressionMiddleware

from . import __version__ as titiler_version
from .auth import get_auth
from .errors import ExceptionHandler, OpenEOException
from .factory import EndpointsFactory
from .health import register_health_endpoints
from .middleware import DynamicCacheControlMiddleware
from .processes import PROCESS_SPECIFICATIONS, process_registry
from .services import get_store, get_tile_store, get_udp_store
from .settings import ApiSettings, AuthSettings, BackendSettings
from .stacapi import LoadCollection, LoadStac, stacApiBackend

STAC_VERSION = "1.0.0"

api_settings = ApiSettings()
auth_settings = AuthSettings()

# BackendSettings requires stac_api_url and store_url to be set via environment variables
try:
    backend_settings = BackendSettings()  # type: ignore[call-arg]
except Exception as err:
    raise ValueError(
        "Missing required environment variables for BackendSettings. "
        "Please set TITILER_OPENEO_STAC_API_URL and TITILER_OPENEO_STORE_URL"
    ) from err

stac_client = stacApiBackend(
    str(backend_settings.stac_api_url),
    exclude_collections=backend_settings.exclude_collections,
)  # type: ignore
service_store = get_store(str(backend_settings.store_url))
udp_store = get_udp_store(str(backend_settings.store_url))
tile_store = (
    get_tile_store(backend_settings.tile_store_url)
    if backend_settings.tile_store_url
    else None
)
auth = get_auth(auth_settings, store=service_store)


def create_app():
    app = FastAPI(
        title=api_settings.name,
        openapi_url="/api",
        docs_url="/api.html",
        description="""TiTiler backend for openEO.

**Documentation**: <a href="https://sentinel-hub.github.io/titiler-openeo" target="_blank">https://sentinel-hub.github.io/titiler-openeo</a>

**Source Code**: <a href="https://github.com/sentinel-hub/titiler-openeo" target="_blank">https://github.com/sentinel-hub/titiler-openeo</a>

        """,
        version=titiler_version,
        root_path=api_settings.root_path,
        debug=api_settings.debug,
    )

    # Set all CORS enabled origins
    if api_settings.cors_origins:
        app.add_middleware(
            CORSMiddleware,
            allow_origins=api_settings.cors_origins,
            allow_credentials=True,
            allow_methods=api_settings.cors_allow_methods,
            allow_headers=["*"],
            expose_headers=["*"],
        )

    app.add_middleware(
        CompressionMiddleware,
        minimum_size=0,
        exclude_mediatype={
            "image/jpeg",
            "image/jpg",
            "image/png",
            "image/jp2",
            "image/webp",
        },
        compression_level=6,
    )

    app.add_middleware(
        DynamicCacheControlMiddleware,
        static_paths=["/static/"],
        dynamic_paths=[
            "/processes/",
            "/jobs/",
            "/collections/",
            "/services/",
            "/results/",
        ],
    )

    # Register backend specific load_collection methods
    loaders = LoadCollection(stac_client)  # type: ignore
    process_registry["load_collection"] = Process(
        spec=PROCESS_SPECIFICATIONS["load_collection"],
        implementation=loaders.load_collection,
    )
    loaders = LoadStac()  # type: ignore
    process_registry["load_stac"] = Process(
        spec=PROCESS_SPECIFICATIONS["load_stac"],
        implementation=loaders.load_stac,
    )

    # Register OpenEO endpoints
    # Create endpoints with optional tile_store
    factory_args = {
        "services_store": service_store,
        "udp_store": udp_store,
        "stac_client": stac_client,
        "process_registry": process_registry,
        "auth": auth,
        "default_services_file": backend_settings.default_services_file,
    }
    if tile_store:
        factory_args["tile_store"] = tile_store

    endpoints = EndpointsFactory(**factory_args)
    app.include_router(endpoints.router)
    app.endpoints = endpoints

    # Kubernetes-style health endpoints. Registered after the OpenEO router
    # so the explicit /healthz and /readyz paths take precedence and are
    # excluded from the OpenAPI schema.
    register_health_endpoints(
        app,
        service_store=service_store,
        tile_store=tile_store,
        stac_client=stac_client,
        auth=auth,
    )

    # Create exception handler instance
    exception_handler = ExceptionHandler(logger=logging.getLogger(__name__))

    # Add OpenEO-specific exception handlers
    app.add_exception_handler(
        OpenEOException, exception_handler.openeo_exception_handler
    )
    app.add_exception_handler(
        RequestValidationError, exception_handler.validation_exception_handler
    )
    app.add_exception_handler(HTTPException, exception_handler.http_exception_handler)
    app.add_exception_handler(Exception, exception_handler.general_exception_handler)

    return app


app = create_app()