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"]

Attributes¶

logger

Classes¶

EntityProvider

Abstract base class for entity providers.

Module Contents¶

class dataflow.registry.providers.base.EntityProvider(entity_type: haive.dataflow.models.EntityType)¶

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

add_configuration(registry_id: str, config_type: haive.dataflow.models.ConfigType, config_data: Any) → None¶

Add a configuration to a registry entity.

Parameters:

registry_id – Registry entity ID
config_type – Type of configuration
config_data – Configuration data

add_dependency(registry_id: str, dependent_id: str, dependency_type: haive.dataflow.models.DependencyType) → None¶

Add a dependency between registry entities.

Parameters:

registry_id – ID of the entity that depends on another
dependent_id – ID of the entity being depended on
dependency_type – Type of dependency

add_environment_vars(registry_id: str, env_vars: dict[str, bool]) → None¶

Add environment variables to a registry entity.

Parameters:

registry_id – Registry entity ID
env_vars – Dictionary mapping environment variable names to required flag

add_import_log(import_session: str, entity_name: str, status: haive.dataflow.models.ImportStatus, message: str | None = None, traceback_str: str | None = None) → None¶

Add an import log entry.

Parameters:

import_session – Import session identifier
entity_name – Name of the entity being imported
status – Import status
message – Optional message
traceback_str – Optional traceback string

abstractmethod discover(module_paths: list[str] | None = None) → list[str]¶

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 – 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: str) → list[str]¶

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 – 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() → list[str]¶

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: Any) → bool¶

Check if an object is a Pydantic model.

Parameters:: obj – Object to check
Returns:: True if it’s a Pydantic model, False otherwise

entity_type¶

dataflow.registry.providers.base.logger¶