process_registry

Registry for mapping process IDs to concrete Process classes.

ProcessRegistry

Central mapping of process_id to Process class.

register classmethod

register(process_cls: type[Process]) -> type[Process]

Register a Process class by its process_id.

Source code in cltk/core/process_registry.py

@classmethod
def register(cls, process_cls: type["Process"]) -> type["Process"]:
    """Register a Process class by its ``process_id``."""
    process_id = getattr(process_cls, "process_id", None)
    if not process_id or not isinstance(process_id, str):
        raise ValueError(
            f"Process class {process_cls.__name__} must define process_id."
        )
    existing = cls._registry.get(process_id)
    if existing and existing is not process_cls:
        raise ValueError(
            f"process_id '{process_id}' already registered for {existing.__name__}."
        )
    cls._registry[process_id] = process_cls
    logger.debug("Registered process_id '%s' -> %s", process_id, process_cls)
    return process_cls

get_process classmethod

get_process(process_id: str) -> type[Process]

Return the registered Process class for process_id.

Source code in cltk/core/process_registry.py

@classmethod
def get_process(cls, process_id: str) -> type["Process"]:
    """Return the registered Process class for ``process_id``."""
    cls._ensure_defaults_loaded()
    try:
        return cls._registry[process_id]
    except KeyError as exc:
        available = ", ".join(sorted(cls._registry))
        raise KeyError(
            f"Unknown process_id '{process_id}'. Available: {available}"
        ) from exc

list_processes classmethod

list_processes() -> dict[str, type[Process]]

Return a snapshot of registered processes.

Source code in cltk/core/process_registry.py

@classmethod
def list_processes(cls) -> dict[str, type["Process"]]:
    """Return a snapshot of registered processes."""
    cls._ensure_defaults_loaded()
    return dict(cls._registry)

register_process

register_process(
    process_cls: type[Process],
) -> type[Process]

Decorate a Process class to register in the ProcessRegistry.

Source code in cltk/core/process_registry.py

def register_process(process_cls: type["Process"]) -> type["Process"]:
    """Decorate a Process class to register in the ProcessRegistry."""
    return ProcessRegistry.register(process_cls)