dataflow.registry.core¶

Core Registry System for Haive.

This module implements the central registry system for the Haive framework, providing functionality for registering, querying, and managing various components such as agents, tools, engines, and other entity types.

The registry system maintains both in-memory storage and database persistence through Supabase integration, allowing components to be discovered, registered, and retrieved across application sessions.

Examples

Basic usage of the registry system:

>>> from haive.dataflow.registry.core import registry_system
>>> from haive.dataflow.registry.models import EntityType
>>>
>>> # Register a new component
>>> entity_id = registry_system.register_entity(
...     name="TextSummarizer",
...     type=EntityType.TOOL,
...     description="Summarizes text documents",
...     module_path="haive.tools.summarizers",
...     class_name="TextSummarizerTool"
... )
>>>
>>> # Query for components by type
>>> tools = registry_system.get_entities_by_type(EntityType.TOOL)
>>> print(f"Found {len(tools)} registered tools")
>>>
>>> # Get a specific component by ID
>>> entity = registry_system.get_entity(entity_id)
>>> print(f"Retrieved entity: {entity.name}")

Classes¶

`ConfigType`	Types of configuration that can be associated with entities.
`DependencyType`	Types of dependencies between entities.
`EntityType`	Types of entities that can be registered.
`ImportStatus`	Status of an import operation.
`RegistrySystem`	Core registry system for managing Haive components.

Functions¶

get_registry_system()

Get the registry system instance (lazy initialization).

Module Contents¶

class dataflow.registry.core.ConfigType¶

Bases: str, enum.Enum

Types of configuration that can be associated with entities.

Initialize self. See help(type(self)) for accurate signature.

class dataflow.registry.core.DependencyType¶

Bases: str, enum.Enum

Types of dependencies between entities.

Initialize self. See help(type(self)) for accurate signature.

class dataflow.registry.core.EntityType¶

Bases: str, enum.Enum

Types of entities that can be registered.

Initialize self. See help(type(self)) for accurate signature.

class dataflow.registry.core.ImportStatus¶

Bases: str, enum.Enum

Status of an import operation.

Initialize self. See help(type(self)) for accurate signature.

class dataflow.registry.core.RegistrySystem¶

Core registry system for managing Haive components.

The registry system serves as a centralized repository for tracking and managing various components in the Haive ecosystem, such as agents, tools, engines, games, and LLM models. It provides both in-memory storage and optional database persistence via Supabase.

Components can be registered, configured, queried, and managed through the registry system, which maintains metadata, configurations, and dependency relationships between components.

_entities¶

In-memory storage for registry items

Type:: dict

_configurations¶

In-memory storage for configurations

Type:: dict

_dependencies¶

In-memory storage for dependencies

Type:: dict

_environment_vars¶

In-memory storage for environment variables

Type:: dict

_import_logs¶

List of import operation logs

Type:: list

_supabase¶: Supabase client for database persistence (if available)

Examples

>>> from haive.dataflow.registry.core import registry_system
>>> from haive.dataflow.registry.models import EntityType
>>>
>>> # Register a new component
>>> entity_id = registry_system.register_entity(
...     name="TextAnalyzer",
...     type=EntityType.TOOL,
...     description="Analyzes text content",
...     module_path="haive.tools.analyzers",
...     class_name="TextAnalyzerTool"
... )
>>>
>>> # Add configuration
>>> registry_system.add_configuration(
...     registry_id=entity_id,
...     config_type="input_schema",
...     config_data={"type": "object", "properties": {"text": {"type": "string"}}}
... )

Initialize the registry system.

add_configuration(registry_id, config_type, config_data)¶

Add a configuration to an entity.

Parameters:

registry_id (str) – ID of the registered entity
config_type (ConfigType) – Type of configuration
config_data (Any) – Configuration data

Returns:

ID of the configuration or None on failure

Return type:

str | None

add_dependency(registry_id, dependent_id, dependency_type)¶

Add a dependency between two 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 (DependencyType) – Type of dependency

Returns:

ID of the dependency or None on failure

Return type:

str | None

add_environment_var(var_name, provider_name, is_required=True, description=None)¶

Add an environment variable to the registry.

Parameters:

var_name (str) – Name of the environment variable
provider_name (str) – Provider this environment variable is for
is_required (bool) – Whether the environment variable is required
description (str | None) – Optional description

Returns:

ID of the environment variable entry or None on failure

Return type:

str | None

add_import_log(import_session, entity_name, entity_type, 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
entity_type (str) – Type of entity
status (ImportStatus) – Import status
message (str | None) – Optional message
traceback_str (str | None) – Optional traceback string

Return type:

None

check_environment_var(var_name)¶

Check if an environment variable is set.

Parameters:: var_name (str) – Name of environment variable to check
Returns:: True if the environment variable is set, False otherwise
Return type:: bool

get_available_providers(entity_type=None)¶

Get all available providers.

Parameters:: entity_type (EntityType | None) – Optional entity type to filter providers by (e.g., LLM_PROVIDER)
Returns:: List of provider data with availability info
Return type:: list[dict[str, Any]]

get_entities_by_type(entity_type)¶

Get all entities of a specific type.

Parameters:: entity_type (EntityType) – Type of entities to retrieve
Returns:: List of entity data
Return type:: list[dict[str, Any]]

get_entity(entity_id)¶

Get an entity by ID.

Parameters:: entity_id (str) – ID of the entity
Returns:: Entity data or None if not found
Return type:: dict[str, Any] | None

get_environment_vars(provider_name=None)¶

Get environment variables, optionally filtered by provider.

Parameters:: provider_name (str | None) – Optional provider to filter by
Returns:: List of environment variable data
Return type:: list[dict[str, Any]]

register_entity(name, entity_type, description=None, metadata=None)¶

Parameters:

name (str) – Name of the entity
entity_type (EntityType) – Type of entity
description (str | None) – Optional description
metadata (dict[str, Any] | None) – Optional metadata dictionary

Returns:

ID of the registered entity

Return type:

str

search_entities(query, entity_type=None, metadata_filter=None)¶

Search for entities based on a query.

Parameters:

query (str) – Search query
entity_type (EntityType | None) – Optional entity type to filter by
metadata_filter (dict[str, Any] | None) – Optional metadata filter

Returns:

List of matching entities

Return type:

list[dict[str, Any]]

dataflow.registry.core.get_registry_system()¶: Get the registry system instance (lazy initialization).