dataflow.registry.providers.base

Base provider class for the Haive Registry System.

This module defines the base provider class that all specific entity providers inherit from. Entity providers are responsible for discovering, registering, and managing specific types of entities in the registry system.

Each entity type (agent, tool, engine, etc.) has its own provider that implements the discovery and registration logic specific to that entity type. The base provider class defines the common interface and functionality shared by all providers.

Classes:

EntityProvider: Abstract base class for all entity providers

Example

Implementing a custom entity provider:

>>> from haive.dataflow.registry.providers.base import EntityProvider
>>> from haive.dataflow.registry.models import EntityType
>>>
>>> class CustomProvider(EntityProvider):
...     def __init__(self):
...         super().__init__(EntityType.CUSTOM)
...
...     def discover(self, module_paths=None):
...         # Custom discovery logic
...         paths = module_paths or self.get_default_search_paths()
...         # ... discovery implementation ...
...         return registered_ids
...
...     def get_default_search_paths(self):
...         return ["my_package.custom_components"]

Classes

EntityProvider

Abstract base class for entity providers.

Module Contents

class dataflow.registry.providers.base.EntityProvider(entity_type)

Bases: abc.ABC

Abstract base class for entity providers.

Entity providers are responsible for discovering, registering, and managing specific types of entities in the registry system. This base class defines the common interface and shared functionality that all entity providers must implement.

Entity providers handle: - Discovering components of a specific type in the codebase - Registering discovered components in the registry system - Managing component metadata, configurations, and dependencies

entity_type

The type of entity this provider handles

Type:

EntityType

Initialize the entity provider.

Parameters:

entity_type (haive.dataflow.models.EntityType) – Type of entity this provider handles (e.g., AGENT, TOOL)

add_configuration(registry_id, config_type, config_data)

Add a configuration to a registry entity.

Parameters:

• registry_id (str) – Registry entity ID

• config_type (haive.dataflow.models.ConfigType) – Type of configuration

• config_data (Any) – Configuration data

Return type:

None

add_dependency(registry_id, dependent_id, dependency_type)

Add a dependency between registry entities.

Parameters:

• registry_id (str) – ID of the entity that depends on another

• dependent_id (str) – ID of the entity being depended on

• dependency_type (haive.dataflow.models.DependencyType) – Type of dependency

Return type:

None

add_environment_vars(registry_id, env_vars)

Add environment variables to a registry entity.

Parameters:

• registry_id (str) – Registry entity ID

• env_vars (dict[str, bool]) – Dictionary mapping environment variable names to required flag

Return type:

None

add_import_log(import_session, entity_name, status, message=None, traceback_str=None)

Add an import log entry.

Parameters:

• import_session (str) – Import session identifier

• entity_name (str) – Name of the entity being imported

• status (haive.dataflow.models.ImportStatus) – Import status

• message (str | None) – Optional message

• traceback_str (str | None) – Optional traceback string

Return type:

None

abstractmethod discover(module_paths=None)

Discover and register entities.

This abstract method must be implemented by concrete provider classes. It should search for components of the provider’s entity type in the specified module paths, and register them in the registry system.

Parameters:

module_paths (list[str] | None) – Optional list of module paths to search. If None, the provider’s default search paths will be used.

Returns:

List of registered entity IDs for the discovered components

Return type:

List[str]

Raises:

NotImplementedError – Must be implemented by subclasses

discover_modules(base_path)

Discover all Python modules under a base path.

This helper method recursively explores a package to find all Python modules. It handles both regular modules and packages, traversing the entire module hierarchy to discover all available modules.

Parameters:

base_path (str) – Base module path to start discovery from (e.g., “haive.agents”)

Returns:

List of fully qualified module paths discovered

Return type:

List[str]

Example

>>> provider = SomeEntityProvider()
>>> modules = provider.discover_modules("haive.tools")
>>> print(f"Discovered modules: {modules}")
Discovered modules: ['haive.tools.text', 'haive.tools.image', ...]

abstractmethod get_default_search_paths()

Get default search paths for entity discovery.

This abstract method must be implemented by concrete provider classes. It should return a list of module paths where components of the provider’s entity type are likely to be found.

Returns:

List of package paths to search for components

Return type:

List[str]

Raises:

NotImplementedError – Must be implemented by subclasses

Example

>>> def get_default_search_paths(self):
...     return [
...         "haive.agents",
...         "haive.core.agents",
...         "my_package.custom_agents"
...     ]

is_pydantic_model(obj)

Check if an object is a Pydantic model.

Parameters:

obj (Any) – Object to check

Returns:

True if it’s a Pydantic model, False otherwise

Return type:

bool