Registry

Workflow and module registries. Workflows self-register at import time using WORKFLOW_REGISTRY.

Component registry for the thesis framework.

Provides a registry system for modules, processors, and other components to enable dynamic loading and plugin architecture.

class thesis.core.registry.WorkflowEntry

Bases: object

Metadata and factory for a registered workflow.

Variables:

• name – Short identifier used on the CLI (e.g. "hcp").

• factory – Callable that accepts (config, context) and returns a Nipype Workflow.

• verifier – Optional callable (config, context) -> List[str] that performs pre-run requirement checks. An empty list means all clear.

• description – Human-readable description shown by thesis list-workflows.

• default_protocol – Protocol name used when neither the CLI --protocol flag nor a per-patient config supplies one.

• default_config – Config name to use when no -c flag is provided (None falls back to the CLI default, "default").

Parameters:

• name (str)

• factory (Callable[[PipelineConfig, ProcessingContext], object])

• verifier (Optional[Callable[[PipelineConfig, ProcessingContext], list[str]]])

• description (str)

• default_protocol (Optional[str])

• default_config (Optional[str])

• is_cohort_level (bool)

name: str

factory: Callable[[PipelineConfig, ProcessingContext], object]

verifier: Callable[[PipelineConfig, ProcessingContext], list[str]] | None = None

description: str = ''

default_protocol: str | None = None

default_config: str | None = None

is_cohort_level: bool = False

__init__(name, factory, verifier=None, description='', default_protocol=None, default_config=None, is_cohort_level=False)

Parameters:

• name (str)

• factory (Callable[[PipelineConfig, ProcessingContext], object])

• verifier (Optional[Callable[[PipelineConfig, ProcessingContext], list[str]]])

• description (str)

• default_protocol (Optional[str])

• default_config (Optional[str])

• is_cohort_level (bool)

Return type:

None

class thesis.core.registry.WorkflowRegistry

Bases: object

Registry for WorkflowEntry objects.

Each workflow self-registers at module import time. The recommended path is the workflow() decorator, which builds a WorkflowEntry and calls register() for you:

from thesis.core.decorators import workflow

@workflow(name="hcp", description="HCP tractography workflow.", protocol="hcp")
def build_workflow(*, config, context):
    ...

Direct registration via register() is still supported for tests or introspection tooling:

>>> WORKFLOW_REGISTRY.register(WorkflowEntry(
...     name="hcp",
...     factory=build_workflow,
...     description="HCP tractography workflow.",
...     default_protocol="hcp",
... ))
>>> entry = WORKFLOW_REGISTRY.get("hcp")

__init__()

Return type:

None

register(entry)

Register a workflow entry.

Parameters:

entry (WorkflowEntry) – The WorkflowEntry to register.

Return type:

None

get(name)

Return the entry for name.

Parameters:

name (str) – Registered workflow name.

Return type:

WorkflowEntry

Returns:

The matching WorkflowEntry.

Raises:

KeyError – If name is not registered, with a helpful message listing available workflows.

has(name)

Return True if name is registered.

Parameters:

name (str)

Return type:

bool

list()

Return sorted list of registered workflow names.

Return type:

List[str]

all_entries()

Return all entries sorted by name.

Return type:

List[WorkflowEntry]