feat: add meta data to pipeline.

.github/workflows/main.yml

@@ -11,7 +11,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"] # "3.7", "3.8", "
+        python-version: ["3.12", "3.13"]
 
     steps:
       - uses: actions/checkout@v3

.gitmodules

@@ -1,3 +0,0 @@
-[submodule "smo-database"]
-	path = smo-database
-	url = https://github.com/Leibniz-HBI/smo-database.git

dabapush/Configuration/PlugInConfiguration.py

@@ -1,9 +1,16 @@
+"""PlugInConfiguration module.
+"""
+
 import abc
-from uuid import  uuid4
+from uuid import uuid4
+
 import yaml
 
+# pylint: disable=W0622
+
+
 class PlugInConfiguration(yaml.YAMLObject):
-    """ """
+    """Abstract Base class for all PlugInConfigurations."""
 
     yaml_tag = "!dabapush:PluginConfiguration"
 
@@ -14,9 +21,6 @@ def __init__(self, name: str, id: str or None) -> None:
         self.id = id if id is not None else str(uuid4())
 
     @classmethod
-    @abc.abstractclassmethod
-    def get_instance(self) -> object or None:
-        """Get a configured instance of the appropriate reader or writer plugin.
-
-        """
-
+    @abc.abstractmethod
+    def get_instance(cls) -> object or None:
+        """Get a configured instance of the appropriate reader or writer plugin."""

dabapush/Configuration/ProjectConfiguration.py

@@ -1,13 +1,14 @@
 """ProjectConfiguration hold all information regarding jobs for a single project."""
+
 # pylint: disable=W0622
 from typing import Dict, List, Optional
 
 import yaml
 from loguru import logger as log
 
-from .ReaderConfiguration import ReaderConfiguration
-from .Registry import Registry
-from .WriterConfiguration import WriterConfiguration
+from dabapush.Configuration.ReaderConfiguration import ReaderConfiguration
+from dabapush.Configuration.Registry import get_reader, get_writer
+from dabapush.Configuration.WriterConfiguration import WriterConfiguration
 
 
 class ProjectConfiguration(yaml.YAMLObject):
@@ -36,23 +37,23 @@ def __init__(
         """Initialize a ProjectConfiguration with optional reader and/or writer dicts"""
         super().__init__()
 
-        # store readers if they are passed into the constructor or else intialize
+        # store readers if they are passed into the constructor or else initialize
         # new list via default arg
         self.readers: Dict[str, ReaderConfiguration] = readers or {}
-        # store writers if they are passed into the constructor or else intialize
+        # store writers if they are passed into the constructor or else initialize
         # new list via default arg
         self.writers: Dict[str, WriterConfiguration] = writers or {}
 
         # initialize project metadata
         self.author = author
         self.name = name
 
-    def add_reader(self, type: str, name: str) -> None:
+    def add_reader(self, kind: str, name: str) -> None:
         """add a reader configuration to the project
 
         Parameters
         ----------
-        type :
+        kind :
             str: registry of the configuration to add
         name :
             str: name of the configuration to add
@@ -68,12 +69,12 @@ def add_reader(self, type: str, name: str) -> None:
 
         """
         # get constructor from registry
-        pinst = Registry.get_reader(type)
-        if pinst is not None:
-            self.readers[name] = pinst(name)
+        configuration_constructor = get_reader(kind)
+        if configuration_constructor is not None:
+            self.readers[name] = configuration_constructor(name)
             log.debug(f'Currently configured readers: {",".join(list(self.readers))}')
         else:
-            raise Exception(f"{type} not found")
+            raise ValueError(f"{kind} not found")
 
     def remove_reader(self, name: str) -> None:
         """Remove a reader from the configuration.
@@ -101,7 +102,7 @@ def list_readers(self) -> List[ReaderConfiguration]:
         # copy stuff
         return list(self.readers.values())
 
-    def add_writer(self, type: str, name: str) -> None:
+    def add_writer(self, kind: str, name: str) -> None:
         """Adds a writer to the configuration.
 
         Parameters
@@ -112,11 +113,11 @@ def add_writer(self, type: str, name: str) -> None:
             str: name of the added writer
         """
         # get constructor from registry
-        pinst = Registry.get_writer(type)
-        if pinst is not None:
-            self.writers[name] = pinst(name)
+        configuration_constructor = get_writer(kind)
+        if configuration_constructor is not None:
+            self.writers[name] = configuration_constructor(name)
         else:
-            raise Exception(f"{type} not found")
+            raise ValueError(f"{kind} not found")
 
     def remove_writer(self, name: str):
         """Removes the specified writer from the configuration.
@@ -141,8 +142,3 @@ def set_name(self, name):
     def set_author(self, author):
         """Sets the project's authors."""
         self.author = author
-
-    @property
-    def __configuration__(self):
-        registry = Registry()
-        return registry

dabapush/Configuration/ReaderConfiguration.py

@@ -1,9 +1,15 @@
-from pathlib import Path
+"""Reader Interface.
+"""
+
+import abc
+
 from .PlugInConfiguration import PlugInConfiguration
 
+# pylint: disable=W0622
+
 
 class ReaderConfiguration(PlugInConfiguration):
-    """ """
+    """Abstract Base class for all ReaderConfigurations."""
 
     yaml_tag = "!dabapush:ReaderConfiguration"
 
@@ -12,5 +18,7 @@ def __init__(self, name, id, read_path: str or None, pattern: str or None) -> No
         self.read_path = read_path if read_path is not None else "."
         self.pattern = pattern if pattern is not None else "*.json"
 
-    def __repr__(self) -> str:
-        return super().__repr__()
+    @classmethod
+    @abc.abstractmethod
+    def get_instance(cls) -> object or None:
+        """Get a configured instance of the appropriate reader or writer plugin."""

dabapush/Configuration/Registry.py

@@ -1,66 +1,89 @@
 """fetching plug-ins from entrypoints and helper methods"""
+
 # pylint: disable=W0622
-from importlib.metadata import entry_points
-from typing import Any, List, Optional
+import platform
+from importlib.metadata import EntryPoint
+from typing import Any, List, Optional, Type
 
 from .ReaderConfiguration import ReaderConfiguration
 from .WriterConfiguration import WriterConfiguration
 
 
-class Registry:
-    """receive plug-ins from entry point"""
-
-    readers = entry_points()["dabapush_readers"]
-    writers = entry_points()["dabapush_writers"]
-
-    # --- static methods --- #
-
-    @staticmethod
-    def get_reader(name: str) -> Optional[ReaderConfiguration]:
-        """
-        params:
-          name: str:
-            registry key to retrieve
-        returns:
-            ReaderConfiguration or None: the requested ReaderConfiguration or None if
-            no matching configuration is found.
-        """
-        candidates = [_ for _ in Registry.readers if _.name == name]
-        try:
-            return candidates[0].load()
-        except IndexError:
-            return None
-
-    @staticmethod
-    def get_writer(name: str) -> Optional[WriterConfiguration]:
-        """
-        params:
-          name: str:
-            registry key to retrieve
-        returns:
-            WriterConfiguration or None: the requested WriterConfiguration or None if
-            no matching configuration is found."""
-        candidates = [_ for _ in Registry.writers if _.name == name]
-        try:
-            return candidates[0].load()
-        except IndexError:
-            return None
-
-    @staticmethod
-    def __ensure_reader__(arg: Any) -> bool:
-        return issubclass(arg, ReaderConfiguration)
-
-    @staticmethod
-    def __ensure_writer__(arg: Any) -> bool:
-        return issubclass(arg, WriterConfiguration)
-
-    @staticmethod
-    def list_all_readers() -> List[str]:
-        """return a list of all readers"""
-        return [_.name for _ in Registry.readers]
-
-    @staticmethod
-    def list_all_writers() -> List[str]:
-        """return a list of all writers"""
-
-        return [_.name for _ in Registry.writers]
+def _get_entrypoint_py38(group: str) -> List[EntryPoint]:
+    from importlib.metadata import entry_points  # pylint: disable=C0415
+
+    return entry_points()[group]
+
+
+def _get_entrypoint_py39(group: str) -> List[EntryPoint]:
+    from importlib.metadata import entry_points  # pylint: disable=C0415
+
+    return list(entry_points().select(group=group))
+
+
+py_major, py_minor, _ = platform.python_version_tuple()
+
+if int(py_major) == 3 and int(py_minor) == 8:
+    _get_entrypoint = _get_entrypoint_py38
+else:
+    _get_entrypoint = _get_entrypoint_py39
+
+
+def readers() -> List[EntryPoint]:
+    """fetch readers from entry point"""
+    return _get_entrypoint("dabapush_readers")
+
+
+def writers() -> List[EntryPoint]:
+    """fetch writers from entry point"""
+    return _get_entrypoint("dabapush_writers")
+
+
+def get_reader(name: str) -> Optional[Type[ReaderConfiguration]]:
+    """
+    params:
+      name: str:
+        registry key to retrieve
+    returns:
+        ReaderConfiguration or None: the requested ReaderConfiguration or None if
+        no matching configuration is found.
+    """
+    candidates = [_ for _ in readers() if _.name == name]
+    try:
+        return candidates[0].load()
+    except IndexError:
+        return None
+
+
+def get_writer(name: str) -> Optional[Type[WriterConfiguration]]:
+    """
+    params:
+      name: str:
+        registry key to retrieve
+    returns:
+        WriterConfiguration or None: the requested WriterConfiguration or None if
+        no matching configuration is found."""
+    candidates = [_ for _ in writers() if _.name == name]
+    try:
+        return candidates[0].load()
+    except IndexError:
+        return None
+
+
+def __ensure_reader__(arg: Any) -> bool:
+    return issubclass(arg, ReaderConfiguration)
+
+
+def __ensure_writer__(arg: Any) -> bool:
+    return issubclass(arg, WriterConfiguration)
+
+
+def list_all_readers() -> List[str]:
+    """return a list of all readers"""
+    return [_.name for _ in readers()]
+
+
+def list_all_writers() -> List[str]:
+    """return a list of all writers"""
+
+    return [_.name for _ in writers()]