dataflow.registry.base¶

Base registry system for Haive components.

This module provides the fundamental registry system that all specific registries inherit from. It handles registration, discovery, database persistence, and retrieval of components.

Classes¶

Registry

Base class for component registries.

RegistryItem

Base model for registry items with metadata.

Functions¶

create(registry, name, *args, **kwargs)

Create a component instance from a registry.

get(registry, name)

Get a component from a registry.

register(registry[, name])

Register a component in a registry.

Module Contents¶

class dataflow.registry.base.Registry¶

Bases: Generic[T]

Base class for component registries.

This class provides the foundation for registering and retrieving components in a type-safe way with added features like: - Automatic component discovery - Metadata tracking - Supabase integration - Caching

entries¶

Registry entries

Type:

Dict[str, RegistryItem]

_instance¶

Singleton instance

Type:

Registry

_discovered¶

Whether auto-discovery has run

Type:

bool

_supabase_client¶

Supabase client for persistence

Singleton pattern to ensure only one registry exists.

create(name, *args, **kwargs)¶

Create an instance of a component.

Parameters:

  • name (str) – Name of the component to create

  • *args – Positional arguments for the constructor

  • **kwargs – Keyword arguments for the constructor

Returns:

Instance of the component if found, None otherwise

Return type:

T | None

discover_components(search_paths=None)¶

Discover components by scanning package paths.

Parameters:

search_paths (list[str] | None) – Optional list of package paths to search

Return type:

None

get(name, load_if_missing=True)¶

Get a component class by name.

Parameters:

  • name (str) – Name of the component to retrieve

  • load_if_missing (bool) – Whether to try loading the class if not already loaded

Returns:

Component class if found, None otherwise

Return type:

type[T] | None

get_default_search_paths()¶

Get default search paths for component discovery.

This method should be overridden by subclasses to provide specific search paths for their component types.

Returns:

List of package paths to search

Return type:

list[str]

get_item_type()¶

Get the default item type for this registry.

This method should be overridden by subclasses to provide the default item type for their registry.

Returns:

String representing the default item type

Return type:

str

get_metadata(name)¶

Get metadata for a registry item.

Parameters:

name (str) – Name of the item

Returns:

Dictionary of metadata if found, empty dict otherwise

Return type:

dict[str, Any]

list_items(item_type=None)¶

List all registered items, optionally filtered by type.

Parameters:

item_type (str | None) – Optional type filter

Returns:

List of item metadata dictionaries

Return type:

list[dict[str, Any]]

list_names(item_type=None)¶

List all registered item names, optionally filtered by type.

Parameters:

item_type (str | None) – Optional type filter

Returns:

List of item names

Return type:

list[str]

register(name=None, item_type='component', **metadata)¶

Register a component in the registry.

Parameters:

  • name (str | None) – Optional custom name (defaults to class name in lowercase)

  • item_type (str) – Type of item (e.g., “game”, “agent”, “component”, “tool”, “toolkit”)

  • **metadata – Additional metadata to store with the registration

Returns:

Decorator function that registers the component

Return type:

collections.abc.Callable[[type[T]], type[T]]

Examples

@registry.register(name=”custom_name”, author=”John”) class MyComponent:

…

class dataflow.registry.base.RegistryItem(/, **data)¶

Bases: pydantic.BaseModel

Base model for registry items with metadata.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

dataflow.registry.base.create(registry, name, *args, **kwargs)¶

Create a component instance from a registry.

Parameters:
dataflow.registry.base.get(registry, name)¶

Get a component from a registry.

Parameters:
dataflow.registry.base.register(registry, name=None, **kwargs)¶

Register a component in a registry.

Parameters: