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.

Attributes

SUPABASE_AVAILABLE
T
logger

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

create(name: str, *args, **kwargs) → T | None

Create an instance of a component.

Parameters:

• name – 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

discover_components(search_paths: list[str] | None = None) → None

Discover components by scanning package paths.

Parameters:

search_paths – Optional list of package paths to search

get(name: str, load_if_missing: bool = True) → type[T] | None

Get a component class by name.

Parameters:

• name – Name of the component to retrieve

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

Returns:

Component class if found, None otherwise

get_default_search_paths() → list[str]

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

get_item_type() → str

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

get_metadata(name: str) → dict[str, Any]

Get metadata for a registry item.

Parameters:

name – Name of the item

Returns:

Dictionary of metadata if found, empty dict otherwise

list_items(item_type: str | None = None) → list[dict[str, Any]]

List all registered items, optionally filtered by type.

Parameters:

item_type – Optional type filter

Returns:

List of item metadata dictionaries

list_names(item_type: str | None = None) → list[str]

List all registered item names, optionally filtered by type.

Parameters:

item_type – Optional type filter

Returns:

List of item names

register(name: str | None = None, item_type: str = 'component', **metadata) → collections.abc.Callable[[type[T]], type[T]]

Register a component in the registry.

Parameters:

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

• item_type – 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

Examples

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

…

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

Bases: pydantic.BaseModel

Base model for registry items with metadata.

class_name: str

class_ref: Any | None = None

description: str = ''

item_type: str = 'component'

metadata: dict[str, Any] = None

module_path: str

name: str

timestamp: datetime.datetime = None

dataflow.registry.base.create(registry: Registry, name: str, *args, **kwargs)

Create a component instance from a registry.

dataflow.registry.base.get(registry: Registry, name: str)

Get a component from a registry.

dataflow.registry.base.register(registry: Registry, name: str | None = None, **kwargs)

Register a component in a registry.

dataflow.registry.base.SUPABASE_AVAILABLE = True

dataflow.registry.base.T

dataflow.registry.base.logger