Fix joins and features routes for new JoinManager

Author: GeoSander

pygeoapi/api/__init__.py

@@ -629,15 +629,14 @@ def __init__(self, config: dict, openapi: dict) -> Self | None:
         self.api_headers = get_api_rules(self.config).response_headers
         self.base_url = get_base_url(self.config)
         self.prefetcher = UrlPrefetcher()
-        self.supports_joins = False
+        self.join_manager = None
 
         setup_logger(self.config['logging'])
 
-        joins_init = getattr(all_apis().get('joins', object()), 'init', None)
-        if callable(joins_init):
-            # Initialize OGC API - Joins:
-            # build reference cache of join tables already/still on the server
-            self.supports_joins = joins_init(config)
+        join_manager = getattr(all_apis().get('joins', object()), 'get_manager', None)  # noqa
+        if callable(join_manager):
+            # Init OGC API - Joins manager (or set to None if not configured)
+            self.join_manager = join_manager(config)
 
         CHARSET[0] = config['server'].get('encoding', 'utf-8')
         if config['server'].get('gzip'):
@@ -911,7 +910,7 @@ def conformance(api: API, request: APIRequest) -> Tuple[dict, int, str]:
                         apis_dict['itemtypes'].CONFORMANCE_CLASSES_FEATURES)
                     # If it's an OGC API - Features provider and joins is
                     # supported, we can also add conformance classes for it
-                    if api.supports_joins:
+                    if api.join_manager:
                         conformance_list.extend(
                             apis_dict['joins'].CONFORMANCE_CLASSES)
                 if provider['type'] == 'record':
@@ -1145,7 +1144,7 @@ def describe_collections(api: API, request: APIRequest,
                 'href': f'{api.get_collections_url()}/{k}/items?f={F_HTML}'  # noqa
             })
 
-            if api.supports_joins and collection_data_type == 'feature':
+            if api.join_manager and collection_data_type == 'feature':
                 # Add links to available OGC API - Joins sources, if any
                 collection['links'].extend([{
                     'type': FORMAT_TYPES[F_JSON],
@@ -1366,7 +1365,7 @@ def describe_collections(api: API, request: APIRequest,
         })
 
     if request.format == F_HTML:  # render
-        fcm['supports_joins'] = api.supports_joins
+        fcm['supports_joins'] = api.join_manager is not None
         fcm['base_url'] = api.base_url
         fcm['collections_path'] = api.get_collections_url()
         if dataset is not None:

pygeoapi/api/itemtypes.py

@@ -47,7 +47,7 @@
 from pygeofilter.parsers.cql2_json import parse as parse_cql2_json
 from pyproj.exceptions import CRSError
 
-from pygeoapi import l10n, join_util
+from pygeoapi import l10n
 from pygeoapi.api import evaluate_limit
 from pygeoapi.crs import (DEFAULT_CRS, DEFAULT_STORAGE_CRS,
                           create_crs_transform_spec, get_supported_crs_list,
@@ -58,7 +58,8 @@
 from pygeoapi.plugin import load_plugin, PLUGINS
 from pygeoapi.provider.base import (
     ProviderGenericError, ProviderTypeError, SchemaType)
-
+from pygeoapi.join.manager import (
+    JoinManager, JoinSourceMissingError, JoinSourceNotFoundError)
 from pygeoapi.util import (filter_providers_by_type, to_json,
                            filter_dict_by_key_value, str2bool,
                            get_provider_by_type, render_j2_template)
@@ -571,20 +572,18 @@ def get_collection_items(
     if join_id:
         # Post-process FeatureCollection and join with CSV data
         try:
-            join_util.perform_join(content, dataset, join_id)
+            api.join_manager.perform_join(content, dataset, join_id)
         except ValueError as e:
             LOGGER.error(f'Invalid request parameter: {e}',
                          exc_info=True)
             return api.get_exception(
                 HTTPStatus.BAD_REQUEST, headers, request.format,
                 'InvalidParameterValue', str(e))
-        except KeyError as e:
-            msg = 'Join source not found'
-            LOGGER.error(f'Unknown joinId: {e}',
-                         exc_info=True)
+        except (JoinSourceNotFoundError, JoinSourceMissingError) as e:
+            LOGGER.error(f'{e}', exc_info=True)
             return api.get_exception(
                 HTTPStatus.NOT_FOUND, headers, request.format,
-                'NotFound', msg)
+                'NotFound', str(e))
         except Exception as e:
             LOGGER.error(f'Failed to perform join: {e}',
                          exc_info=True)
@@ -999,20 +998,18 @@ def get_collection_item(api: API, request: APIRequest,
     if join_id:
         # Post-process Feature and join with CSV data
         try:
-            join_util.perform_join(content, dataset, join_id)
+            api.join_manager.perform_join(content, dataset, join_id)
         except ValueError as e:
             LOGGER.error(f'Invalid request parameter: {e}',
                          exc_info=True)
             return api.get_exception(
                 HTTPStatus.BAD_REQUEST, headers, request.format,
                 'InvalidParameterValue', str(e))
-        except KeyError as e:
-            msg = 'Join source not found'
-            LOGGER.error(f'Unknown joinId: {e}',
-                         exc_info=True)
+        except (JoinSourceNotFoundError, JoinSourceMissingError) as e:
+            LOGGER.error(f'{e}', exc_info=True)
             return api.get_exception(
                 HTTPStatus.NOT_FOUND, headers, request.format,
-                'NotFound', msg)
+                'NotFound', str(e))
         except Exception as e:
             LOGGER.error(f'Failed to perform join: {e}',
                          exc_info=True)
@@ -1104,7 +1101,7 @@ def get_oas_30(cfg: dict, locale: str) -> tuple[list[dict[str, str]], dict[str,
     from pygeoapi.openapi import OPENAPI_YAML, get_visible_collections
 
     # We should add a joinId query parameter if OGC API - Joins is enabled
-    joins_enabled = join_util.enabled(cfg)
+    joins_enabled = JoinManager.configured(cfg)
 
     join_id_param = {
         'name': 'joinId',
@@ -1415,9 +1412,13 @@ def get_oas_30(cfg: dict, locale: str) -> tuple[list[dict[str, str]], dict[str,
                 }
             }
 
+            if joins_enabled:
+                # Inject joinId parameter into GET /items/{featureId} operation
+                paths[f'{items_path}/{{featureId}}']['get']['parameters'].append(join_id_param)  # noqa
+
             try:
                 schema_ref = p.get_schema()
-                paths[f'{collection_name_path}/items/{{featureId}}']['get']['responses']['200'] = {  # noqa
+                paths[f'{items_path}/{{featureId}}']['get']['responses']['200'] = {  # noqa
                     'content': {
                         schema_ref[0]: {
                             'schema': schema_ref[1]

pygeoapi/api/joins.py

@@ -73,16 +73,16 @@
 ]
 
 
-def init(cfg: dict) -> bool:
+def get_manager(cfg: dict) -> JoinManager | None:
     """
-    Shortcut to initialize join utility with config.
+    Shortcut to initialize a JoinManager instance.
     Called dynamically by the main `API.__init__` method.
 
     :param cfg: pygeoapi configuration dict
 
-    :returns: True if OGC API - Joins is available and initialized
+    :returns: `JoinManager` instance or None (if OGC API - Joins is disabled)
     """
-    return join_util.init(cfg)
+    return JoinManager.from_config(cfg)
 
 
 def get_oas_30(cfg: dict, locale: str) -> tuple[list[dict[str, str]], dict[str, dict]]:  # noqa
@@ -97,7 +97,7 @@ def get_oas_30(cfg: dict, locale: str) -> tuple[list[dict[str, str]], dict[str,
 
     paths = {}
 
-    if not join_util.enabled(cfg):
+    if not get_manager(cfg):
         LOGGER.info('OpenAPI: skipping OGC API - Joins endpoints setup')
         return [], {'paths': paths}
 
@@ -540,7 +540,7 @@ def list_joins(api: API, request: APIRequest,
         return _not_found(api, request, headers, msg)
 
     try:
-        sources = join_util.list_sources(dataset)
+        sources = api.join_manager.list_sources(dataset)
     except Exception as e:
         LOGGER.error(str(e), exc_info=True)
         return _server_error(api, request, headers, str(e))
@@ -645,7 +645,7 @@ def join_details(api: API, request: APIRequest,
         return _not_found(api, request, headers, msg)
 
     try:
-        details = join_util.read_join_source(dataset, join_id)
+        details = api.join_manager.read_join_source(dataset, join_id)
 
         uri = f'{api.get_collections_url()}/{dataset}'
         output = {
@@ -697,10 +697,9 @@ def join_details(api: API, request: APIRequest,
     except ValueError as e:
         LOGGER.error(f'Invalid request parameter: {e}', exc_info=True)
         return _bad_request(api, request, headers, str(e))
-    except KeyError as e:
-        msg = 'Collection or join source not found'
-        LOGGER.error(f'Invalid parameter value: {e}', exc_info=True)
-        return _not_found(api, request, headers, msg)
+    except (JoinSourceNotFoundError, JoinSourceMissingError) as e:
+        LOGGER.error(f'{e}', exc_info=True)
+        return _not_found(api, request, headers, str(e))
     except Exception as e:
         LOGGER.error(f'Failed to retrieve join: {e}', exc_info=True)
         msg = f'Failed to retrieve join: {str(e)}'
@@ -739,7 +738,7 @@ def create_join(api: API, request: APIRequest,
     """
     headers, collections, dataset = _prepare(api, request, collection)
 
-    if not api.supports_joins:
+    if not api.join_manager:
         # TODO: perhaps a 406 Not Acceptable would be better?
         msg = 'OGC API - Joins is not available on this instance'
         return _server_error(api, request, headers, msg)
@@ -768,7 +767,7 @@ def create_join(api: API, request: APIRequest,
         # Get provider locale (if any)
         prv_locale = l10n.get_plugin_locale(provider_def, request.raw_locale)
 
-        details = join_util.process_csv(dataset, provider, request.form)
+        details = api.join_manager.process_csv(dataset, provider, request.form)
 
         uri = f'{api.get_collections_url()}/{dataset}'
         join_id = details['id']
@@ -865,13 +864,14 @@ def delete_join(api: API, request: APIRequest,
         return _not_found(api, request, headers, msg)
 
     try:
-        if not join_util.remove_source(dataset, join_id):
+        if not api.join_manager.remove_source(dataset, join_id):
             msg = f'Join source {join_id} not found for collection {dataset}'
             return _not_found(api, request, headers, msg)
     except ValueError as e:
         LOGGER.error(f'Invalid request parameter: {e}', exc_info=True)
         return _bad_request(api, request, headers, str(e))
     except Exception as e:
+        # e.g. unable to delete file from disk
         LOGGER.error(f'Failed to delete join source: {e}',
                      exc_info=True)
         msg = f'Failed to delete join: {str(e)}'

pygeoapi/join/manager.py

@@ -78,7 +78,8 @@ def __init__(self, source_dir: Path, **kwargs):
         self._cleanup_sources()
 
         LOGGER.debug(
-            f'JoinManager initialized with source_dir: {self._source_dir}')
+            f'{self.__class__.__name__} initialized with source_dir: '
+            f'{self._source_dir}')
 
     @property
     def source_dir(self) -> Path:
@@ -95,6 +96,16 @@ def max_files(self) -> int:
         """Maximum number of join source files to keep at any time."""
         return self._max_files
 
+    @staticmethod
+    def configured(config: dict) -> bool:
+        """Quick check to see if OGC API - Joins section is present in the
+        pygeoapi config dict, without actually initializing a JoinManager."""
+        try:
+            return 'joins' in config.get('server', {})
+        except KeyError:
+            # pygeoapi was configured without OGC API - Joins
+            return False
+
     @classmethod
     def from_config(cls, config: dict) -> Optional['JoinManager']:
         """
@@ -108,6 +119,7 @@ def from_config(cls, config: dict) -> Optional['JoinManager']:
             joins_config = config.get('server', {})['joins']
         except KeyError:
             # pygeoapi was configured without OGC API - Joins:
+            LOGGER.debug('pygeoapi server.joins not configured')
             return None
 
         # Check if 'joins' key was set, but without further configuration
@@ -156,10 +168,9 @@ def from_config(cls, config: dict) -> Optional['JoinManager']:
         # Create and return manager
         try:
             manager = cls(source_dir, max_days=max_days, max_files=max_files)
-            LOGGER.debug('JoinManager successfully created')
             return manager
         except Exception as e:
-            LOGGER.error(f'Failed to create JoinManager: {e}')
+            LOGGER.error(f'Failed to initialize {cls.__name__}: {e}')
             return None
 
     @contextmanager
@@ -649,7 +660,7 @@ def remove_source(self, collection_id: str, join_id: str) -> bool:
             # Source file was removed, but reference still existed: 200
             return True
 
-        # Remove the JSON file from disk
+        # Remove the JSON file from disk (note: this is noisy)
         deleted = self._delete_source(source_path)
 
         # Remove reference (clean up orphan)