add functionality to load plugins dynamically

Author: pvgenuchten

pygeometa/schemas/__init__.py

@@ -46,37 +46,97 @@
 import importlib
 import logging
 import os
+import pkgutil
+from typing import Dict, Type, List
 
 from pygeometa.schemas.base import BaseOutputSchema
 
 LOGGER = logging.getLogger(__name__)
 THISDIR = os.path.dirname(os.path.realpath(__file__))
 
-SCHEMAS = {
-    'dcat': 'pygeometa.schemas.dcat.DCATOutputSchema',
-    'iso19139': 'pygeometa.schemas.iso19139.ISO19139OutputSchema',
-    'iso19139-2': 'pygeometa.schemas.iso19139_2.ISO19139_2OutputSchema',
-    'iso19139-hnap': 'pygeometa.schemas.iso19139_hnap.ISO19139HNAPOutputSchema',  # noqa
-    'oarec-record': 'pygeometa.schemas.ogcapi_records.OGCAPIRecordOutputSchema',  # noqa
-    'schema-org': 'pygeometa.schemas.schema_org.SchemaOrgOutputSchema',
-    'stac-item': 'pygeometa.schemas.stac.STACItemOutputSchema',
-    'wmo-cmp': 'pygeometa.schemas.wmo_cmp.WMOCMPOutputSchema',
-    'wmo-wcmp2': 'pygeometa.schemas.wmo_wcmp2.WMOWCMP2OutputSchema',
-    'wmo-wigos': 'pygeometa.schemas.wmo_wigos.WMOWIGOSOutputSchema',
-    'cwl': 'pygeometa.schemas.cwl.CWLOutputSchema'
-}
+# runtime mapping: schema_key -> class object
+_DISCOVERED_SCHEMAS: Dict[str, Type[BaseOutputSchema]] = {}
+_DISCOVERY_DONE = False
 
 
+def _discover_schemas():
+    """Discover local schema packages (folders with __init__.py) in ./schemas.
+
+    For each discovered package module, import it and choose the first class
+    that is a subclass of BaseOutputSchema (excluding BaseOutputSchema itself).
+    The mapping key will be:
+      - module attribute SCHEMA_NAME if defined, else
+      - the package (folder) name.
+
+    This function is idempotent and caches results in _DISCOVERED_SCHEMAS.
+    """
+    global _DISCOVERY_DONE, _DISCOVERED_SCHEMAS
+    if _DISCOVERY_DONE:
+        return
+
+    LOGGER.debug("Discovering schema packages in %s", THISDIR)
+    pkg_name = __name__  # 'pygeometa.schemas'
+    # Use pkgutil to iterate subpackages defined by filesystem packages adjacent to this file.
+    # pkgutil.walk_packages won't work directly without package loader, so we handle dirs.
+    try:
+        entries = sorted(os.listdir(THISDIR))
+    except OSError:
+        entries = []
+
+    for entry in entries:
+        path = os.path.join(THISDIR, entry)
+        # look for directories with __init__.py only
+        if not os.path.isdir(path):
+            continue
+        init_py = os.path.join(path, "__init__.py")
+        if not os.path.isfile(init_py):
+            continue
+
+        module_name = f"{pkg_name}.{entry}"
+        try:
+            module = importlib.import_module(module_name)
+        except Exception as exc:  # defensive: log and skip bad packages
+            LOGGER.exception("Failed to import schema package %s: %s", module_name, exc)
+            continue
+
+        # allow module to explicitly provide a name
+        schema_key = getattr(module, "name", entry)
+
+        # find first class that subclasses BaseOutputSchema
+        found_cls = None
+        for attr_name in dir(module):
+            try:
+                attr = getattr(module, attr_name)
+            except Exception:
+                continue
+            if isinstance(attr, type) and issubclass(attr, BaseOutputSchema) and attr is not BaseOutputSchema:
+                found_cls = attr
+                break
+
+        if found_cls is None:
+            LOGGER.warning("No BaseOutputSchema subclass found in %s; skipping", module_name)
+            continue
+
+        if schema_key in _DISCOVERED_SCHEMAS:
+            LOGGER.warning("Duplicate schema key '%s' (from %s). Keeping first discovered.", schema_key, module_name)
+            continue
+
+        _DISCOVERED_SCHEMAS[schema_key] = found_cls
+        LOGGER.info("Discovered schema '%s' -> %s.%s", schema_key, module_name, found_cls.__name__)
+
+    _DISCOVERY_DONE = True
+
 def get_supported_schemas(details: bool = False,
                           include_autodetect: bool = False) -> list:
     """
-    Get supported schemas
+    Get supported schemas.
 
     :param details: provide read/write details
     :param include_autodetect: include magic auto detection mode
 
-    :returns: list of supported schemas
+    :returns: list of supported schemas (strings) or details matrix
     """
+    _discover_schemas()
 
     def has_mode(plugin: BaseOutputSchema, mode: str) -> bool:
         enabled = False
@@ -90,36 +150,47 @@ def has_mode(plugin: BaseOutputSchema, mode: str) -> bool:
 
         return enabled
 
-    schema_matrix = []
-
     LOGGER.debug('Generating list of supported schemas')
 
     if not details:
+        schema_names = []
+        for cls in _DISCOVERED_SCHEMAS.values():
+            try:
+                schema_names.append(cls().name)
+            except Exception:
+                continue
         if include_autodetect:
-            schemas_keys = list(SCHEMAS.keys())
-            schemas_keys.append('autodetect')
-            return schemas_keys
-        else:
-            return SCHEMAS.keys()
+            schema_names.append("autodetect")
+        return schema_names
 
-    for key in SCHEMAS.keys():
-        schema = load_schema(key)
-        can_read = has_mode(schema, 'import_')
-        can_write = has_mode(schema, 'write')
+    schema_matrix = []
+    for key, cls in _DISCOVERED_SCHEMAS.items():
+        nm = key
+        try:
+            schema_inst = cls()
+            nm = schema_inst.name
+            can_read = has_mode(schema_inst, "import_")
+            can_write = has_mode(schema_inst, "write")
+            description = getattr(schema_inst, "description", "")
+        except Exception:
+            LOGGER.exception("Error instantiating schema class for key %s", key)
+            can_read = False
+            can_write = False
+            description = ""
 
         schema_matrix.append({
-            'id': key,
-            'description': schema.description,
-            'read': can_read,
-            'write': can_write
+            "id": nm,
+            "description": description,
+            "read": can_read,
+            "write": can_write
         })
 
     if include_autodetect:
         schema_matrix.append({
-            'id': 'autodetect',
-            'description': 'Auto schema detection',
-            'read': True,
-            'write': False
+            "id": "autodetect",
+            "description": "Auto schema detection",
+            "read": True,
+            "write": False
         })
 
     return schema_matrix
@@ -133,28 +204,25 @@ def load_schema(schema_name: str) -> BaseOutputSchema:
 
     :returns: plugin object
     """
+    _discover_schemas()
+    LOGGER.debug("Available schemas: %s", list(_DISCOVERED_SCHEMAS.keys()))
 
-    LOGGER.debug(f'Schemas: {SCHEMAS.keys()}')
+    # allow 'autodetect' to be handled elsewhere (kept for parity)
+    if schema_name == "autodetect":
+        raise InvalidSchemaError("Autodetect is not a concrete schema")
 
-    if schema_name not in SCHEMAS.keys():
-        msg = f'Schema {schema_name} not found'
+    if schema_name not in _DISCOVERED_SCHEMAS:
+        msg = f"Schema {schema_name} not found"
         LOGGER.exception(msg)
         raise InvalidSchemaError(msg)
 
-    name = SCHEMAS[schema_name]
-
-    if '.' in name:  # dotted path
-        packagename, classname = name.rsplit('.', 1)
-    else:
-        raise InvalidSchemaError(f'Schema path {name} not found')
-
-    LOGGER.debug(f'package name: {packagename}')
-    LOGGER.debug(f'class name: {classname}')
-
-    module = importlib.import_module(packagename)
-    class_ = getattr(module, classname)
-
-    return class_()
+    cls = _DISCOVERED_SCHEMAS[schema_name]
+    try:
+        return cls()
+    except Exception as exc:
+        msg = f"Failed to instantiate schema {schema_name}: {exc}"
+        LOGGER.exception(msg)
+        raise InvalidSchemaError(msg)
 
 
 class InvalidSchemaError(Exception):

pygeometa/schemas/ogcapi_records/__init__.py

@@ -47,7 +47,7 @@
 import os
 from typing import Union
 
-from pygeometa import __version__
+
 from pygeometa.core import get_charstring
 from pygeometa.helpers import generate_datetime, json_dumps
 from pygeometa.schemas.base import BaseOutputSchema
@@ -247,7 +247,8 @@ def write(self, mcf: dict, stringify: str = True) -> Union[dict, str]:
         for value in mcf['distribution'].values():
             record['links'].append(self.generate_link(value))
 
-        record['generated_by'] = f'pygeometa {__version__}'
+        from importlib.metadata import version
+        record['generated_by'] = f'pygeometa {version("pygeometa")}'
 
         if stringify:
             return json_dumps(record)