Plugins & Registry

The plugin registry lets external packages register concrete system implementations so the CLI can discover and instantiate them by name — without any changes to the aptdata core.

---

How it works

sequenceDiagram
    participant Pkg as Your adapter package
    participant Reg as aptdata.plugins.registry
    participant CLI as aptdata CLI

    Pkg->>Reg: registry.register("my_system", MySystem)
    CLI->>Reg: registry.get("my_system")
    Reg-->>CLI: MySystem class
    CLI->>CLI: aptdata run my_system

---

Registering a system

# my_package/systems.py
from pydantic.dataclasses import dataclass as pydantic_dataclass
from aptdata.core import BaseSystem, IFlow
from aptdata.plugins import registry


@pydantic_dataclass
class SalesSystem(BaseSystem):
    def __post_init__(self) -> None:
        self._flows: list[IFlow] = []

    def register_flow(self, flow: IFlow) -> None:
        self._flows.append(flow)

    def run(self) -> None:
        for flow in self._flows:
            flow.run([])


# Register at import time so the CLI can find it
registry.register("sales_system", SalesSystem)

Then run it:

aptdata run sales_system --env prod

---

Auto-discovery with entry points

You can auto-register systems when your package is installed by declaring a aptdata.systems entry-point group in your pyproject.toml:

[tool.poetry.plugins."aptdata.systems"]
sales_system = "my_package.systems:SalesSystem"

Note

Built-in entry-point auto-discovery is planned for a future release. For now, call registry.register() explicitly at import time.

---

_SystemRegistry API

aptdata.plugins._SystemRegistry

Simple name → system-class mapping.

Source code in aptdata/plugins/__init__.py

class _SystemRegistry:
    """Simple name → system-class mapping."""

    def __init__(self) -> None:
        self._store: dict[str, type[ISystem]] = {}

    def register(self, name: str, system_cls: type[ISystem]) -> None:
        """Register *system_cls* under *name*.

        Parameters
        ----------
        name:
            Unique identifier used on the CLI (e.g. ``"pipeline_x"``).
        system_cls:
            A concrete subclass of :class:`~aptdata.core.system.ISystem`.
        """
        self._store[name] = system_cls

    def get(self, name: str) -> type[ISystem] | None:
        """Return the system class registered under *name*, or ``None``."""
        return self._store.get(name)

    def list_systems(self) -> list[str]:
        """Return a sorted list of all registered system names."""
        return sorted(self._store)

Functions

register(name, system_cls)

Register system_cls under name.

Parameters:

Name	Type	Description	Default
name	str	Unique identifier used on the CLI (e.g. "pipeline_x").	required
system_cls	type[ISystem]	A concrete subclass of :class:~aptdata.core.system.ISystem.	required

Source code in aptdata/plugins/__init__.py

def register(self, name: str, system_cls: type[ISystem]) -> None:
    """Register *system_cls* under *name*.

    Parameters
    ----------
    name:
        Unique identifier used on the CLI (e.g. ``"pipeline_x"``).
    system_cls:
        A concrete subclass of :class:`~aptdata.core.system.ISystem`.
    """
    self._store[name] = system_cls

get(name)

Return the system class registered under name, or None.

Source code in aptdata/plugins/__init__.py

def get(self, name: str) -> type[ISystem] | None:
    """Return the system class registered under *name*, or ``None``."""
    return self._store.get(name)

list_systems()

Return a sorted list of all registered system names.

Source code in aptdata/plugins/__init__.py

def list_systems(self) -> list[str]:
    """Return a sorted list of all registered system names."""
    return sorted(self._store)

---

Global singleton

aptdata.plugins.registry = _SystemRegistry() module-attribute