__init__.py

# =================================================================
#
# Authors: Tom Kralidis <tomkralidis@gmail.com>
#          Francesco Bartoli <xbartolone@gmail.com>
#          Sander Schaminee <sander.schaminee@geocat.net>
#          John A Stevenson <jostev@bgs.ac.uk>
#          Colin Blackburn <colb@bgs.ac.uk>
#          Ricardo Garcia Silva <ricardo.garcia.silva@geobeyond.it>
#
# Copyright (c) 2026 Tom Kralidis
# Copyright (c) 2026 Francesco Bartoli
# Copyright (c) 2022 John A Stevenson and Colin Blackburn
# Copyright (c) 2023 Ricardo Garcia Silva
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation
# files (the "Software"), to deal in the Software without
# restriction, including without limitation the rights to use,
# copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the
# Software is furnished to do so, subject to the following
# conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
#
# =================================================================

"""
Root level code of pygeoapi, parsing content provided by web framework.
Returns content from plugins and sets responses.
"""

from collections import ChainMap
from copy import deepcopy
from datetime import datetime
from functools import partial
from gzip import compress
from http import HTTPStatus
import logging
import re
import sys
from typing import Any, Tuple, Union, Self

from babel import Locale
from dateutil.parser import parse as dateparse
import pytz

from pygeoapi import __version__, l10n
from pygeoapi.api.collection import gen_collection, OGC_RELTYPES_BASE
from pygeoapi.formats import FORMAT_TYPES, F_GZIP, F_HTML, F_JSON, F_JSONLD
from pygeoapi.linked_data import jsonldify, jsonldify_collection
from pygeoapi.log import setup_logger
from pygeoapi.plugin import load_plugin
from pygeoapi.process.manager.base import get_manager
from pygeoapi.provider import filter_providers_by_type, get_provider_by_type
from pygeoapi.provider.base import ProviderGenericError, ProviderTypeError

from pygeoapi.util import (
    TEMPLATESDIR, UrlPrefetcher, filter_dict_by_key_value, get_api_rules,
    get_base_url, get_typed_value, render_j2_template, to_json,
    get_choice_from_headers, get_from_headers
)

LOGGER = logging.getLogger(__name__)

#: Return headers for requests (e.g:X-Powered-By)
HEADERS = {
    'Content-Type': 'application/json',
    'X-Powered-By': f'pygeoapi {__version__}'
}

CHARSET = ['utf-8']

#: Locale used for system responses (e.g. exceptions)
SYSTEM_LOCALE = l10n.Locale('en', 'US')

CONFORMANCE_CLASSES = [
    'http://www.opengis.net/spec/ogcapi-common-1/1.0/conf/core',
    'http://www.opengis.net/spec/ogcapi-common-2/1.0/conf/collections',
    'http://www.opengis.net/spec/ogcapi-common-1/1.0/conf/landing-page',
    'http://www.opengis.net/spec/ogcapi-common-1/1.0/conf/json',
    'http://www.opengis.net/spec/ogcapi-common-1/1.0/conf/html',
    'http://www.opengis.net/spec/ogcapi-common-1/1.0/conf/oas30'
]


def all_apis() -> dict:
    """
    Return all supported API modules

    NOTE: this is a function and not a constant to avoid import loops

    :returns: `dict` of API provider type, API module
    """

    from . import (coverages, environmental_data_retrieval, itemtypes, maps,
                   processes, pubsub, tiles, stac)

    return {
        'coverage': coverages,
        'edr': environmental_data_retrieval,
        'itemtypes': itemtypes,
        'map': maps,
        'process': processes,
        'pubsub': pubsub,
        'tile': tiles,
        'stac': stac
    }


def apply_gzip(headers: dict, content: Union[str, bytes]) -> Union[str, bytes]:
    """
    Compress content if requested in header.
    """
    charset = CHARSET[0]

    if F_GZIP in get_from_headers(headers, 'content-encoding'):
        try:
            if isinstance(content, bytes):
                # bytes means Content-Type needs to be set upstream
                content = compress(content)
            else:
                headers['Content-Type'] = \
                    f"{headers['Content-Type']}; charset={charset}"
                content = compress(content.encode(charset))
        except TypeError as err:
            headers.pop('Content-Encoding')
            LOGGER.error(f'Error in compression: {err}')
    return content


class APIRequest:
    """
    Transforms an incoming server-specific Request into an object
    with some generic helper methods and properties.

    This allows writing straightforward API functions supporting all
    web platforms such as with this example:

    .. code-block:: python

       def example_method(api: API, request: APIRequest, custom_arg):
          headers = request.get_response_headers()

          # generate response_body here

          return headers, HTTPStatus.OK, response_body


    Basic request validation is done automatically by web platform specific
    adapters such as ``execute_from_flask`` in flask_app.py . If you want to
    support custom formats (e.g. ``f=xml``), it needs to be registered with
    `skip_valid_check=True` and you can use the following code for custom
    validation. If `xml` was requested, we set the `Content-Type` ourselves.
    For the standard formats, the `APIRequest` object sets the `Content-Type`.

    .. code-block:: python

       def example_method(api: API, request: APIRequest, custom_arg):
          if not request.is_valid(['xml']):
             return api.get_format_exception(request)

          content_type = 'application/xml' if request.format == 'xml' else None
          headers = request.get_response_headers(content_type)

          # generate response_body here

          return headers, HTTPStatus.OK, response_body

    Note that you don't *have* to call :meth:`is_valid`, but that you can also
    perform a custom check on the requested output format by looking at the
    :attr:`format` property.
    Other query parameters are available through the :attr:`params` property as
    a `dict`. The request body is available through the :attr:`data` property.

    :param request:             The web platform specific Request instance.
    :param supported_locales:   List or set of supported Locale instances.
    """

    def __init__(self, request, supported_locales):
        # Set default request data
        self._data = b''

        # Copy request query parameters
        self._args = self._get_params(request)

        # Get path info
        if hasattr(request, 'scope'):
            self._path_info = request.scope['path'].strip('/')
        elif hasattr(request.headers, 'environ'):
            self._path_info = request.headers.environ['PATH_INFO'].strip('/')
        elif hasattr(request, 'path_info'):
            self._path_info = request.path_info

        # Extract locale from params or headers
        self._raw_locale, self._locale = self._get_locale(request.headers,
                                                          supported_locales)

        # Get received headers
        self._headers = self.get_request_headers(request.headers)

        # Determine format
        self._format = self._get_format()

    @classmethod
    def from_flask(cls, request, supported_locales) -> 'APIRequest':
        """Factory class similar to with_data, but only for flask requests"""
        api_req = cls(request, supported_locales)
        api_req._data = request.data
        return api_req

    @classmethod
    async def from_starlette(cls, request, supported_locales) -> 'APIRequest':
        """Factory class similar to with_data, but only for starlette requests
        """
        api_req = cls(request, supported_locales)
        api_req._data = await request.body()
        return api_req

    @classmethod
    def from_django(cls, request, supported_locales) -> 'APIRequest':
        """Factory class similar to with_data, but only for django requests"""
        api_req = cls(request, supported_locales)
        api_req._data = request.body
        return api_req

    @staticmethod
    def _get_params(request):
        """
        Extracts the query parameters from the `Request` object.

        :param request: A Flask or Starlette Request instance
        :returns: `ImmutableMultiDict` or empty `dict`
        """

        if hasattr(request, 'args'):
            # Return ImmutableMultiDict from Flask request
            return request.args
        elif hasattr(request, 'query_params'):
            # Return ImmutableMultiDict from Starlette request
            return request.query_params
        elif hasattr(request, 'GET'):
            # Return QueryDict from Django GET request
            return request.GET
        elif hasattr(request, 'POST'):
            # Return QueryDict from Django GET request
            return request.POST
        LOGGER.debug('No query parameters found')
        return {}

    def _get_locale(self, headers: dict,
                    supported_locales: Union[list, Locale]) -> tuple:
        """
        Detects locale from "lang=<language>" param or `Accept-Language`
        header. Returns a tuple of (raw, locale) if found in params or headers.
        Returns a tuple of (raw default, default locale) if not found.

        :param headers: A dict with Request headers
        :param supported_locales: List or set of supported Locale instances
        :returns: A tuple of (str, Locale)
        """

        raw = None
        try:
            default_locale = l10n.str2locale(supported_locales[0])
        except (TypeError, IndexError, l10n.LocaleError) as err:
            # This should normally not happen, since the API class already
            # loads the supported languages from the config, which raises
            # a LocaleError if any of these languages are invalid.
            LOGGER.error(err)
            raise ValueError(f"{self.__class__.__name__} must be initialized"
                             f"with a list of valid supported locales")

        for field, mapping in ((l10n.QUERY_PARAM, self._args),
                               ('accept-language', headers)):

            loc_strs = get_choice_from_headers(mapping, field, all=True)
            if loc_strs:
                if raw is None:
                    # This is the first-found locale string: set as raw
                    raw = get_from_headers(mapping, field)

                # Check if locale string is a good match for the UI
                loc = l10n.best_match(loc_strs, supported_locales)
                is_override = field is l10n.QUERY_PARAM
                if loc != default_locale or is_override:
                    return raw, loc

        return raw, default_locale

    def _get_format(self, extra_formats: dict = {}) -> Union[str, None]:
        """
        Get `Request` format type from query parameters or headers.

        :param extra_formats: Dict of extra dataset specific formats

        :returns: format value or None if not found/specified
        """

        # Optional f=html or f=json query param
        # Overrides Accept header and might differ from FORMAT_TYPES
        format_ = (self._args.get('f') or '').strip()
        if format_:
            return format_

        # Format not specified: get from Accept headers (MIME types)
        # e.g. Accept: 'text/html;q=0.5,application/ld+json'
        types_ = get_choice_from_headers(self.headers, 'accept', all=True)
        if types_ is None:
            return

        # Add formatters to accepted format types
        extra_formats_mimes = {
            k: v.mimetype for k, v in extra_formats.items()
            if hasattr(v, 'mimetype')
        }
        merged_format_types = FORMAT_TYPES | extra_formats_mimes

        # Lookup formatter by mimetype
        mimes = {
            merged_format_types[k].split(';')[0]: k
            for k in merged_format_types
        }
        for type_ in types_:
            if type_ in mimes:
                return mimes[type_]

    @property
    def data(self) -> bytes:
        """Returns the additional data send with the Request (bytes)"""
        return self._data

    @property
    def params(self) -> dict:
        """Returns the Request query parameters dict"""
        return self._args

    @property
    def path_info(self) -> str:
        """Returns the web server request path info part"""
        return self._path_info

    @property
    def locale(self) -> l10n.Locale:
        """
        Returns the user-defined locale from the request object.
        If no locale has been defined or if it is invalid,
        the default server locale is returned.

        .. note::   The locale here determines the language in which pygeoapi
                    should return its responses. This may not be the language
                    that the user requested. It may also not be the language
                    that is supported by a collection provider, for example.
                    For this reason, you should pass the `raw_locale` property
                    to the :func:`l10n.get_plugin_locale` function, so that
                    the best match for the provider can be determined.

        :returns: babel.core.Locale
        """

        return self._locale

    @property
    def raw_locale(self) -> Union[str, None]:
        """
        Returns the raw locale string from the `Request` object.
        If no "lang" query parameter or `Accept-Language` header was found,
        `None` is returned.
        Pass this value to the :func:`l10n.get_plugin_locale` function to let
        the provider determine a best match for the locale, which may be
        different from the locale used by pygeoapi's UI.

        :returns: a locale string or None
        """

        return self._raw_locale

    @property
    def format(self) -> Union[str, None]:
        """
        Returns the content type format from the
        request query parameters or headers.

        :returns: Format name or None
        """

        return self._format

    @property
    def headers(self) -> dict:
        """
        Returns the dictionary of the headers from
        the request.

        :returns: Request headers dictionary
        """

        return self._headers

    def get_linkrel(self, format_: str) -> str:
        """
        Returns the hyperlink relationship (rel) attribute value for
        the given API format string.

        The string is compared against the request format and if it matches,
        the value 'self' is returned. Otherwise, 'alternate' is returned.
        However, if `format_` is 'json' and *no* request format was found,
        the relationship 'self' is returned as well (JSON is the default).

        :param format_: The format to compare the request format against.
        :returns: A string 'self' or 'alternate'.
        """

        fmt = format_.lower()
        if fmt == self._format or (fmt == F_JSON and not self._format):
            return 'self'
        return 'alternate'

    def is_valid(self, additional_formats=None) -> bool:
        """
        Returns True if:
            - the format is not set (None)
            - the requested format is supported
            - the requested format exists in a list if additional formats

        .. note::   Format names are matched in a case-insensitive manner.

        :param additional_formats: Optional additional supported formats list

        :returns: bool
        """

        if not self._format:
            return True
        if self._format in FORMAT_TYPES.keys():
            return True
        if self._format in (f.lower() for f in (additional_formats or ())):
            return True
        return False

    def get_response_headers(self, force_lang: l10n.Locale | None = None,
                             force_type: str | None = None,
                             force_encoding: str | None = None,
                             **custom_headers) -> dict:
        """
        Prepares and returns a dictionary with Response object headers.

        This method always adds a 'Content-Language' header, where the value
        is determined by the 'lang' query parameter or 'Accept-Language'
        header from the request.
        If no language was requested, the default pygeoapi language is used,
        unless a `force_lang` override was specified (see notes below).

        A 'Content-Type' header is also always added to the response.
        If the user does not specify `force_type`, the header is based on
        the `format` APIRequest property. If that is invalid, the default MIME
        type `application/json` is used.

        ..note::    If a `force_lang` override is applied, that language
                    is always set as the 'Content-Language', regardless of
                    a 'lang' query parameter or 'Accept-Language' header.
                    If an API response always needs to be in the same
                    language, 'force_lang' should be set to that language.

        :param force_lang: An optional Content-Language header override.
        :param force_type: An optional Content-Type header override.
        :param force_encoding: An optional Content-Encoding header override.
        :returns: A header dict
        """

        headers = HEADERS.copy()
        headers.update(**custom_headers)
        l10n.set_response_language(headers, force_lang or self._locale)
        if force_type:
            # Set custom MIME type if specified
            headers['Content-Type'] = force_type
        elif self.is_valid() and self._format:
            # Set MIME type for valid formats
            headers['Content-Type'] = FORMAT_TYPES[self._format]

        if F_GZIP in FORMAT_TYPES:
            if force_encoding:
                headers['Content-Encoding'] = force_encoding
            elif F_GZIP in get_from_headers(self._headers, 'accept-encoding'):  # noqa
                headers['Content-Encoding'] = F_GZIP

        return headers

    def get_request_headers(self, headers: dict) -> dict:
        """
        Obtains and returns a dictionary with Request object headers.

        This method adds the headers of the original request and
        makes them available to the API object.

        :returns: A header dict
        """

        headers_ = {item[0]: item[1] for item in headers.items()}
        return headers_


class API:
    """API object"""

    def __init__(self, config: dict, openapi: dict,
                 asyncapi: dict = {}) -> Self | None:
        """
        constructor

        :param config: configuration dict
        :param openapi: openapi dict
        :param asyncapi: asyncapi dict

        :returns: `pygeoapi.API` instance
        """

        self.config = config
        self.openapi = openapi
        self.asyncapi = asyncapi
        self.api_headers = get_api_rules(self.config).response_headers
        self.base_url = get_base_url(self.config)
        self.prefetcher = UrlPrefetcher()
        self.pubsub_client = None

        CHARSET[0] = config['server'].get('encoding', 'utf-8')
        if config['server'].get('gzip'):
            FORMAT_TYPES[F_GZIP] = 'application/gzip'
            FORMAT_TYPES.move_to_end(F_JSON)

        # Process language settings (first locale is default!)
        self.locales = l10n.get_locales(config)
        self.default_locale = self.locales[0]

        if 'templates' not in self.config['server']:
            self.config['server']['templates'] = {'path': TEMPLATESDIR}

        if 'pretty_print' not in self.config['server']:
            self.config['server']['pretty_print'] = False

        self.pretty_print = self.config['server']['pretty_print']

        setup_logger(self.config['logging'])

        # Create config clone for HTML templating with modified base URL
        self.tpl_config = deepcopy(self.config)
        self.tpl_config['server']['url'] = self.base_url

        self.manager = get_manager(self.config)
        LOGGER.info('Process manager plugin loaded')

        if self.config.get('pubsub') is not None:
            LOGGER.debug('Loading PubSub client')
            self.pubsub_client = load_plugin('pubsub', self.config['pubsub'])

    def get_exception(self, status: int, headers: dict, format_: str | None,
                      code: str, description: str) -> Tuple[dict, int, str]:
        """
        Exception handler

        :param status: HTTP status code
        :param headers: dict of HTTP response headers
        :param format_: format string
        :param code: OGC API exception code
        :param description: OGC API exception code

        :returns: tuple of headers, status, and message
        """

        exception_info = sys.exc_info()

        LOGGER.error(
            description,
            exc_info=exception_info if exception_info[0] is not None else None
        )
        exception = {
            'code': code,
            'type': code,
            'description': description
        }

        if format_ == F_HTML:
            headers['Content-Type'] = FORMAT_TYPES[F_HTML]
            content = render_j2_template(
                self.tpl_config, self.config['server']['templates'],
                'exception.html', exception, SYSTEM_LOCALE)
        else:
            content = to_json(exception, self.pretty_print)

        if status == HTTPStatus.NO_CONTENT:
            LOGGER.error('HTTP 204 detected, suppressing content')
            content = ''

        return headers, status, content

    def get_format_exception(self,
                             request: APIRequest) -> Tuple[dict, int, str]:
        """
        Returns a format exception.

        :param request: An APIRequest instance.

        :returns: tuple of (headers, status, message)
        """

        # Content-Language is in the system locale (ignore language settings)
        headers = request.get_response_headers(SYSTEM_LOCALE,
                                               **self.api_headers)
        msg = 'Invalid format requested'
        LOGGER.error(f'{msg}: {request.format}')
        return self.get_exception(
            HTTPStatus.BAD_REQUEST, headers,
            request.format, 'InvalidParameterValue', msg)

    def get_collections_url(self) -> str:
        return f"{self.base_url}/collections"

    def get_dataset_templates(self, dataset: str) -> dict:
        templates = self.config['resources'][dataset].get('templates')

        return templates or self.tpl_config['server']['templates']


@jsonldify
def landing_page(api: API,
                 request: APIRequest) -> Tuple[dict, int, str]:
    """
    Provide API landing page

    :param request: A request object

    :returns: tuple of headers, status code, content
    """

    fcm = {
        'links': [],
        'title': l10n.translate(
            api.config['metadata']['identification']['title'],
            request.locale),
        'description':
            l10n.translate(
                api.config['metadata']['identification']['description'],
                request.locale)
    }

    LOGGER.debug('Creating links')
    # TODO: put title text in config or translatable files?
    fcm['links'] = [{
        'rel': 'about',
        'type': 'text/html',
        'title': l10n.translate(
            api.config['metadata']['identification']['title'],
            request.locale),
        'href': api.config['metadata']['identification']['url']
    }, {
        'rel': request.get_linkrel(F_JSON),
        'type': FORMAT_TYPES[F_JSON],
        'title': l10n.translate('This document as JSON', request.locale),
        'href': f"{api.base_url}?f={F_JSON}"
    }, {
        'rel': request.get_linkrel(F_JSONLD),
        'type': FORMAT_TYPES[F_JSONLD],
        'title': l10n.translate('This document as RDF (JSON-LD)', request.locale),  # noqa
        'href': f"{api.base_url}?f={F_JSONLD}"
    }, {
        'rel': request.get_linkrel(F_HTML),
        'type': FORMAT_TYPES[F_HTML],
        'title': l10n.translate('This document as HTML', request.locale),
        'href': f"{api.base_url}?f={F_HTML}",
        'hreflang': api.default_locale
    }, {
        'rel': 'service-desc',
        'type': 'application/vnd.oai.openapi+json;version=3.0',
        'title': l10n.translate('The OpenAPI definition as JSON', request.locale),  # noqa
        'href': f"{api.base_url}/openapi"
    }, {
        'rel': 'service-doc',
        'type': FORMAT_TYPES[F_HTML],
        'title': l10n.translate('The OpenAPI definition as HTML', request.locale),  # noqa
        'href': f"{api.base_url}/openapi?f={F_HTML}",
        'hreflang': api.default_locale
    }, {
        'rel': 'conformance',
        'type': FORMAT_TYPES[F_JSON],
        'title': l10n.translate('Conformance', request.locale),
        'href': f"{api.base_url}/conformance"
    }, {
        'rel': 'data',
        'type': FORMAT_TYPES[F_JSON],
        'title': l10n.translate('Collections', request.locale),
        'href': api.get_collections_url()
    }, {
        'rel': f'{OGC_RELTYPES_BASE}/processes',
        'type': FORMAT_TYPES[F_JSON],
        'title': l10n.translate('Processes', request.locale),
        'href': f"{api.base_url}/processes"
    }, {
        'rel': f'{OGC_RELTYPES_BASE}/tiling-schemes',
        'type': FORMAT_TYPES[F_JSON],
        'title': l10n.translate('The list of supported tiling schemes as JSON', request.locale),  # noqa
        'href': f"{api.base_url}/TileMatrixSets?f=json"
    }, {
        'rel': f'{OGC_RELTYPES_BASE}/tiling-schemes',
        'type': FORMAT_TYPES[F_HTML],
        'title': l10n.translate('The list of supported tiling schemes as HTML', request.locale),  # noqa
        'href': f"{api.base_url}/TileMatrixSets?f=html"
    }]

    if api.pubsub_client is not None and not api.pubsub_client.hidden:
        LOGGER.debug('Adding PubSub broker link')
        pubsub_link = {
            'rel': 'hub',
            'type': 'application/json',
            'title': l10n.translate('Pub/Sub broker', request.locale),
            'href': api.pubsub_client.broker_safe_url
        }
        if api.pubsub_client.channel is not None:
            pubsub_link['channel'] = api.pubsub_client.channel

        fcm['links'].append(pubsub_link)

    if api.manager.is_async:
        fcm['links'].append({
            'rel': f'{OGC_RELTYPES_BASE}/job-list',
            'type': FORMAT_TYPES[F_JSON],
            'title': l10n.translate('Jobs', request.locale),
            'href': f"{api.base_url}/jobs"
        })
        fcm['links'].append({
            'rel': f'{OGC_RELTYPES_BASE}/job-list',
            'type': FORMAT_TYPES[F_HTML],
            'title': l10n.translate('Jobs', request.locale),
            'href': f"{api.base_url}/jobs?f=html"
        })

    if api.asyncapi:
        fcm['links'].append({
            'rel': 'service-doc',
            'type': 'text/html',
            'title': l10n.translate('The AsyncAPI definition as HTML', request.locale),  # noqa
            'href': f'{api.base_url}/asyncapi?f=html'
        })
        fcm['links'].append({
            'rel': 'service-desc',
            'type': 'application/asyncapi+json',
            'title': l10n.translate('The AsyncAPI definition as JSON', request.locale),  # noqa
            'href': f'{api.base_url}/asyncapi?f=json'
        })

    headers = request.get_response_headers(**api.api_headers)
    if request.format == F_HTML:  # render

        for resource_type in ['collection', 'process', 'stac-collection']:
            fcm[resource_type] = False

            found = filter_dict_by_key_value(api.config['resources'],
                                             'type', resource_type)
            if found:
                fcm[resource_type] = True
                if resource_type == 'collection':  # check for tiles
                    for key, value in found.items():
                        if filter_providers_by_type(value['providers'],
                                                    'tile'):
                            fcm['tile'] = True

        if api.manager.is_async:
            fcm['jobs'] = True

        if api.pubsub_client is not None and not api.pubsub_client.hidden:
            fcm['pubsub'] = {
                'name': api.pubsub_client.name,
                'url': api.pubsub_client.broker_safe_url,
                'channel': api.pubsub_client.channel,
                'asyncapi': api.asyncapi
            }

        content = render_j2_template(
            api.tpl_config, api.config['server']['templates'],
            'landing_page.html', fcm, request.locale)

        return headers, HTTPStatus.OK, content

    if request.format == F_JSONLD:
        return headers, HTTPStatus.OK, to_json(
            api.fcmld, api.pretty_print)

    return headers, HTTPStatus.OK, to_json(fcm, api.pretty_print)


def openapi_(api: API, request: APIRequest) -> Tuple[dict, int, str]:
    """
    Provide OpenAPI document

    :param request: A request object
    :param openapi: dict of OpenAPI definition

    :returns: tuple of headers, status code, content
    """
    headers = request.get_response_headers(**api.api_headers)

    if request.format == F_HTML:
        template = 'openapi/swagger.html'
        if request._args.get('ui') == 'redoc':
            template = 'openapi/redoc.html'

        path = f'{api.base_url}/openapi'
        data = {
            'openapi-document-path': path
        }
        content = render_j2_template(
            api.tpl_config, api.config['server']['templates'], template, data,
            request.locale)

        return headers, HTTPStatus.OK, content

    headers['Content-Type'] = 'application/vnd.oai.openapi+json;version=3.0'  # noqa

    if isinstance(api.openapi, dict):
        return headers, HTTPStatus.OK, to_json(api.openapi,
                                               api.pretty_print)
    else:
        return headers, HTTPStatus.OK, api.openapi


def asyncapi_(api: API, request: APIRequest) -> Tuple[dict, int, str]:
    """
    Provide AsyncAPI document

    :param request: A request object

    :returns: tuple of headers, status code, content
    """

    headers = request.get_response_headers(**api.api_headers)

    if not api.asyncapi:
        msg = 'AsyncAPI not supported/configured'
        return api.get_exception(
            HTTPStatus.NOT_IMPLEMENTED, headers, request.format,
            'NoApplicableCode', msg)

    if request.format == F_HTML:
        template = 'asyncapi.html'

        path = f'{api.base_url}/asyncapi'
        data = {
            'asyncapi-document-path': path
        }
        content = render_j2_template(
            api.tpl_config, api.config['server']['templates'], template, data,
            request.locale)

        return headers, HTTPStatus.OK, content

    headers['Content-Type'] = 'application/asyncapi+json'

    return headers, HTTPStatus.OK, to_json(api.asyncapi, api.pretty_print)


def conformance(api: API, request: APIRequest) -> Tuple[dict, int, str]:
    """
    Provide conformance definition

    :param request: A request object

    :returns: tuple of headers, status code, content
    """

    apis_dict = all_apis()

    conformance_list = list(CONFORMANCE_CLASSES)

    for key, value in api.config['resources'].items():
        if value['type'] == 'process':
            conformance_list.extend(
                apis_dict['process'].CONFORMANCE_CLASSES)
        else:
            for provider in value['providers']:
                if provider['type'] in apis_dict:
                    conformance_list.extend(
                        apis_dict[provider['type']].CONFORMANCE_CLASSES)
                if provider['type'] == 'feature':
                    conformance_list.extend(
                        apis_dict['itemtypes'].CONFORMANCE_CLASSES_FEATURES)  # noqa
                if provider['type'] == 'record':
                    conformance_list.extend(
                        apis_dict['itemtypes'].CONFORMANCE_CLASSES_RECORDS)

    if api.pubsub_client is not None:
        conformance_list.extend(apis_dict['pubsub'].CONFORMANCE_CLASSES)

    conformance = {
        'conformsTo': sorted(list(set(conformance_list)))
    }

    headers = request.get_response_headers(**api.api_headers)
    if request.format == F_HTML:  # render
        content = render_j2_template(
            api.tpl_config, api.config['server']['templates'],
            'conformance.html', conformance, request.locale)

        return headers, HTTPStatus.OK, content

    return headers, HTTPStatus.OK, to_json(conformance, api.pretty_print)


@jsonldify
def describe_collections(api: API, request: APIRequest,
                         dataset: str | None = None) -> Tuple[dict, int, str]:
    """
    Provide collection metadata

    :param request: A request object
    :param dataset: name of collection

    :returns: tuple of headers, status code, content
    """

    headers = request.get_response_headers(**api.api_headers)

    fcm = {
        'collections': [],
        'links': []
    }

    collections = filter_dict_by_key_value(api.config['resources'],
                                           'type', 'collection')

    if all([dataset is not None, dataset not in collections.keys()]):
        msg = 'Collection not found'
        return api.get_exception(
            HTTPStatus.NOT_FOUND, headers, request.format, 'NotFound', msg)

    if dataset is not None:
        collections_dict = {dataset: api.config['resources'][dataset]}
    else:
        collections_dict = collections

    LOGGER.debug('Creating collections')
    for k, v in collections_dict.items():
        if v.get('visibility', 'default') == 'hidden' \
                and dataset is None:
            LOGGER.debug(f'Skipping hidden layer: {k}')
            continue

        try:
            fcm['collections'].append(
                gen_collection(api, request, k, request.locale))
        except Exception as err:
            LOGGER.warning(f'Error generating collection {k}: {err}')
            if dataset is None:
                LOGGER.debug('Skipping failed dataset')
            else:
                return api.get_exception(
                    HTTPStatus.INTERNAL_SERVER_ERROR, headers, request.format,
                    'NoApplicableCode', 'Error generating collection')

    if dataset is not None:
        fcm = fcm['collections'][0]

    if dataset is None:
        # TODO: translate
        fcm['links'].append({
            'type': FORMAT_TYPES[F_JSON],
            'rel': request.get_linkrel(F_JSON),
            'title': l10n.translate('This document as JSON', request.locale),  # noqa
            'href': f'{api.get_collections_url()}?f={F_JSON}'
        })
        fcm['links'].append({
            'type': FORMAT_TYPES[F_JSONLD],
            'rel': request.get_linkrel(F_JSONLD),
            'title': l10n.translate('This document as RDF (JSON-LD)', request.locale),  # noqa
            'href': f'{api.get_collections_url()}?f={F_JSONLD}'
        })
        fcm['links'].append({
            'type': FORMAT_TYPES[F_HTML],
            'rel': request.get_linkrel(F_HTML),
            'title': l10n.translate('This document as HTML', request.locale),  # noqa
            'href': f'{api.get_collections_url()}?f={F_HTML}'
        })

    if request.format == F_HTML:  # render
        fcm['base_url'] = api.base_url
        fcm['collections_path'] = api.get_collections_url()
        if dataset is not None:
            tpl_config = api.get_dataset_templates(dataset)
            content = render_j2_template(api.tpl_config, tpl_config,
                                         'collections/collection.html',
                                         fcm, request.locale)
        else:
            content = render_j2_template(
                api.tpl_config, api.config['server']['templates'],
                'collections/index.html', fcm, request.locale)

        return headers, HTTPStatus.OK, content

    if request.format == F_JSONLD:
        jsonld = api.fcmld.copy()
        if dataset is not None:
            jsonld['dataset'] = jsonldify_collection(api, fcm,
                                                     request.locale)
        else:
            jsonld['dataset'] = [
                jsonldify_collection(api, c, request.locale)
                for c in fcm.get('collections', [])
            ]
        return headers, HTTPStatus.OK, to_json(jsonld, api.pretty_print)

    return headers, HTTPStatus.OK, to_json(fcm, api.pretty_print)


def get_collection_schema(api: API, request: Union[APIRequest, Any],
                          dataset: str) -> Tuple[dict, int, str]:
    """
    Returns a collection schema

    :param request: A request object
    :param dataset: dataset name

    :returns: tuple of headers, status code, content
    """

    headers = request.get_response_headers(**api.api_headers)

    if any([dataset is None,
            dataset not in api.config['resources'].keys()]):

        msg = 'Collection not found'
        return api.get_exception(
            HTTPStatus.NOT_FOUND, headers, request.format, 'NotFound', msg)

    LOGGER.debug('Creating collection schema')
    try:
        LOGGER.debug('Loading feature provider')
        p = load_plugin('provider', get_provider_by_type(
            api.config['resources'][dataset]['providers'], 'feature'))
    except ProviderTypeError:
        try:
            LOGGER.debug('Loading coverage provider')
            p = load_plugin('provider', get_provider_by_type(
                api.config['resources'][dataset]['providers'], 'coverage'))  # noqa
        except ProviderTypeError:
            try:
                LOGGER.debug('Loading record provider')
                p = load_plugin('provider', get_provider_by_type(
                    api.config['resources'][dataset]['providers'], 'record'))
            except ProviderTypeError:
                LOGGER.debug('Loading edr provider')
                p = load_plugin('provider', get_provider_by_type(
                    api.config['resources'][dataset]['providers'], 'edr'))
    except ProviderGenericError as err:
        LOGGER.error(err)
        return api.get_exception(
            err.http_status_code, headers, request.format,
            err.ogc_exception_code, err.message)

    schema = {
        'type': 'object',
        'title': l10n.translate(
            api.config['resources'][dataset]['title'], request.locale),
        'properties': {},
        '$schema': 'http://json-schema.org/draft/2019-09/schema',
        '$id': f'{api.get_collections_url()}/{dataset}/schema'
    }

    if p.type != 'coverage':
        schema['properties']['geometry'] = {
            'format': 'geometry-any',
            'x-ogc-role': 'primary-geometry'
        }

    for k, v in p.fields.items():
        if p.properties:
            if k not in p.properties:
                continue

        schema['properties'][k] = v
        if v['type'] == 'float':
            schema['properties'][k]['type'] = 'number'
        if v.get('format') is None:
            schema['properties'][k].pop('format', None)

        if k == p.id_field:
            schema['properties'][k]['x-ogc-role'] = 'id'
        if k == p.time_field:
            schema['properties'][k]['x-ogc-role'] = 'primary-instant'

    if request.format == F_HTML:  # render
        tpl_config = api.get_dataset_templates(dataset)
        schema['title'] = l10n.translate(
            api.config['resources'][dataset]['title'], request.locale)

        schema['collections_path'] = api.get_collections_url()
        schema['dataset_path'] = f'{api.get_collections_url()}/{dataset}'

        content = render_j2_template(api.tpl_config, tpl_config,
                                     'collections/schema.html',
                                     schema, request.locale)

        return headers, HTTPStatus.OK, content

    headers['Content-Type'] = 'application/schema+json'

    return headers, HTTPStatus.OK, to_json(schema, api.pretty_print)


def validate_bbox(value: list | None = None) -> list:
    """
    Helper function to validate bbox parameter

    :param value: `list` of minx, miny, maxx, maxy

    :returns: bbox as `list` of `float` values
    """

    if value is None:
        LOGGER.debug('bbox is empty')
        return []

    bbox = value.split(',')

    if len(bbox) not in [4, 6]:
        msg = 'bbox should be either 4 values (minx,miny,maxx,maxy) ' \
              'or 6 values (minx,miny,minz,maxx,maxy,maxz)'
        LOGGER.debug(msg)
        raise ValueError(msg)

    try:
        bbox = [float(c) for c in bbox]
    except ValueError as err:
        msg = 'bbox values must be numbers'
        err.args = (msg,)
        LOGGER.debug(msg)
        raise

    if (len(bbox) == 4 and bbox[1] > bbox[3]) \
            or (len(bbox) == 6 and bbox[1] > bbox[4]):
        msg = 'miny should be less than maxy'
        LOGGER.debug(msg)
        raise ValueError(msg)

    if (len(bbox) == 4 and bbox[0] > bbox[2]) \
            or (len(bbox) == 6 and bbox[0] > bbox[3]):
        msg = 'minx is greater than maxx (possibly antimeridian bbox)'
        LOGGER.debug(msg)

    if len(bbox) == 6 and bbox[2] > bbox[5]:
        msg = 'minz should be less than maxz'
        LOGGER.debug(msg)
        raise ValueError(msg)

    return bbox


def validate_datetime(resource_def: dict,
                      datetime_: str | None = None) -> str:
    """
    Helper function to validate temporal parameter

    :param resource_def: `dict` of configuration resource definition
    :param datetime_: `str` of datetime parameter

    :returns: `str` of datetime input, if valid
    """

    # TODO: pass datetime to query as a `datetime` object
    # we would need to ensure partial dates work accordingly
    # as well as setting '..' values to `None` so that underlying
    # providers can just assume a `datetime.datetime` object
    #
    # NOTE: needs testing when passing partials from API to backend

    datetime_invalid = False

    if datetime_ is not None and 'temporal' in resource_def:

        dateparse_begin = partial(dateparse, default=datetime.min)
        dateparse_end = partial(dateparse, default=datetime.max)
        unix_epoch = datetime(1970, 1, 1, 0, 0, 0)
        dateparse_ = partial(dateparse, default=unix_epoch)

        te = resource_def['temporal']

        try:
            if te['begin'] is not None and te['begin'].tzinfo is None:
                te['begin'] = te['begin'].replace(tzinfo=pytz.UTC)
            if te['end'] is not None and te['end'].tzinfo is None:
                te['end'] = te['end'].replace(tzinfo=pytz.UTC)
        except AttributeError:
            msg = 'Configured times should be RFC3339'
            LOGGER.error(msg)
            raise ValueError(msg)

        if '/' in datetime_:  # envelope
            LOGGER.debug('detected time range')
            LOGGER.debug('Validating time windows')

            # normalize "" to ".." (actually changes datetime_)
            datetime_ = re.sub(r'^/', '../', datetime_)
            datetime_ = re.sub(r'/$', '/..', datetime_)

            datetime_begin, datetime_end = datetime_.split('/')
            if datetime_begin != '..':
                datetime_begin = dateparse_begin(datetime_begin)
                if datetime_begin.tzinfo is None:
                    datetime_begin = datetime_begin.replace(
                        tzinfo=pytz.UTC)

            if datetime_end != '..':
                datetime_end = dateparse_end(datetime_end)
                if datetime_end.tzinfo is None:
                    datetime_end = datetime_end.replace(tzinfo=pytz.UTC)

            datetime_invalid = any([
                (te['end'] is not None and datetime_begin != '..' and
                    datetime_begin > te['end']),
                (te['begin'] is not None and datetime_end != '..' and
                    datetime_end < te['begin'])
            ])

        else:  # time instant
            LOGGER.debug('detected time instant')
            datetime__ = dateparse_(datetime_)
            if datetime__ != '..':
                if datetime__.tzinfo is None:
                    datetime__ = datetime__.replace(tzinfo=pytz.UTC)
            datetime_invalid = any([
                (te['begin'] is not None and datetime__ != '..' and
                    datetime__ < te['begin']),
                (te['end'] is not None and datetime__ != '..' and
                    datetime__ > te['end'])
            ])

    if datetime_invalid:
        msg = 'datetime parameter out of range'
        LOGGER.debug(msg)
        raise ValueError(msg)

    return datetime_


def validate_subset(value: str) -> dict:
    """
    Helper function to validate subset parameter

    :param value: `subset` parameter

    :returns: dict of axis/values
    """

    subsets = {}

    for s in value.split(','):
        LOGGER.debug(f'Processing subset {s}')
        m = re.search(r'(.*)\((.*)\)', s)
        subset_name, values = m.group(1, 2)

        if '"' in values:
            LOGGER.debug('Values are strings')
            if values.count('"') % 2 != 0:
                msg = 'Invalid format: subset should be like axis("min"[:"max"])'  # noqa
                LOGGER.error(msg)
                raise ValueError(msg)
            try:
                LOGGER.debug('Value is an interval')
                m = re.search(r'"(\S+)":"(\S+)"', values)
                values = list(m.group(1, 2))
            except AttributeError:
                LOGGER.debug('Value is point')
                m = re.search(r'"(.*)"', values)
                values = [m.group(1)]
        else:
            LOGGER.debug('Values are numbers')
            try:
                LOGGER.debug('Value is an interval')
                m = re.search(r'(\S+):(\S+)', values)
                values = list(m.group(1, 2))
            except AttributeError:
                LOGGER.debug('Value is point')
                values = [values]

        subsets[subset_name] = list(map(get_typed_value, values))

    return subsets


def evaluate_limit(requested: Union[None, int], server_limits: dict,
                   collection_limits: dict) -> int:
    """
    Helper function to evaluate limit parameter

    :param requested: the limit requested by the client
    :param server_limits: `dict` of server limits
    :param collection_limits: `dict` of collection limits

    :returns: `int` of evaluated limit
    """

    effective_limits = ChainMap(collection_limits, server_limits)

    default = effective_limits.get('default_items', 10)
    max_ = effective_limits.get('max_items', 10)
    on_exceed = effective_limits.get('on_exceed', 'throttle')

    LOGGER.debug(f'Requested limit: {requested}')
    LOGGER.debug(f'Default limit: {default}')
    LOGGER.debug(f'Maximum limit: {max_}')
    LOGGER.debug(f'On exceed: {on_exceed}')

    if requested is None:
        LOGGER.debug('no limit requested; returning default')
        return default

    if isinstance(requested, int):
        requested2 = requested
    else:
        requested2 = get_typed_value(requested)

    if not isinstance(requested2, int):
        raise ValueError('limit value should be an integer')

    if requested2 <= 0:
        raise ValueError('limit value should be strictly positive')
    elif requested2 > max_ and on_exceed == 'error':
        raise ValueError(f'Limit of {max_} exceeded')
    else:
        LOGGER.debug('limit requested')
        return min(requested2, max_)