Registry Module¶

Dynamic component loading and registration.

ComponentRegistry¶

Registry for managing and instantiating components.

ComponentRegistry ¶

ComponentRegistry()

Registry for component classes and instances.

The registry allows: - Registering component classes by name - Creating instances from type paths - Validating instances match declared types

Example

registry = ComponentRegistry()

# Register a class
registry.register_class("my_component", MyComponent)

# Create instance from registered class
instance = registry.create("my_component", "instance_name")

# Or create from type path
instance = registry.create_from_path(
    "myapp.components.MyComponent",
    "instance_name"
)

Initialize an empty registry.

register_class ¶

register_class(name: str, component_class: type[BaseComponent]) -> None

Register a component class.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name to register the class under	required
`component_class`	`type[BaseComponent]`	The component class to register	required

Raises:

Type	Description
`ConfigurationError`	If name already registered or class invalid

get_class ¶

get_class(name: str) -> Optional[type[BaseComponent]]

Get a registered component class by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the registered class	required

Returns:

Type	Description
`Optional[type[BaseComponent]]`	The component class, or None if not found

create ¶

create(name: str, instance_name: str) -> BaseComponent

Create an instance from a registered class.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the registered class	required
`instance_name`	`str`	Name for the new instance	required

Returns:

Type	Description
`BaseComponent`	New component instance

Raises:

Type	Description
`ConfigurationError`	If class not registered

create_from_path ¶

create_from_path(type_path: str, instance_name: str) -> BaseComponent

Create an instance from a type path.

Parameters:

Name	Type	Description	Default
`type_path`	`str`	Fully qualified class path	required
`instance_name`	`str`	Name for the new instance	required

Returns:

Type	Description
`BaseComponent`	New component instance

list_registered ¶

list_registered() -> list[str]

List all registered class names.

Returns:

Type	Description
`list[str]`	List of registered class names

Functions¶

load_component_class¶

Load a component class from its fully qualified path.

load_component_class ¶

load_component_class(type_path: str) -> type[BaseComponent]

Load a component class from its type path.

Parameters:

Name	Type	Description	Default
`type_path`	`str`	Fully qualified class path (e.g., "myapp.components.MyComponent")	required

Returns:

Type	Description
`type[BaseComponent]`	The component class

Raises:

Type	Description
`ConfigurationError`	If the path is invalid, module not found, class not found, or class is not a BaseComponent

validate_component_type¶

Validate that a component instance matches its declared type.

validate_component_type ¶

validate_component_type(component: BaseComponent, expected_type_path: str) -> Optional[str]

Validate that a component instance matches its declared type path.

Parameters:

Name	Type	Description	Default
`component`	`BaseComponent`	The component instance to validate	required
`expected_type_path`	`str`	The declared type path from config	required

Returns:

Type	Description
`Optional[str]`	Error message if validation fails, None if valid

Usage Examples¶

Basic Registry Usage¶

from flowengine import ComponentRegistry, BaseComponent

class MyComponent(BaseComponent):
    def process(self, context):
        return context

# Create registry and register class
registry = ComponentRegistry()
registry.register_class("my_component", MyComponent)

# Create instance
component = registry.create("my_component", "instance_name")

Auto-Loading from Type Path¶

from flowengine import ComponentRegistry

registry = ComponentRegistry()

# Load from fully qualified path
component = registry.create_from_path(
    "myapp.components.ProcessorComponent",
    "processor"
)

Using with FlowEngine¶

from flowengine import ConfigLoader, FlowEngine, ComponentRegistry

config = ConfigLoader.load("flow.yaml")

# Option 1: Auto-instantiate all components
engine = FlowEngine.from_config(config)

# Option 2: Use custom registry
registry = ComponentRegistry()
registry.register_class("processor", CustomProcessor)

engine = FlowEngine.from_config(config, registry=registry)

Type Validation¶

from flowengine import load_component_class, validate_component_type

# Load class from path
component_class = load_component_class("myapp.MyComponent")
component = component_class("my_instance")

# Validate type matches
error = validate_component_type(
    component,
    "myapp.MyComponent"
)

if error:
    print(f"Type mismatch: {error}")

Listing Registered Components¶

registry = ComponentRegistry()
registry.register_class("fetcher", FetchComponent)
registry.register_class("processor", ProcessComponent)
registry.register_class("saver", SaveComponent)

registered = registry.list_registered()
print(registered)  # ["fetcher", "processor", "saver"]