Extend plugin system with PluginContext while maintaining backwards

compatibility

Author: francbartoli

pygeoapi/plugin.py

@@ -1,8 +1,10 @@
 # =================================================================
 #
 # Authors: Tom Kralidis <tomkralidis@gmail.com>
+#          Francesco Bartoli <xbartolone@gmail.com>
 #
 # Copyright (c) 2026 Tom Kralidis
+# Copyright (c) 2026 Francesco Bartoli
 #
 # Permission is hereby granted, free of charge, to any person
 # obtaining a copy of this software and associated documentation
@@ -30,10 +32,57 @@
 
 import importlib
 import logging
-from typing import Any
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
 
 LOGGER = logging.getLogger(__name__)
 
+
+@dataclass
+class PluginContext:
+    """
+    Inject dependencies with a context object into plugins.
+
+    This allows passing runtime dependencies to plugins without
+    relying on global state or complex config dictionaries.
+
+    Attributes:
+        config: Original plugin configuration dictionary
+        logger: Optional injected logger instance
+        locales: Optional list of supported locale codes
+        base_url: Optional API base URL for link generation
+
+    Example:
+        >>> from pygeoapi.plugin import PluginContext, load_plugin
+        >>> context = PluginContext(
+        ...     config={'name': 'GeoJSON', 'type': 'feature', 'data': 'obs.geojson'},
+        ...     logger=custom_logger,
+        ...     base_url='https://api.example.com'
+        ... )
+        >>> provider = load_plugin('provider', context.config, context=context)
+    """
+
+    config: Dict[str, Any]
+    logger: Optional[Any] = None
+    locales: Optional[List[str]] = None
+    base_url: Optional[str] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        """
+        Convert to plain dict format for backwards compatibility.
+
+        :returns: Dictionary with config and injected dependencies
+        """
+        result = dict(self.config)
+        if self.logger:
+            result["_logger"] = self.logger
+        if self.base_url:
+            result["_base_url"] = self.base_url
+        if self.locales:
+            result["_locales"] = self.locales
+        return result
+
+
 #: Loads provider plugins to be used by pygeoapi,\
 #: formatters and processes available
 PLUGINS = {
@@ -94,14 +143,32 @@
 }
 
 
-def load_plugin(plugin_type: str, plugin_def: dict) -> Any:
+def load_plugin(
+    plugin_type: str, plugin_def: dict, context: Optional[PluginContext] = None
+) -> Any:
     """
-    loads plugin by name
+    Loads plugin by name with optional dependency injection.
 
-    :param plugin_type: type of plugin (provider, formatter)
-    :param plugin_def: plugin definition
+    :param plugin_type: type of plugin (provider, formatter, process, etc.)
+    :param plugin_def: plugin definition dictionary
+    :param context: optional context with injected dependencies
 
     :returns: plugin object
+
+    Example:
+        # Plain mode (backwards compatible)
+        >>> provider = load_plugin('provider', {
+        ...     'name': 'GeoJSON',
+        ...     'type': 'feature',
+        ...     'data': 'obs.geojson'
+        ... })
+
+        # Modern mode with dependencies
+        >>> context = PluginContext(
+        ...     config={'name': 'GeoJSON', 'type': 'feature', 'data': 'obs.geojson'},
+        ...     logger=custom_logger
+        ... )
+        >>> provider = load_plugin('provider', context.config, context=context)
     """
 
     name = plugin_def['name']
@@ -130,7 +197,23 @@ def load_plugin(plugin_type: str, plugin_def: dict) -> Any:
 
     module = importlib.import_module(packagename)
     class_ = getattr(module, classname)
-    plugin = class_(plugin_def)
+
+    # Support injected dependencies via PluginContext
+    if context is not None:
+        # Try context-aware constructor first
+        try:
+            plugin = class_(plugin_def, context=context)
+            LOGGER.debug(f"{name} initialized with PluginContext")
+        except TypeError as err:
+            # Fallback: legacy constructor without context parameter
+            LOGGER.debug(
+                f"{name} does not support PluginContext, "
+                f"using legacy init: {err}"
+            )
+            plugin = class_(plugin_def)
+    else:
+        # Plain mode: no more context provided
+        plugin = class_(plugin_def)
 
     return plugin
 

pygeoapi/process/base.py

@@ -2,9 +2,11 @@
 #
 # Authors: Tom Kralidis <tomkralidis@gmail.com>
 #          Francesco Martinelli <francesco.martinelli@ingv.it>
+#          Francesco Bartoli <xbartolone@gmail.com>
 #
 # Copyright (c) 2022 Tom Kralidis
 # Copyright (c) 2024 Francesco Martinelli
+# Copyright (c) 2026 Francesco Bartoli
 #
 # Permission is hereby granted, free of charge, to any person
 # obtaining a copy of this software and associated documentation
@@ -29,23 +31,31 @@
 #
 # =================================================================
 
+from __future__ import annotations
+
 import logging
-from typing import Any, Tuple, Optional
+from typing import Any, Optional, Tuple, TYPE_CHECKING
 
 from pygeoapi.error import GenericError
 
+if TYPE_CHECKING:
+    from pygeoapi.plugin import PluginContext
+
 LOGGER = logging.getLogger(__name__)
 
 
 class BaseProcessor:
     """generic Processor ABC. Processes are inherited from this class"""
 
-    def __init__(self, processor_def: dict, process_metadata: dict):
+    def __init__(self, processor_def: dict, process_metadata: dict,
+                 context: Optional[PluginContext] = None,
+                ):
         """
         Initialize object
 
         :param processor_def: processor definition
         :param process_metadata: process metadata `dict`
+        :param context: optional PluginContext with injected dependencies
 
         :returns: pygeoapi.processor.base.BaseProvider
         """
@@ -54,6 +64,18 @@ def __init__(self, processor_def: dict, process_metadata: dict):
         self.metadata = process_metadata
         self.supports_outputs = False
 
+        # Dependencies support
+        self._context = context
+        if context and context.logger:
+            self._logger = context.logger
+        else:
+            self._logger = LOGGER  # Global fallback
+
+    @property
+    def logger(self):
+        """Get logger (injected or global)"""
+        return self._logger
+
     def set_job_id(self, job_id: str) -> None:
         """
         Set the job_id within the processor

pygeoapi/process/manager/base.py

@@ -31,17 +31,22 @@
 #
 # =================================================================
 
+from __future__ import annotations
+
 import collections
 import json
 import logging
+import uuid
 from multiprocessing import dummy
 from pathlib import Path
-from typing import Any, Dict, Tuple, Optional, OrderedDict
-import uuid
+from typing import Any, Dict, Optional, OrderedDict, Tuple, TYPE_CHECKING
 
 import requests
 
 from pygeoapi.plugin import load_plugin
+
+if TYPE_CHECKING:
+    from pygeoapi.plugin import PluginContext
 from pygeoapi.process.base import (
     BaseProcessor,
     JobNotFoundError,
@@ -64,11 +69,12 @@ class BaseManager:
     """generic Manager ABC"""
     processes: OrderedDict[str, Dict]
 
-    def __init__(self, manager_def: dict):
+    def __init__(self, manager_def: dict, context: Optional[PluginContext] = None):
         """
         Initialize object
 
         :param manager_def: manager definition
+        :param context: optional PluginContext with injected dependencies
 
         :returns: `pygeoapi.process.manager.base.BaseManager`
         """
@@ -91,6 +97,18 @@ def __init__(self, manager_def: dict):
         for id_, process_conf in manager_def.get('processes', {}).items():
             self.processes[id_] = dict(process_conf)
 
+        # Dependencies support
+        self._context = context
+        if context and context.logger:
+            self._logger = context.logger
+        else:
+            self._logger = LOGGER  # Global fallback
+
+    @property
+    def logger(self):
+        """Get logger (injected or global)"""
+        return self._logger
+
     def get_processor(self, process_id: str) -> BaseProcessor:
         """Instantiate a processor.
 
@@ -426,7 +444,7 @@ def execute_process(
             # do we support sync?
             process_supports_sync = (
                 ProcessExecutionMode.sync_execute.value in job_control_options
-                )
+            )
             if not process_supports_sync:
                 LOGGER.debug('Asynchronous execution')
                 handler = self._execute_handler_async
@@ -489,7 +507,7 @@ def _send_in_progress_notification(self, subscriber: Optional[Subscriber]):
             )
 
     def _send_success_notification(
-            self, subscriber: Optional[Subscriber], outputs: Any
+        self, subscriber: Optional[Subscriber], outputs: Any
     ):
         if subscriber:
             response = requests.post(subscriber.success_uri, json=outputs)

pygeoapi/provider/base.py

@@ -27,14 +27,20 @@
 #
 # =================================================================
 
+from __future__ import annotations
+
 import json
 import logging
 from enum import Enum
 from http import HTTPStatus
+from typing import TYPE_CHECKING, Optional
 
 from pygeoapi.crs import DEFAULT_STORAGE_CRS, get_crs
 from pygeoapi.error import GenericError
 
+if TYPE_CHECKING:
+    from pygeoapi.plugin import PluginContext
+
 LOGGER = logging.getLogger(__name__)
 
 
@@ -48,11 +54,12 @@ class SchemaType(Enum):
 class BaseProvider:
     """generic Provider ABC"""
 
-    def __init__(self, provider_def):
+    def __init__(self, provider_def, context: Optional[PluginContext] = None):
         """
         Initialize object
 
         :param provider_def: provider definition
+        :param context: optional PluginContext with injected dependencies
 
         :returns: pygeoapi.provider.base.BaseProvider
         """
@@ -91,6 +98,28 @@ def __init__(self, provider_def):
         self.crs = None
         self.num_bands = None
 
+        # Dependencies support
+        self._context = context
+        if context and context.logger:
+            self._logger = context.logger
+        else:
+            self._logger = LOGGER  # Global fallback
+
+        if context and context.locales:
+            self._locales = context.locales
+        else:
+            self._locales = []
+
+    @property
+    def logger(self):
+        """Get logger (injected or global)"""
+        return self._logger
+
+    @property
+    def locales(self):
+        """Get supported locales (injected or global)"""
+        return self._locales
+
     def get_fields(self):
         """
         Get provider field information (names, types)