praval.storage.storage_registry

Storage Registry System

Provides centralized registration, discovery, and management of storage providers available to Praval agents. This follows the same pattern as the tool registry for consistent framework design.

Functions

get_storage_provider(provider_name[, agent_name])	Get a storage provider from the global registry.
get_storage_registry()	Get the global storage registry instance.
list_storage_providers(**kwargs)	List available storage providers from the global registry.
register_storage_provider(provider, **kwargs)	Register a storage provider in the global registry.

Classes

StorageRegistry()

Central registry for managing storage providers available to agents.

class praval.storage.storage_registry.StorageRegistry

Bases: object

Central registry for managing storage providers available to agents.

Features: - Provider registration and discovery - Type-based organization (relational, object, vector, etc.) - Access control and permissions - Connection pooling and lifecycle management - Usage statistics and health monitoring - Multi-provider query routing

__init__()

async register_provider(provider, replace_existing=False, permissions=None, auto_connect=None)

Register a storage provider in the registry.

Parameters:

• provider (BaseStorageProvider) – Provider instance to register

• replace_existing (bool) – Whether to replace existing provider with same name

• permissions (Optional[List[str]]) – List of agent names allowed to use this provider

• auto_connect (bool) – Whether to auto-connect the provider

Return type:

bool

Returns:

True if registration successful, False otherwise

async unregister_provider(provider_name)

Unregister a provider from the registry.

Parameters:

provider_name (str) – Name of provider to unregister

Return type:

bool

Returns:

True if unregistration successful, False if provider not found

get_provider(provider_name, agent_name=None)

Get a provider by name with permission checking.

Parameters:

• provider_name (str) – Name of the provider to retrieve

• agent_name (str) – Name of the agent requesting the provider

Return type:

BaseStorageProvider

Returns:

BaseStorageProvider instance

Raises:

• StorageNotFoundError – If provider not found

• StoragePermissionError – If agent lacks permission

list_providers(storage_type=None, agent_name=None, connected_only=False)

List available providers with optional filtering.

Parameters:

• storage_type (Optional[StorageType]) – Filter by storage type

• agent_name (Optional[str]) – Filter by agent permissions

• connected_only (bool) – Only return connected providers

Return type:

List[str]

Returns:

List of provider names

get_providers_by_type(storage_type)

Get all providers of a specific storage type.

Return type:

List[str]

Parameters:

storage_type (StorageType)

get_storage_types()

Get list of all available storage types.

Return type:

List[StorageType]

async execute_query(provider_name, query, agent_name=None)

Execute a query on a specific provider.

Parameters:

• provider_name (str) – Name of provider to query

• query (StorageQuery) – Query to execute

• agent_name (Optional[str]) – Name of requesting agent

Return type:

StorageResult

Returns:

StorageResult with query outcome

async health_check_all()

Perform health checks on all registered providers.

Return type:

Dict[str, Dict[str, Any]]

get_usage_stats(provider_name=None)

Get usage statistics for providers.

Return type:

Union[Dict[str, Any], Dict[str, Dict[str, Any]]]

Parameters:

provider_name (str | None)

get_registry_info()

Get comprehensive information about the registry.

Return type:

Dict[str, Any]

set_permissions(provider_name, agent_names)

Set permissions for a provider.

Parameters:

• provider_name (str)

• agent_names (List[str])

add_permission(provider_name, agent_name)

Add permission for an agent to use a provider.

Parameters:

• provider_name (str)

• agent_name (str)

remove_permission(provider_name, agent_name)

Remove permission for an agent to use a provider.

Parameters:

• provider_name (str)

• agent_name (str)

block_provider(provider_name)

Block a provider from being used.

Parameters:

provider_name (str)

unblock_provider(provider_name)

Unblock a previously blocked provider.

Parameters:

provider_name (str)

async cleanup_registry()

Clean up disconnected providers and expired data.

praval.storage.storage_registry.get_storage_registry()

Get the global storage registry instance.

Return type:

StorageRegistry

async praval.storage.storage_registry.register_storage_provider(provider, **kwargs)

Register a storage provider in the global registry.

Return type:

bool

Parameters:

provider (BaseStorageProvider)

praval.storage.storage_registry.get_storage_provider(provider_name, agent_name=None)

Get a storage provider from the global registry.

Return type:

BaseStorageProvider

Parameters:

• provider_name (str)

• agent_name (str)

praval.storage.storage_registry.list_storage_providers(**kwargs)

List available storage providers from the global registry.

Return type:

List[str]