API reference¶

def normalize_config(config: dict) -> dict:
    """Validate the configuration, then fill the keys the runtime assumes.

    Mutates and returns the given dict: callers (including the reload
    path) share one config object, and pygeoapi copies it internally
    anyway. Only absent keys are filled — a tenant's own values stay.

    Validation lives here rather than in the loader because loading a
    document and vouching for it are different jobs: the loader is asked
    for partial documents on purpose. This function is the boundary
    where a document becomes a configuration something is built from,
    and both `build_openapi` and `build_api` come through it, so boot,
    reload and CLI are covered without anyone having to remember.

    It validates the **effective** configuration — after `${VAR}`
    interpolation. The source document is a different thing: `port:
    ${PORT}` is a string where the schema wants an integer, and
    rejecting it would be rejecting a perfectly good deployment.
    """
    server = config.setdefault("server", {})
    # `server.bind` is required by pygeoapi's schema and read by nobody
    # here: upstream uses it in its own runners (`starlette_app.py`,
    # `flask_app.py`, `django_app.py`) to bind a socket it opens itself.
    # fastgeoapi's host and port come from `fastgeoapi run` and uvicorn.
    # Refusing to start over it would be validation harming the people
    # it protects — a working deployment stopped by a key that changes
    # nothing — so it is filled with what is actually bound, taken from
    # the same environment `main` exports before loading.
    bind = server.setdefault("bind", {})
    # ruff: ignore[hardcoded-bind-all-interfaces]
    bind.setdefault("host", os.environ.get("HOST", "0.0.0.0"))  # nosec B104
    bind.setdefault("port", int(os.environ.get("PORT", "5000")))

    _validate(config)
    problem = extension_problem(config)
    if problem is not None:
        where, message = problem
        raise ConfigValidationError(f"invalid fastgeoapi configuration at {where}: {message}")
    limits = server.setdefault("limits", {})
    for key, value in _LIMIT_DEFAULTS.items():
        limits.setdefault(key, value)
    for resource in config.get("resources", {}).values():
        _default_to_the_items(resource.get("providers") or [])
    return config

def build_openapi(config: dict) -> dict:
    """Generate the OpenAPI in memory, with the fastgeoapi fixes at the source.

    The document goes through the openapi-pydantic model once: validated,
    corrected, dumped.
    """
    generated = get_oas(normalize_config(config))
    openapi = OpenAPI.model_validate(generated)
    fix_queryables_response_schema(openapi)
    fix_conformance_and_collections_responses(openapi)
    drop_unfiltered_cql2_operations(openapi, config)
    fix_tileset_list_response(openapi)
    describe_tilesets(openapi)
    fix_map_operations(openapi, config)
    drop_unused_tags(openapi)
    return dump_openapi(openapi, like=generated)

def build_api(config: dict, openapi: dict) -> API:
    """pygeoapi's API instance, with no file nor env var in between.

    Also drops pygeoapi's translated-config cache. ``l10n.translate_struct``
    memoises the config it hands to the HTML templates in a MODULE-level
    dict keyed by locale, so after a reload swap the browser would keep
    seeing config-derived values (the site title, the item-limit options)
    from the previous config while JSON already served the new one. A new
    API instance means a new config, so the cache must go with it.
    """
    from pygeoapi import l10n

    l10n._cfg_cache.clear()
    # Cached plugin instances were built from the previous config: a
    # reload must not keep serving providers that point at data the new
    # config no longer describes.
    invalidate_plugin_cache()
    return API(normalize_config(config), openapi)

async def execute(
    api: API,
    api_function: Callable,
    request: Request,
    *args,
    skip_valid_check: bool = False,
) -> Response:
    """Adaptation of ``execute_from_starlette`` with an explicit API instance."""
    api_request = await APIRequest.from_starlette(request, api.locales)
    if not skip_valid_check and not api_request.is_valid():
        headers, status, content = api.get_format_exception(api_request)
    else:
        loop = asyncio.get_running_loop()
        headers, status, content = await loop.run_in_executor(
            None, _call_threadsafe, loop, api_function, api, api_request, *args
        )
        if status != HTTPStatus.NO_CONTENT:
            content = apply_gzip(headers, content)
    return _to_response(headers, status, content)

def build_routes(
    api: API, specs: frozenset[str] | None = None, http_cache: HttpCache | None = None
) -> list[Route]:
    """The route table, closed over the ``api`` instance.

    Adapted 1:1 from ``pygeoapi/starlette_app.py`` (admin excluded: in
    its place fastgeoapi has the reload webhook). Every entry is tagged
    with its spec group (ADR-0005): ``specs=None`` returns the COMPLETE
    table — the coverage contract the parity test checks against
    upstream — while the sub-app mounts only the active groups.
    ``http_cache`` gives the map and tile routes their cache headers;
    None sends none.
    """

    async def landing_page(request: Request) -> Response:
        return await execute(api, core_api.landing_page, request)

    async def openapi_(request: Request) -> Response:
        return await execute(api, core_api.openapi_, request)

    async def asyncapi_(request: Request) -> Response:
        return await execute(api, core_api.asyncapi_, request)

    async def conformance(request: Request) -> Response:
        # fastgeoapi patch: filter the classes by the configured providers.
        from app.pygeoapi.api.conformance import conformance as conformance_fn

        return await execute(api, conformance_fn, request)

    async def tilematrixset(request: Request) -> Response:
        return await execute(
            api,
            tiles_api.tilematrixset,
            request,
            _path_param(request, "tileMatrixSetId"),
        )

    async def tilematrixsets(request: Request) -> Response:
        return await execute(api, tiles_api.tilematrixsets, request)

    async def collection_schema(request: Request) -> Response:
        return await execute(
            api,
            core_api.get_collection_schema,
            request,
            _path_param(request, "collection_id"),
        )

    async def collection_queryables(request: Request) -> Response:
        return await execute(
            api,
            itemtypes_api.get_collection_queryables,
            request,
            _path_param(request, "collection_id"),
        )

    async def collection_tiles(request: Request) -> Response:
        return await execute(
            api,
            tiles_api.get_collection_tiles,
            request,
            _path_param(request, "collection_id"),
        )

    async def collection_tiles_metadata(request: Request) -> Response:
        return await execute(
            api,
            _collection_tiles_metadata,
            request,
            _path_param(request, "collection_id"),
            _path_param(request, "tileMatrixSetId"),
            skip_valid_check=True,
        )

    # ADR-0010: which collections have a natively async tile provider.
    # Probing instantiates the provider, so it runs off the loop once per
    # configured collection; the dict lives with this table, which the
    # sub-app rebuilds on every reload, and unknown names are not kept.
    natives: dict[str, tuple | None] = {}

    async def collection_items_tiles(request: Request) -> Response:
        args = (
            _path_param(request, "collection_id"),
            _path_param(request, "tileMatrixSetId"),
            _path_param(request, "tile_matrix"),
            _path_param(request, "tileRow"),
            _path_param(request, "tileCol"),
        )
        dataset = args[0]
        if dataset in natives:
            probe = natives[dataset]
        else:
            probe = await asyncio.to_thread(async_tiles.native_tile_provider, api, dataset)
            if dataset in api.config["resources"]:
                natives[dataset] = probe
        if probe is not None:
            # A native provider: awaited on the loop, no thread involved.
            api_request = await APIRequest.from_starlette(request, api.locales)
            headers, status, content = await async_tiles.get_collection_tiles_data(
                api, api_request, *args, *probe, http_cache=http_cache
            )
            if status not in _BODILESS:
                content = _gzip_without_timestamp(headers, content)
            return _to_response(headers, status, content)
        return await execute(
            api,
            tiles_api.get_collection_tiles_data,
            request,
            *args,
            skip_valid_check=True,
        )

    async def collection_items(request: Request) -> Response:
        collection_id = _path_param(request, "collection_id")
        item_id = _path_param(request, "item_id")
        if item_id is None:
            if request.method == "POST":
                if request.headers.get("content-type") == "application/geo+json":
                    return await execute(
                        api,
                        itemtypes_api.manage_collection_item,
                        request,
                        "create",
                        collection_id,
                        skip_valid_check=True,
                    )
                return await execute(
                    api,
                    itemtypes_api.get_collection_items,
                    request,
                    collection_id,
                    skip_valid_check=True,
                )
            if request.method == "OPTIONS":
                return await execute(
                    api,
                    itemtypes_api.manage_collection_item,
                    request,
                    "options",
                    collection_id,
                    skip_valid_check=True,
                )
            return await execute(
                api,
                itemtypes_api.get_collection_items,
                request,
                collection_id,
                skip_valid_check=True,
            )
        if request.method in ("DELETE", "PUT", "OPTIONS"):
            action = {
                "DELETE": "delete",
                "PUT": "update",
                "OPTIONS": "options",
            }[request.method]
            return await execute(
                api,
                itemtypes_api.manage_collection_item,
                request,
                action,
                collection_id,
                item_id,
                skip_valid_check=True,
            )
        return await execute(
            api,
            itemtypes_api.get_collection_item,
            request,
            collection_id,
            item_id,
        )

    async def collection_coverage(request: Request) -> Response:
        return await execute(
            api,
            coverages_api.get_collection_coverage,
            request,
            _path_param(request, "collection_id"),
            skip_valid_check=True,
        )

    # Which collections have a natively async map provider, probed once
    # per collection like the tile providers above.
    map_natives: dict[str, tuple | None] = {}

    async def collection_map(request: Request) -> Response:
        dataset = _path_param(request, "collection_id")
        style = _path_param(request, "style_id")
        if dataset in map_natives:
            probe = map_natives[dataset]
        else:
            probe = await asyncio.to_thread(async_maps.native_map_provider, api, dataset)
            if dataset in api.config["resources"]:
                map_natives[dataset] = probe
        if probe is None:
            return await execute(api, maps_api.get_collection_map, request, dataset, style)
        api_request = await APIRequest.from_starlette(request, api.locales)
        if not api_request.is_valid():
            headers, status, content = api.get_format_exception(api_request)
        else:
            drawn = await _unless_disconnected(
                request,
                async_maps.get_collection_map(
                    api, api_request, dataset, style, *probe, http_cache=http_cache
                ),
            )
            if drawn is None:
                return Response(status_code=CLIENT_CLOSED_REQUEST)
            headers, status, content = drawn
        if status not in _BODILESS:
            content = _gzip_without_timestamp(headers, content)
        return _to_response(headers, status, content)

    async def get_processes(request: Request) -> Response:
        return await execute(
            api,
            processes_api.describe_processes,
            request,
            _path_param(request, "process_id"),
        )

    async def get_jobs(request: Request) -> Response:
        job_id = _path_param(request, "job_id")
        if job_id is None:
            return await execute(api, processes_api.get_jobs, request)
        if request.method == "DELETE":
            return await execute(api, processes_api.delete_job, request, job_id)
        return await execute(api, processes_api.get_jobs, request, job_id)

    async def execute_process_jobs(request: Request) -> Response:
        return await execute(
            api,
            processes_api.execute_process,
            request,
            _path_param(request, "process_id"),
        )

    async def get_job_result(request: Request) -> Response:
        # fastgeoapi patch: job response corrected downstream.
        from app.pygeoapi.api.processes import patch_response

        response = await execute(
            api,
            processes_api.get_job_result,
            request,
            _path_param(request, "job_id"),
        )
        return patch_response(response=response)

    async def edr_query(request: Request) -> Response:
        collection_id = _path_param(request, "collection_id")
        instance_id = _path_param(request, "instance_id")
        if collection_id and "/instances/" in collection_id:
            collection_id, _, instance_id = collection_id.partition("/instances/")
        if request.url.path.endswith("instances") or (
            instance_id is not None and request.url.path.endswith(instance_id)
        ):
            return await execute(
                api,
                edr_api.get_collection_edr_instances,
                request,
                collection_id,
                instance_id,
            )
        location_id = _path_param(request, "location_id")
        query_type = "locations" if location_id is not None else request["path"].split("/")[-1]
        return await execute(
            api,
            edr_api.get_collection_edr_query,
            request,
            collection_id,
            instance_id,
            query_type,
            location_id,
            skip_valid_check=True,
        )

    async def collections(request: Request) -> Response:
        # fastgeoapi patch: the items described from the provider that serves them.
        return await execute(
            api,
            describe_collections,
            request,
            _path_param(request, "collection_id"),
        )

    async def stac_catalog_root(request: Request) -> Response:
        return await execute(api, stac_api.get_stac_root, request)

    async def stac_catalog_path(request: Request) -> Response:
        return await execute(api, stac_api.get_stac_path, request, _path_param(request, "path"))

    async def stac_landing_page(request: Request) -> Response:
        return await execute(api, stac_api.landing_page, request)

    async def stac_search(request: Request) -> Response:
        return await execute(api, stac_api.search, request)

    table: list[tuple[str, Route]] = [
        ("core", Route("/", landing_page)),
        ("core", Route("/openapi", openapi_)),
        ("core", Route("/asyncapi", asyncapi_)),
        ("core", Route("/conformance", conformance)),
        ("tiles", Route("/TileMatrixSets/{tileMatrixSetId}", tilematrixset)),
        ("tiles", Route("/TileMatrixSets", tilematrixsets)),
        (
            "core",
            Route("/collections/{collection_id:path}/schema", collection_schema),
        ),
        (
            "core",
            Route(
                "/collections/{collection_id:path}/queryables",
                collection_queryables,
            ),
        ),
        (
            "tiles",
            Route("/collections/{collection_id:path}/tiles", collection_tiles),
        ),
        (
            "tiles",
            Route(
                "/collections/{collection_id:path}/tiles/{tileMatrixSetId}",
                collection_tiles_metadata,
            ),
        ),
        (
            "tiles",
            Route(
                "/collections/{collection_id:path}/tiles/{tileMatrixSetId}/metadata",
                collection_tiles_metadata,
            ),
        ),
        (
            "tiles",
            Route(
                "/collections/{collection_id:path}/tiles/{tileMatrixSetId}/{tile_matrix}/{tileRow}/{tileCol}",
                collection_items_tiles,
            ),
        ),
        (
            "features",
            Route(
                "/collections/{collection_id:path}/items",
                collection_items,
                methods=["GET", "POST", "OPTIONS"],
            ),
        ),
        (
            "features",
            Route(
                "/collections/{collection_id:path}/items/{item_id:path}",
                collection_items,
                methods=["GET", "PUT", "DELETE", "OPTIONS"],
            ),
        ),
        (
            "coverages",
            Route(
                "/collections/{collection_id:path}/coverage",
                collection_coverage,
            ),
        ),
        # The styled path comes first: `{collection_id:path}` is greedy, and
        # after the plain `/map` it would swallow `roads/styles/night` as a
        # collection id. pygeoapi lists them the other way round
        # (`starlette_app.py:733-734`, 0.24), so in pygeoapi no styled map
        # is ever reached.
        (
            "maps",
            Route(
                "/collections/{collection_id:path}/styles/{style_id:path}/map",
                collection_map,
            ),
        ),
        (
            "maps",
            Route("/collections/{collection_id:path}/map", collection_map),
        ),
        ("processes", Route("/processes", get_processes)),
        ("processes", Route("/processes/{process_id}", get_processes)),
        ("processes", Route("/jobs", get_jobs)),
        (
            "processes",
            Route("/jobs/{job_id}", get_jobs, methods=["GET", "DELETE"]),
        ),
        (
            "processes",
            Route(
                "/processes/{process_id}/execution",
                execute_process_jobs,
                methods=["POST"],
            ),
        ),
        ("processes", Route("/jobs/{job_id}/results", get_job_result)),
        ("edr", Route("/collections/{collection_id:path}/position", edr_query)),
        ("edr", Route("/collections/{collection_id:path}/area", edr_query)),
        ("edr", Route("/collections/{collection_id:path}/cube", edr_query)),
        ("edr", Route("/collections/{collection_id:path}/radius", edr_query)),
        (
            "edr",
            Route("/collections/{collection_id:path}/trajectory", edr_query),
        ),
        ("edr", Route("/collections/{collection_id:path}/corridor", edr_query)),
        (
            "edr",
            Route("/collections/{collection_id:path}/locations", edr_query),
        ),
        (
            "edr",
            Route(
                "/collections/{collection_id:path}/locations/{location_id}",
                edr_query,
            ),
        ),
        (
            "edr",
            Route("/collections/{collection_id:path}/instances", edr_query),
        ),
        (
            "edr",
            Route(
                "/collections/{collection_id:path}/instances/{instance_id}",
                edr_query,
            ),
        ),
        (
            "edr",
            Route(
                "/collections/{collection_id:path}/instances/{instance_id}/position",
                edr_query,
            ),
        ),
        (
            "edr",
            Route(
                "/collections/{collection_id:path}/instances/{instance_id}/area",
                edr_query,
            ),
        ),
        (
            "edr",
            Route(
                "/collections/{collection_id:path}/instances/{instance_id}/cube",
                edr_query,
            ),
        ),
        (
            "edr",
            Route(
                "/collections/{collection_id:path}/instances/{instance_id}/radius",
                edr_query,
            ),
        ),
        (
            "edr",
            Route(
                "/collections/{collection_id:path}/instances/{instance_id}/trajectory",
                edr_query,
            ),
        ),
        (
            "edr",
            Route(
                "/collections/{collection_id:path}/instances/{instance_id}/corridor",
                edr_query,
            ),
        ),
        (
            "edr",
            Route(
                "/collections/{collection_id:path}/instances/{instance_id}/locations",
                edr_query,
            ),
        ),
        (
            "edr",
            Route(
                "/collections/{collection_id:path}/instances/{instance_id}/locations/{location_id}",
                edr_query,
            ),
        ),
        ("core", Route("/collections", collections)),
        ("core", Route("/collections/{collection_id:path}", collections)),
        ("stac", Route("/stac", stac_catalog_root)),
        ("stac", Route("/stac/{path:path}", stac_catalog_path)),
        ("stac", Route("/stac-api", stac_landing_page)),
        (
            "stac",
            Route("/stac-api/search", stac_search, methods=["GET", "POST"]),
        ),
    ]
    if specs is None:
        return [route for _, route in table]
    return [route for spec, route in table if spec in specs]

def build_pygeoapi_subapp(
    config: dict,
    openapi: dict,
    *,
    http_cache: HttpCache | None = None,
    pages: NativePages | None = None,
) -> Starlette:
    """Complete Starlette sub-app, same shape as the former APP import.

    Mounts only the spec groups the config activates (ADR-0005): the
    reload webhook rebuilds the sub-app, so the mounted set follows
    every config update. ``http_cache`` defaults to the policy of
    fastgeoapi's settings. ``pages`` serves fastgeoapi's own HTML pages
    and their compiled assets under ``/_html``; None takes the choice of
    the settings, which keep pygeoapi's pages by default.
    """
    from app.html.activation import default_pages
    from app.pygeoapi.registry import active_specs

    api = build_api(config, openapi)
    static_dir = Path(pygeoapi.__file__).parent / "static"
    try:
        static_dir = Path(config["server"]["templates"]["static"])
    except KeyError:
        pass
    url_prefix = get_api_rules(config).get_url_prefix("starlette")
    routes = build_routes(
        api,
        specs=active_specs(config),
        http_cache=http_cache if http_cache is not None else default_http_cache(),
    )
    mounts = [Mount("/static", StaticFiles(directory=static_dir))]
    if pages is None:
        pages = default_pages()
    if pages is not None:
        routes = pages.wrap(api, routes)
        if pages.static.is_dir():
            mounts.append(Mount("/_html", app=ImmutableFiles(directory=pages.static)))
    return Starlette(routes=[*mounts, Mount(url_prefix or "/", routes=routes)])

def split_source(source: str) -> tuple[str, str]:
    """Split an object reference into (base, relative key)."""
    if source.startswith(_URL_SCHEMES):
        base, _, key = source.rpartition("/")
        return f"{base}/", key
    path = Path(source)
    if not path.is_absolute():
        path = Path.cwd() / path
    return str(path.parent), path.name

def is_remote(source: str) -> bool:
    """Whether ``source`` names an object across a network: a bucket key or an http(s) URL."""
    return source.startswith(_URL_SCHEMES) and not source.startswith("file://")

def obstore_options(store_options: dict | None, base: str = "") -> dict:
    """``store_options`` spelled for obstore, as :func:`load_store` builds a store with them."""
    return _for_obstore(dict(store_options or {}), base)

def load_store(base: str, store_options: dict | None = None) -> ObjectStore:
    """Build the backend for a base URL/directory.

    ``store_options`` are the cloud store settings — ``region``,
    ``skip_signature`` for public data, ``endpoint`` for an
    S3-compatible service (and an explicit one wins over the process
    environment). They are meaningless for a local path and ignored
    there.
    """
    if base.startswith(_URL_SCHEMES):
        from obstore.store import from_url

        if base.startswith(("http://", "https://")):
            # The HTTP store adds its own slash before the key: a base ending in
            # one asks for `prefix//key`, which some servers answer with 404.
            base = base.rstrip("/")
        if store_options:
            config = _for_obstore(store_options, base)
            with _explicit_endpoint_wins(config):
                # ty: `from_url` is overloaded per provider-specific config
                # type, and ours is a plain mapping read from the tenant's
                # configuration — the value is only known at runtime.
                return ObstoreStore(
                    from_url(base, config=config)  # ty: ignore[no-matching-overload]
                )
        return ObstoreStore(from_url(base))
    from obstore.store import LocalStore

    return ObstoreStore(LocalStore(base))