Tiles Metadata provider refactor

docs/source/data-publishing/ogcapi-tiles.rst

@@ -46,7 +46,6 @@ This code block shows how to configure pygeoapi to read Mapbox vector tiles gene
          data: tests/data/tiles/ne_110m_lakes  # local directory tree
          # data: http://localhost:9000/ne_110m_lakes/{z}/{x}/{y}.pbf # tiles stored on a MinIO bucket
          options:
-             metadata_format: default # default | tilejson
              zoom:
                  min: 0
                  max: 5
@@ -77,7 +76,6 @@ This code block shows how to configure pygeoapi to read Mapbox vector tiles from
          # if you don't use precision 0, you will be requesting for aggregations which are not supported in the 
          # free version of elastic
          options:
-             metadata_format: default # default | tilejson
              zoom:
                  min: 0
                  max: 5

pygeoapi/api.py

@@ -74,6 +74,7 @@
 from pygeoapi.provider.base import (
     ProviderGenericError, ProviderConnectionError, ProviderNotFoundError,
     ProviderTypeError)
+from pygeoapi.models.provider.base import TilesMetadataFormat
 
 from pygeoapi.models.cql import CQLModel
 from pygeoapi.util import (dategetter, RequestedProcessExecutionMode,
@@ -85,8 +86,6 @@
                            get_crs_from_uri, get_supported_crs_list,
                            CrsTransformSpec, transform_bbox)
 
-from pygeoapi.models.provider.base import TilesMetadataFormat
-
 LOGGER = logging.getLogger(__name__)
 
 #: Return headers for requests (e.g:X-Powered-By)
@@ -2686,15 +2685,12 @@ def get_collection_tiles(self, request: Union[APIRequest, Any],
 
             tiles['tilesets'].append(tile_matrix)
 
-        metadata_format = p.options['metadata_format']
-
         if request.format == F_HTML:  # render
             tiles['id'] = dataset
             tiles['title'] = l10n.translate(
                 self.config['resources'][dataset]['title'], SYSTEM_LOCALE)
             tiles['tilesets'] = [
                 scheme.tileMatrixSet for scheme in p.get_tiling_schemes()]
-            tiles['format'] = metadata_format
             tiles['bounds'] = \
                 self.config['resources'][dataset]['extents']['spatial']['bbox']
             tiles['minzoom'] = p.options['zoom']['min']
@@ -2789,7 +2785,7 @@ def get_collection_tiles_metadata(
         :returns: tuple of headers, status code, content
         """
 
-        if not request.is_valid():
+        if not request.is_valid([TilesMetadataFormat.TILEJSON]):
             return self.get_format_exception(request)
         headers = request.get_response_headers(**self.api_headers)
 
@@ -2825,47 +2821,30 @@ def get_collection_tiles_metadata(
             return self.get_exception(HTTPStatus.NOT_FOUND, headers,
                                       request.format, 'NotFound', msg)
 
-        metadata_format = TilesMetadataFormat[
-            str(p.options['metadata_format']).upper()]
-
         # Set response language to requested provider locale
         # (if it supports language) and/or otherwise the requested pygeoapi
         # locale (or fallback default locale)
         l10n.set_response_language(headers, prv_locale, request.locale)
 
-        if request.format == F_HTML:  # render
-            tiles_metadata = p.get_metadata(
-                dataset=dataset, server_url=self.base_url,
-                layer=p.get_layer(), tileset=matrix_id,
-                metadata_format=TilesMetadataFormat.TILEJSON,
-                language=prv_locale)
-            metadata = dict()
-            metadata['metadata'] = tiles_metadata
-            metadata['id'] = dataset
-            metadata['title'] = l10n.translate(
-                self.config['resources'][dataset]['title'], request.locale)
-            metadata['tileset'] = matrix_id
-            metadata['format'] = metadata_format.value
-            metadata['collections_path'] = self.get_collections_url()
+        tiles_metadata = p.get_metadata(
+            dataset=dataset, server_url=self.base_url,
+            layer=p.get_layer(), tileset=matrix_id,
+            metadata_format=request._format, title=l10n.translate(
+                self.config['resources'][dataset]['title'],
+                request.locale),
+            description=l10n.translate(
+                self.config['resources'][dataset]['description'],
+                request.locale),
+            language=prv_locale)
 
+        if request.format == F_HTML:  # render
             content = render_j2_template(self.tpl_config,
                                          'collections/tiles/metadata.html',
-                                         metadata, request.locale)
+                                         tiles_metadata, request.locale)
 
             return headers, HTTPStatus.OK, content
         else:
-            tiles_metadata = p.get_metadata(
-                dataset=dataset, server_url=self.base_url,
-                layer=p.get_layer(), tileset=matrix_id,
-                metadata_format=metadata_format, title=l10n.translate(
-                    self.config['resources'][dataset]['title'],
-                    request.locale),
-                description=l10n.translate(
-                    self.config['resources'][dataset]['description'],
-                    request.locale),
-                language=prv_locale)
-
-        return headers, HTTPStatus.OK, tiles_metadata
+            return headers, HTTPStatus.OK, tiles_metadata
 
     @gzip
     @pre_process

pygeoapi/models/provider/base.py

@@ -38,13 +38,14 @@
 from pydantic import BaseModel
 
 
-class TilesMetadataFormat(Enum):
+class TilesMetadataFormat(str, Enum):
     # Tile Set Metadata
-    DEFAULT = "Tile Set Metadata"
+    JSON = "JSON"
+    JSONLD = "JSONLD"
     # TileJSON 3.0
-    TILEJSON = "TileJSON"
-    # Custom JSON
-    CUSTOMJSON = "Custom"
+    TILEJSON = "TILEJSON"
+    # HTML (default)
+    HTML = "HTML"
 
 
 # Tile Set Metadata Enums

pygeoapi/provider/base_mvt.py

@@ -29,19 +29,13 @@
 #
 # =================================================================
 
-import json
 import logging
-import requests
-from pathlib import Path
 from urllib.parse import urlparse
 
 from pygeoapi.provider.tile import BaseTileProvider
-from pygeoapi.provider.base import ProviderConnectionError
 from pygeoapi.models.provider.base import (
-    TileMatrixSetEnum, TilesMetadataFormat, TileSetMetadata, LinkType,
-    GeospatialDataType)
-from pygeoapi.models.provider.mvt import MVTTilesJson
-from pygeoapi.util import is_url, url_join
+    TileMatrixSetEnum, TilesMetadataFormat)
+from pygeoapi.util import url_join
 
 LOGGER = logging.getLogger(__name__)
 
@@ -125,100 +119,102 @@ def get_tiles(self, layer=None, tileset=None,
 
         raise NotImplementedError()
 
+    def get_html_metadata(self, dataset, server_url, layer, tileset,
+                          title, description, keywords, **kwargs):
+        """
+        Gets tile metadata informations in html format
+
+        :param dataset: dataset name
+        :param server_url: server base url
+        :param layer: mvt tile layer name
+        :param tileset: mvt tileset name
+        :param metadata_format: format for metadata,
+                            enum TilesMetadataFormat
+        :param title: title name
+        :param description: description name
+        :param keywords: keywords list
+
+        :returns: `dict` of JSON metadata
+        """
+
+        raise NotImplementedError()
+
+    def get_default_metadata(self, dataset, server_url, layer, tileset,
+                             title, description, keywords, **kwargs):
+        """
+        Gets tile metadata in default Tile Set Metadata format
+
+        :param dataset: dataset name
+        :param server_url: server base url
+        :param layer: mvt tile layer name
+        :param tileset: mvt tileset name
+        :param metadata_format: format for metadata,
+                            enum TilesMetadataFormat
+        :param title: title name
+        :param description: description name
+        :param keywords: keywords list
+
+        :returns: `dict` of JSON metadata
+        """
+        raise NotImplementedError()
+
+    def get_vendor_metadata(self, dataset, server_url, layer, tileset,
+                              title, description, keywords, **kwargs):
+        """
+        Gets tile metadata in Tilejson format
+
+        :param dataset: dataset name
+        :param server_url: server base url
+        :param layer: mvt tile layer name
+        :param tileset: mvt tileset name
+        :param metadata_format: format for metadata,
+                            enum TilesMetadataFormat
+        :param title: title name
+        :param description: description name
+        :param keywords: keywords list
+
+        :returns: `dict` of JSON metadata
+        """
+
+        raise NotImplementedError()
+
     def get_metadata(self, dataset, server_url, layer=None,
                      tileset=None, metadata_format=None, title=None,
                      description=None, keywords=None, **kwargs):
         """
-        Gets tile metadata
+        Gets tiles metadata
 
         :param dataset: dataset name
         :param server_url: server base url
         :param layer: mvt tile layer name
         :param tileset: mvt tileset name
         :param metadata_format: format for metadata,
                             enum TilesMetadataFormat
+        :param title: title name
+        :param description: description name
+        :param keywords: keywords list
 
         :returns: `dict` of JSON metadata
         """
 
-        if is_url(self.data):
-            url = urlparse(self.data)
-            base_url = f'{url.scheme}://{url.netloc}'
-            if metadata_format == TilesMetadataFormat.TILEJSON:
-                with requests.Session() as session:
-                    session.get(base_url)
-                    resp = session.get(f'{base_url}/{layer}/metadata.json')
-                    resp.raise_for_status()
-                metadata_json_content = resp.json()
-        else:
-            if not isinstance(self.service_metadata_url, Path):
-                msg = f'Wrong data path configuration: {self.service_metadata_url}'  # noqa
-                LOGGER.error(msg)
-                raise ProviderConnectionError(msg)
-
-            if self.service_metadata_url.exists():
-                with open(self.service_metadata_url, 'r') as md_file:
-                    metadata_json_content = json.loads(md_file.read())
-
-        service_url = url_join(
-            server_url,
-            f'collections/{dataset}/tiles/{tileset}/{{tileMatrix}}/{{tileRow}}/{{tileCol}}?f=mvt')  # noqa
-
-        content = {}
-        if metadata_format == TilesMetadataFormat.TILEJSON:
-            if 'metadata_json_content' in locals():
-                content = MVTTilesJson(**metadata_json_content)
-                content.tiles = service_url
-                content.vector_layers = json.loads(
-                        metadata_json_content["json"])["vector_layers"]
-                return content.dict()
-            else:
-                msg = f'No tiles metadata json available: {self.service_metadata_url}'  # noqa
-                LOGGER.error(msg)
-                raise ProviderConnectionError(msg)
-        elif metadata_format == TilesMetadataFormat.CUSTOMJSON:
-            if 'metadata_json_content' in locals():
-                content = metadata_json_content
-                if 'json' in metadata_json_content:
-                    content['json'] = json.loads(metadata_json_content['json'])
-                return content
-            else:
-                msg = f'No custom JSON for tiles metadata available: {self.service_metadata_url}'  # noqa
-                LOGGER.error(msg)
-                raise ProviderConnectionError(msg)
+        if metadata_format.upper() == TilesMetadataFormat.JSON:
+            return self.get_default_metadata(dataset, server_url, layer,
+                                             tileset, title, description,
+                                             keywords, **kwargs)
+        elif metadata_format.upper() == TilesMetadataFormat.TILEJSON:
+            return self.get_vendor_metadata(dataset, server_url, layer,
+                                              tileset, title, description,
+                                              keywords, **kwargs)
+        elif metadata_format.upper() == TilesMetadataFormat.HTML:
+            return self.get_html_metadata(dataset, server_url, layer,
+                                          tileset, title, description,
+                                          keywords, **kwargs)
+        elif metadata_format.upper() == TilesMetadataFormat.JSONLD:
+            return self.get_default_metadata(dataset, server_url, layer,
+                                             tileset, title, description,
+                                             keywords, **kwargs)
         else:
-            tiling_schemes = self.get_tiling_schemes()
-            # Default values
-            tileMatrixSetURI = tiling_schemes[0].tileMatrixSetURI
-            crs = tiling_schemes[0].crs
-            # Checking the selected matrix in configured tiling_schemes
-            for schema in tiling_schemes:
-                if (schema.tileMatrixSet == tileset):
-                    crs = schema.crs
-                    tileMatrixSetURI = schema.tileMatrixSetURI
-
-            content = TileSetMetadata(title=title, description=description,
-                                      keywords=keywords, crs=crs,
-                                      tileMatrixSetURI=tileMatrixSetURI)
-
-            links = []
-            service_url_link_type = "application/vnd.mapbox-vector-tile"
-            service_url_link_title = f'{tileset} vector tiles for {layer}'
-            service_url_link = LinkType(href=service_url, rel="item",
-                                        type=service_url_link_type,
-                                        title=service_url_link_title)
-            links.append(service_url_link)
-
-            content.links = links
-
-            if 'metadata_json_content' in locals():
-                vector_layers = json.loads(
-                    metadata_json_content["json"])["vector_layers"]
-                layers = []
-                for vector_layer in vector_layers:
-                    layers.append(GeospatialDataType(id=vector_layer['id']))
-                content.layers = layers
-            return content.dict(exclude_none=True)
+            raise NotImplementedError(f"_{metadata_format.upper()}_ is not supported") # noqa
 
     def get_tms_links(self):
         """