Hub

provide.foundation.hub

Provide Foundation Hub - Component and Command Coordination System

The hub module provides a unified system for registering, discovering, and managing components and CLI commands across the provide-io ecosystem.

Key Features: - Multi-dimensional component registry - CLI command registration and discovery - Entry point discovery - Integration with Click framework - Type-safe decorators using Python 3.11+ features

Example Usage

from provide.foundation.hub import Hub, register_command

class MyResource: def init(self, name: str): self.name = name

@register_command("init") def init_command(): pass

hub = Hub() hub.add_component(MyResource, name="my_resource", version="1.0.0") resource_class = hub.get_component("my_resource") command = hub.get_command("init")

Classes

AsyncContextResource

AsyncContextResource(resource_factory: Any)

Bases: AbstractAsyncContextManager[Any]

Base class for async context-managed resources.

Initialize with a resource factory.

Source code in provide/foundation/hub/protocols.py

def __init__(self, resource_factory: Any) -> None:
    """Initialize with a resource factory."""
    self._resource_factory = resource_factory
    self._resource: Any = None

Functions

__aenter__ async

__aenter__() -> Any

Enter async context and acquire resource.

Source code in provide/foundation/hub/protocols.py

async def __aenter__(self) -> Any:
    """Enter async context and acquire resource."""
    self._resource = await self._resource_factory()
    return self._resource

__aexit__ async

__aexit__(exc_type: Any, exc_val: Any, exc_tb: Any) -> None

Exit async context and cleanup resource.

Source code in provide/foundation/hub/protocols.py

async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    """Exit async context and cleanup resource."""
    if self._resource and hasattr(self._resource, "dispose_async"):
        await self._resource.dispose_async()
    elif self._resource and hasattr(self._resource, "dispose"):
        self._resource.dispose()

AsyncDisposable

Bases: Protocol

Protocol for components that require async cleanup.

Functions

dispose_async async

dispose_async() -> None

Dispose of the component and clean up resources asynchronously.

Source code in provide/foundation/hub/protocols.py

async def dispose_async(self) -> None:
    """Dispose of the component and clean up resources asynchronously."""
    ...

AsyncInitializable

Bases: Protocol

Protocol for components that support async lazy initialization.

Functions

initialize_async async

initialize_async() -> Any

Initialize the component asynchronously.

Source code in provide/foundation/hub/protocols.py

async def initialize_async(self) -> Any:
    """Initialize the component asynchronously."""
    ...

AsyncResourceManager

Bases: ABC

Abstract base class for async resource managers.

Functions

acquire_resource_async abstractmethod async

acquire_resource_async(resource_id: str) -> Any

Acquire a resource by ID asynchronously.

Source code in provide/foundation/hub/protocols.py

@abstractmethod
async def acquire_resource_async(self, resource_id: str) -> Any:
    """Acquire a resource by ID asynchronously."""
    pass

cleanup_all_async abstractmethod async

cleanup_all_async() -> None

Clean up all managed resources asynchronously.

Source code in provide/foundation/hub/protocols.py

@abstractmethod
async def cleanup_all_async(self) -> None:
    """Clean up all managed resources asynchronously."""
    pass

release_resource_async abstractmethod async

release_resource_async(resource_id: str) -> None

Release a resource by ID asynchronously.

Source code in provide/foundation/hub/protocols.py

@abstractmethod
async def release_resource_async(self, resource_id: str) -> None:
    """Release a resource by ID asynchronously."""
    pass

ComponentCategory

Bases: Enum

Predefined component categories for Foundation.

These are the standard dimension values used internally by Foundation. External components can still use custom string dimensions for compatibility.

Container

Container(
    hub: Hub | None = None, registry: Registry | None = None
)

Dependency Injection Container.

A focused API for dependency injection patterns, wrapping the Hub with a simpler interface for type-based registration and resolution.

This container follows the Composition Root pattern where all dependencies are registered at application startup and then resolved as needed.

Example

container = Container() container.register(DatabaseClient, db_instance) container.register(HTTPClient, http_instance)

Resolve with automatic dependency injection

service = container.resolve(MyService)

MyService.init(db, http) called automatically

Pattern

The Container is designed for the Composition Root pattern: 1. Create container at app startup (main.py) 2. Register all core dependencies 3. Resolve application entry points 4. Pass dependencies explicitly (no global access)

This matches the idiomatic patterns in Go and Rust, making it easier to adopt for developers from those ecosystems.

Initialize the DI container.

Parameters:

Name	Type	Description	Default
hub	Hub | None	Optional Hub instance (creates new if not provided)	None
registry	Registry | None	Optional Registry instance (creates new if not provided)	None

Source code in provide/foundation/hub/container.py

def __init__(self, hub: Hub | None = None, registry: Registry | None = None) -> None:
    """Initialize the DI container.

    Args:
        hub: Optional Hub instance (creates new if not provided)
        registry: Optional Registry instance (creates new if not provided)
    """
    if hub is not None:
        self._hub = hub
    elif registry is not None:
        # Create a Hub with the provided registry
        from provide.foundation.context import CLIContext

        self._hub = Hub(context=CLIContext(), component_registry=registry)
    else:
        # Create a new isolated Hub
        from provide.foundation.context import CLIContext

        self._hub = Hub(context=CLIContext())

Functions

__enter__

__enter__() -> Container

Context manager entry.

Example

with Container() as container: ... container.register(Database, db) ... service = container.resolve(MyService)

Source code in provide/foundation/hub/container.py

def __enter__(self) -> Container:
    """Context manager entry.

    Example:
        >>> with Container() as container:
        ...     container.register(Database, db)
        ...     service = container.resolve(MyService)
    """
    return self

__exit__

__exit__(*args: object) -> None

Context manager exit.

Source code in provide/foundation/hub/container.py

def __exit__(self, *args: object) -> None:
    """Context manager exit."""
    # Cleanup is handled by the Hub
    pass

clear

clear() -> None

Clear all registered dependencies.

Warning

This clears the underlying Hub registry. Use with caution.

Source code in provide/foundation/hub/container.py

def clear(self) -> None:
    """Clear all registered dependencies.

    Warning:
        This clears the underlying Hub registry. Use with caution.
    """
    self._hub.clear()

get

get(type_hint: type[T]) -> T | None

Get a registered instance by type.

Parameters:

Name	Type	Description	Default
type_hint	type[T]	Type to retrieve	required

Returns:

Type	Description
T | None	Registered instance or None if not found

Example

db = container.get(Database)

Source code in provide/foundation/hub/container.py

def get(self, type_hint: type[T]) -> T | None:
    """Get a registered instance by type.

    Args:
        type_hint: Type to retrieve

    Returns:
        Registered instance or None if not found

    Example:
        >>> db = container.get(Database)
    """
    # Access the component registry directly
    return self._hub._component_registry.get_by_type(type_hint)

has

has(type_hint: type[Any]) -> bool

Check if a type is registered.

Parameters:

Name	Type	Description	Default
type_hint	type[Any]	Type to check	required

Returns:

Type	Description
bool	True if type is registered

Example

if container.has(Database): ... db = container.get(Database)

Source code in provide/foundation/hub/container.py

def has(self, type_hint: type[Any]) -> bool:
    """Check if a type is registered.

    Args:
        type_hint: Type to check

    Returns:
        True if type is registered

    Example:
        >>> if container.has(Database):
        ...     db = container.get(Database)
    """
    return self.get(type_hint) is not None

register

register(
    type_hint: type[T], instance: T, name: str | None = None
) -> Container

Register a dependency by type.

Parameters:

Name	Type	Description	Default
type_hint	type[T]	Type to register under	required
instance	T	Instance to register	required
name	str | None	Optional name for named registration	None

Returns:

Type	Description
Container	Self for method chaining

Example

container.register(Database, db).register(Cache, cache)

Source code in provide/foundation/hub/container.py

def register(
    self,
    type_hint: type[T],
    instance: T,
    name: str | None = None,
) -> Container:
    """Register a dependency by type.

    Args:
        type_hint: Type to register under
        instance: Instance to register
        name: Optional name for named registration

    Returns:
        Self for method chaining

    Example:
        >>> container.register(Database, db).register(Cache, cache)
    """
    self._hub.register(type_hint, instance, name)
    return self

resolve

resolve(cls: type[T], **overrides: Any) -> T

Resolve a class with dependency injection.

Parameters:

Name	Type	Description	Default
cls	type[T]	Class to instantiate	required
**overrides	Any	Explicitly provided dependencies	{}

Returns:

Type	Description
T	New instance with dependencies injected

Example

service = container.resolve(MyService)

Or with overrides:

service = container.resolve(MyService, logger=custom_logger)

Source code in provide/foundation/hub/container.py

def resolve(
    self,
    cls: type[T],
    **overrides: Any,
) -> T:
    """Resolve a class with dependency injection.

    Args:
        cls: Class to instantiate
        **overrides: Explicitly provided dependencies

    Returns:
        New instance with dependencies injected

    Example:
        >>> service = container.resolve(MyService)
        >>> # Or with overrides:
        >>> service = container.resolve(MyService, logger=custom_logger)
    """
    return self._hub.resolve(cls, **overrides)

Disposable

Bases: Protocol

Protocol for components that require cleanup.

Functions

dispose

dispose() -> None

Dispose of the component and clean up resources.

Source code in provide/foundation/hub/protocols.py

def dispose(self) -> None:
    """Dispose of the component and clean up resources."""
    ...

HealthCheckable

Bases: Protocol

Protocol for components that support health checks.

Functions

health_check

health_check() -> dict[str, Any]

Check component health status.

Source code in provide/foundation/hub/protocols.py

def health_check(self) -> dict[str, Any]:
    """Check component health status."""
    ...

Hub

Hub(
    context: CLIContext | None = None,
    component_registry: Registry | None = None,
    command_registry: Registry | None = None,
    use_shared_registries: bool = False,
)

Bases: CoreHub

Central hub for managing components, commands, and Foundation integration.

The Hub provides a unified interface for: - Registering components and commands - Discovering plugins via entry points - Creating Click CLI applications - Managing component lifecycle - Foundation system initialization

Example

hub = Hub() hub.add_component(MyResource, "resource") hub.add_command(init_cmd, "init") hub.initialize_foundation()

Create CLI with all commands

cli = hub.create_cli() cli()

Initialize the hub.

Parameters:

Name	Type	Description	Default
context	CLIContext | None	Foundation CLIContext for configuration	None
component_registry	Registry | None	Custom component registry	None
command_registry	Registry | None	Custom command registry	None
use_shared_registries	bool	If True, use global shared registries	False

Source code in provide/foundation/hub/manager.py

def __init__(
    self,
    context: CLIContext | None = None,
    component_registry: Registry | None = None,
    command_registry: Registry | None = None,
    use_shared_registries: bool = False,
) -> None:
    """Initialize the hub.

    Args:
        context: Foundation CLIContext for configuration
        component_registry: Custom component registry
        command_registry: Custom command registry
        use_shared_registries: If True, use global shared registries

    """
    # Determine if we should use shared registries
    use_shared = should_use_shared_registries(use_shared_registries, component_registry, command_registry)

    # Setup registries
    if component_registry:
        comp_registry = component_registry
    elif use_shared:
        comp_registry = get_component_registry()
    else:
        comp_registry = Registry()

    if command_registry:
        cmd_registry = command_registry
    elif use_shared:
        cmd_registry = get_command_registry()
    else:
        cmd_registry = Registry()

    # Initialize core hub functionality
    super().__init__(context, comp_registry, cmd_registry)

    # Initialize Foundation management, injecting self
    self._foundation = FoundationManager(hub=self, registry=self._component_registry)

Functions

clear

clear(dimension: str | None = None) -> None

Clear registrations and dispose of resources properly.

Parameters:

Name	Type	Description	Default
dimension	str | None	Optional dimension to clear (None = all)	None

Source code in provide/foundation/hub/manager.py

def clear(self, dimension: str | None = None) -> None:
    """Clear registrations and dispose of resources properly.

    Args:
        dimension: Optional dimension to clear (None = all)

    """
    # Clear core hub registrations (this will now dispose resources)
    super().clear(dimension)

    # Reset Foundation state when clearing all or foundation-specific dimensions
    if dimension is None or dimension in ("singleton", "foundation"):
        self._foundation.clear_foundation_state()

dispose_all

dispose_all() -> None

Dispose of all managed resources without clearing registrations.

Source code in provide/foundation/hub/manager.py

def dispose_all(self) -> None:
    """Dispose of all managed resources without clearing registrations."""
    self._component_registry.dispose_all()
    if hasattr(self._command_registry, "dispose_all"):
        self._command_registry.dispose_all()

get_foundation_config

get_foundation_config() -> Any | None

Get the current Foundation configuration.

Source code in provide/foundation/hub/manager.py

def get_foundation_config(self) -> Any | None:
    """Get the current Foundation configuration."""
    return self._foundation.get_foundation_config()

get_foundation_logger

get_foundation_logger(name: str | None = None) -> Any

Get Foundation logger instance through Hub.

Auto-initializes Foundation if not already done. Thread-safe with fallback behavior.

Parameters:

Name	Type	Description	Default
name	str | None	Logger name (e.g., module name)	None

Returns:

Type	Description
Any	Configured logger instance

Source code in provide/foundation/hub/manager.py

def get_foundation_logger(self, name: str | None = None) -> Any:
    """Get Foundation logger instance through Hub.

    Auto-initializes Foundation if not already done.
    Thread-safe with fallback behavior.

    Args:
        name: Logger name (e.g., module name)

    Returns:
        Configured logger instance

    """
    return self._foundation.get_foundation_logger(name)

initialize_foundation

initialize_foundation(
    config: Any = None, force: bool = False
) -> None

Initialize Foundation system through Hub.

Single initialization method replacing all setup_* functions. Thread-safe and idempotent, unless force=True.

Parameters:

Name	Type	Description	Default
config	Any	Optional TelemetryConfig (defaults to from_env)	None
force	bool	If True, force re-initialization even if already initialized	False

Source code in provide/foundation/hub/manager.py

def initialize_foundation(self, config: Any = None, force: bool = False) -> None:
    """Initialize Foundation system through Hub.

    Single initialization method replacing all setup_* functions.
    Thread-safe and idempotent, unless force=True.

    Args:
        config: Optional TelemetryConfig (defaults to from_env)
        force: If True, force re-initialization even if already initialized

    """
    self._foundation.initialize_foundation(config, force)

is_foundation_initialized

is_foundation_initialized() -> bool

Check if Foundation system is initialized.

Source code in provide/foundation/hub/manager.py

def is_foundation_initialized(self) -> bool:
    """Check if Foundation system is initialized."""
    return self._foundation.is_foundation_initialized()

Initializable

Bases: Protocol

Protocol for components that support lazy initialization.

Functions

initialize

initialize() -> Any

Initialize the component.

Source code in provide/foundation/hub/protocols.py

def initialize(self) -> Any:
    """Initialize the component."""
    ...

Registry

Registry()

Multi-dimensional registry for storing and retrieving objects.

Supports hierarchical organization by dimension (component, command, etc.) and name within each dimension. This is a generic registry that can be used for any type of object storage and retrieval.

Thread-safe: All operations are protected by an RLock for safe concurrent access.

Note: Uses threading.RLock (not asyncio.Lock) for thread safety. For async-only applications with high-frequency registry access in request hot-paths (>10k req/sec with runtime registration), consider using an async-native registry implementation with asyncio.Lock. For typical use cases (initialization-time registration, CLI apps, read-heavy workloads), the threading lock has negligible impact.

See: docs/architecture/design-decisions.md#threading-model

Initialize an empty registry.

Source code in provide/foundation/hub/registry.py

def __init__(self) -> None:
    """Initialize an empty registry."""
    # Use managed lock for deadlock prevention
    # Lock is registered during Foundation initialization via register_foundation_locks()
    from provide.foundation.concurrency.locks import get_lock_manager

    self._lock = get_lock_manager().get_lock("foundation.registry")
    self._registry: dict[str, dict[str, RegistryEntry]] = defaultdict(dict)
    self._aliases: dict[str, tuple[str, str]] = {}
    # Type-based registry for dependency injection
    self._type_registry: dict[type[Any], Any] = {}

Functions

__contains__

__contains__(key: str | tuple[str, str]) -> bool

Check if an item exists in the registry.

Source code in provide/foundation/hub/registry.py

def __contains__(self, key: str | tuple[str, str]) -> bool:
    """Check if an item exists in the registry."""
    with self._lock:
        if isinstance(key, tuple):
            dimension, name = key
            return name in self._registry[dimension]
        return any(key in dim_reg for dim_reg in self._registry.values())

__iter__

__iter__() -> Iterator[RegistryEntry]

Iterate over all registry entries.

Source code in provide/foundation/hub/registry.py

def __iter__(self) -> Iterator[RegistryEntry]:
    """Iterate over all registry entries."""
    with self._lock:
        # Create a snapshot to avoid holding lock during iteration
        entries: list[RegistryEntry] = []
        for dim_registry in self._registry.values():
            entries.extend(dim_registry.values())
    # Yield outside the lock
    yield from entries

__len__

__len__() -> int

Get total number of registered items.

Source code in provide/foundation/hub/registry.py

def __len__(self) -> int:
    """Get total number of registered items."""
    with self._lock:
        return sum(len(dim_reg) for dim_reg in self._registry.values())

clear

clear(dimension: str | None = None) -> None

Clear the registry or a specific dimension.

Source code in provide/foundation/hub/registry.py

def clear(self, dimension: str | None = None) -> None:
    """Clear the registry or a specific dimension."""
    with self._lock:
        if dimension is not None:
            # Dispose of resources before clearing
            self._dispose_resources(dimension)
            self._registry[dimension].clear()

            aliases_to_remove = [alias for alias, (dim, _) in self._aliases.items() if dim == dimension]
            for alias in aliases_to_remove:
                del self._aliases[alias]
        else:
            # Dispose of all resources before clearing
            self._dispose_all_resources()
            self._registry.clear()
            self._aliases.clear()
            self._type_registry.clear()

dispose_all

dispose_all() -> None

Dispose of all registered resources properly.

Source code in provide/foundation/hub/registry.py

def dispose_all(self) -> None:
    """Dispose of all registered resources properly."""
    with self._lock:
        self._dispose_all_resources()

get

get(name: str, dimension: str | None = None) -> Any | None

Get an item from the registry.

Parameters:

Name	Type	Description	Default
name	str	Name or alias of the item	required
dimension	str | None	Optional dimension to search in	None

Returns:

Type	Description
Any | None	The registered value or None if not found

Source code in provide/foundation/hub/registry.py

def get(
    self,
    name: str,
    dimension: str | None = None,
) -> Any | None:
    """Get an item from the registry.

    Args:
        name: Name or alias of the item
        dimension: Optional dimension to search in

    Returns:
        The registered value or None if not found

    """
    with self._lock:
        if dimension is not None:
            entry = self._registry[dimension].get(name)
            if entry:
                return entry.value

        if name in self._aliases:
            dim_key, real_name = self._aliases[name]
            if dimension is None or dim_key == dimension:
                entry = self._registry[dim_key].get(real_name)
                if entry:
                    return entry.value

        if dimension is None:
            for dim_registry in self._registry.values():
                if name in dim_registry:
                    return dim_registry[name].value

        return None

get_by_type

get_by_type(type_hint: type[Any]) -> Any | None

Get a registered instance by its type.

Parameters:

Name	Type	Description	Default
type_hint	type[Any]	Type to look up	required

Returns:

Type	Description
Any | None	Registered instance or None if not found

Example

db = registry.get_by_type(DatabaseClient)

Source code in provide/foundation/hub/registry.py

def get_by_type(self, type_hint: type[Any]) -> Any | None:
    """Get a registered instance by its type.

    Args:
        type_hint: Type to look up

    Returns:
        Registered instance or None if not found

    Example:
        >>> db = registry.get_by_type(DatabaseClient)
    """
    with self._lock:
        return self._type_registry.get(type_hint)

get_entry

get_entry(
    name: str, dimension: str | None = None
) -> RegistryEntry | None

Get the full registry entry.

Source code in provide/foundation/hub/registry.py

def get_entry(
    self,
    name: str,
    dimension: str | None = None,
) -> RegistryEntry | None:
    """Get the full registry entry."""
    with self._lock:
        if dimension is not None:
            return self._registry[dimension].get(name)

        if name in self._aliases:
            dim_key, real_name = self._aliases[name]
            if dimension is None or dim_key == dimension:
                return self._registry[dim_key].get(real_name)

        if dimension is None:
            for dim_registry in self._registry.values():
                if name in dim_registry:
                    return dim_registry[name]

        return None

list_all

list_all() -> dict[str, list[str]]

List all dimensions and their items.

Source code in provide/foundation/hub/registry.py

def list_all(self) -> dict[str, list[str]]:
    """List all dimensions and their items."""
    with self._lock:
        return {dimension: list(items.keys()) for dimension, items in self._registry.items()}

list_dimension

list_dimension(dimension: str) -> list[str]

List all names in a dimension.

Source code in provide/foundation/hub/registry.py

def list_dimension(
    self,
    dimension: str,
) -> list[str]:
    """List all names in a dimension."""
    with self._lock:
        return list(self._registry[dimension].keys())

list_types

list_types() -> list[type[Any]]

List all registered types.

Returns:

Type	Description
list[type[Any]]	List of registered types

Source code in provide/foundation/hub/registry.py

def list_types(self) -> list[type[Any]]:
    """List all registered types.

    Returns:
        List of registered types
    """
    with self._lock:
        return list(self._type_registry.keys())

register

register(
    name: str,
    value: Any,
    dimension: str = "default",
    metadata: dict[str, Any] | None = None,
    aliases: list[str] | None = None,
    replace: bool = False,
) -> RegistryEntry

Register an item in the registry.

Parameters:

Name	Type	Description	Default
name	str	Unique name within the dimension	required
value	Any	The item to register	required
dimension	str	Registry dimension for categorization	'default'
metadata	dict[str, Any] | None	Optional metadata about the item	None
aliases	list[str] | None	Optional list of aliases for this item	None
replace	bool	Whether to replace existing entries	False

Returns:

Type	Description
RegistryEntry	The created registry entry

Raises:

Type	Description
ValueError	If name already exists and replace=False

Source code in provide/foundation/hub/registry.py

def register(
    self,
    name: str,
    value: Any,
    dimension: str = "default",
    metadata: dict[str, Any] | None = None,
    aliases: list[str] | None = None,
    replace: bool = False,
) -> RegistryEntry:
    """Register an item in the registry.

    Args:
        name: Unique name within the dimension
        value: The item to register
        dimension: Registry dimension for categorization
        metadata: Optional metadata about the item
        aliases: Optional list of aliases for this item
        replace: Whether to replace existing entries

    Returns:
        The created registry entry

    Raises:
        ValueError: If name already exists and replace=False

    """
    with self._lock:
        if not replace and name in self._registry[dimension]:
            raise AlreadyExistsError(
                f"Item '{name}' already registered in dimension '{dimension}'. "
                "Use replace=True to override.",
                code="REGISTRY_ITEM_EXISTS",
                item_name=name,
                dimension=dimension,
            )

        entry = RegistryEntry(
            name=name,
            dimension=dimension,
            value=value,
            metadata=metadata or {},
        )

        self._registry[dimension][name] = entry

        if aliases:
            for alias in aliases:
                self._aliases[alias] = (dimension, name)

        # Emit event instead of direct logging to break circular dependency
        from provide.foundation.hub.events import emit_registry_event

        emit_registry_event(
            operation="register",
            item_name=name,
            dimension=dimension,
            has_metadata=bool(metadata),
            aliases=aliases,
        )

        return entry

register_type

register_type(
    type_hint: type[Any],
    instance: Any,
    name: str | None = None,
) -> None

Register an instance by its type for dependency injection.

This enables type-based lookup which is essential for DI patterns.

Parameters:

Name	Type	Description	Default
type_hint	type[Any]	Type to register under	required
instance	Any	Instance to register	required
name	str | None	Optional name for standard registry (defaults to type name)	None

Example

registry.register_type(DatabaseClient, db_instance) db = registry.get_by_type(DatabaseClient)

Source code in provide/foundation/hub/registry.py

def register_type(
    self,
    type_hint: type[Any],
    instance: Any,
    name: str | None = None,
) -> None:
    """Register an instance by its type for dependency injection.

    This enables type-based lookup which is essential for DI patterns.

    Args:
        type_hint: Type to register under
        instance: Instance to register
        name: Optional name for standard registry (defaults to type name)

    Example:
        >>> registry.register_type(DatabaseClient, db_instance)
        >>> db = registry.get_by_type(DatabaseClient)
    """
    with self._lock:
        self._type_registry[type_hint] = instance

        # Also register in standard registry for backward compatibility
        if name is not None:
            self.register(
                name=name,
                value=instance,
                dimension="types",
                metadata={"type": type_hint},
                replace=True,
            )

remove

remove(name: str, dimension: str | None = None) -> bool

Remove an item from the registry.

Returns:

Type	Description
bool	True if item was removed, False if not found

Source code in provide/foundation/hub/registry.py

def remove(
    self,
    name: str,
    dimension: str | None = None,
) -> bool:
    """Remove an item from the registry.

    Returns:
        True if item was removed, False if not found

    """
    with self._lock:
        if dimension is not None:
            if name in self._registry[dimension]:
                del self._registry[dimension][name]

                aliases_to_remove = [
                    alias for alias, (dim, n) in self._aliases.items() if dim == dimension and n == name
                ]
                for alias in aliases_to_remove:
                    del self._aliases[alias]

                # Emit event instead of direct logging to break circular dependency
                from provide.foundation.hub.events import emit_registry_event

                emit_registry_event(
                    operation="remove",
                    item_name=name,
                    dimension=dimension,
                )
                return True
        else:
            for dim_key, dim_registry in self._registry.items():
                if name in dim_registry:
                    del dim_registry[name]

                    aliases_to_remove = [
                        alias for alias, (d, n) in self._aliases.items() if d == dim_key and n == name
                    ]
                    for alias in aliases_to_remove:
                        del self._aliases[alias]

                    # Emit event instead of direct logging to break circular dependency
                    from provide.foundation.hub.events import emit_registry_event

                    emit_registry_event(
                        operation="remove",
                        item_name=name,
                        dimension=dim_key,
                    )
                    return True

        return False

RegistryEntry

A single entry in the registry.

Attributes

key property

key: tuple[str, str]

Get the registry key for this entry.

ResourceManager

Bases: ABC

Abstract base class for resource managers.

Functions

acquire_resource abstractmethod

acquire_resource(resource_id: str) -> Any

Acquire a resource by ID.

Source code in provide/foundation/hub/protocols.py

@abstractmethod
def acquire_resource(self, resource_id: str) -> Any:
    """Acquire a resource by ID."""
    pass

cleanup_all abstractmethod

cleanup_all() -> None

Clean up all managed resources.

Source code in provide/foundation/hub/protocols.py

@abstractmethod
def cleanup_all(self) -> None:
    """Clean up all managed resources."""
    pass

release_resource abstractmethod

release_resource(resource_id: str) -> None

Release a resource by ID.

Source code in provide/foundation/hub/protocols.py

@abstractmethod
def release_resource(self, resource_id: str) -> None:
    """Release a resource by ID."""
    pass

Functions

clear_hub

clear_hub() -> None

Clear the global hub instance.

This is primarily used for testing to reset Foundation state between test runs.

Source code in provide/foundation/hub/manager.py

def clear_hub() -> None:
    """Clear the global hub instance.

    This is primarily used for testing to reset Foundation state
    between test runs.

    """
    global _global_hub
    if _global_hub:
        _global_hub.clear()
    _global_hub = None

create_container

create_container() -> Container

Create a new DI container.

Convenience function for creating containers.

Returns:

Type	Description
Container	New Container instance

Example

container = create_container() container.register(Database, db_instance)

Source code in provide/foundation/hub/container.py

def create_container() -> Container:
    """Create a new DI container.

    Convenience function for creating containers.

    Returns:
        New Container instance

    Example:
        >>> container = create_container()
        >>> container.register(Database, db_instance)
    """
    return Container()

get_component_registry

get_component_registry() -> Registry

Get the global component registry.

Source code in provide/foundation/hub/components.py

def get_component_registry() -> Registry:
    """Get the global component registry."""
    return _component_registry

get_hub

get_hub() -> Hub

Get the global shared hub instance (singleton pattern).

This function acts as the Composition Root for the global singleton instance. It is maintained for backward compatibility and convenience.

Note: For building testable and maintainable applications, the recommended approach is to use a Container or Hub instance created at your application's entry point for explicit dependency management. This global accessor should be avoided in application code.

Thread-safe: Uses double-checked locking pattern for efficient lazy initialization.

Auto-Initialization Behavior: This function automatically initializes the Foundation system on first access. The initialization is: - Idempotent: Safe to call multiple times - Thread-safe: Uses lock manager for coordination - Lazy: Only happens on first access

Returns:

Type	Description
Hub	Global Hub instance (created and initialized if needed)

Example

hub = get_hub() hub.register_command("my_command", my_function)

Note

For isolated Hub instances (testing, advanced use cases), use:

hub = Hub(use_shared_registries=False)

Source code in provide/foundation/hub/manager.py

def get_hub() -> Hub:
    """Get the global shared hub instance (singleton pattern).

    This function acts as the Composition Root for the global singleton instance.
    It is maintained for backward compatibility and convenience.

    **Note:** For building testable and maintainable applications, the recommended
    approach is to use a `Container` or `Hub` instance created at your application's
    entry point for explicit dependency management. This global accessor should be
    avoided in application code.

    Thread-safe: Uses double-checked locking pattern for efficient lazy initialization.

    **Auto-Initialization Behavior:**
    This function automatically initializes the Foundation system on first access.
    The initialization is:
    - **Idempotent**: Safe to call multiple times
    - **Thread-safe**: Uses lock manager for coordination
    - **Lazy**: Only happens on first access

    Returns:
        Global Hub instance (created and initialized if needed)

    Example:
        >>> hub = get_hub()
        >>> hub.register_command("my_command", my_function)

    Note:
        For isolated Hub instances (testing, advanced use cases), use:
        >>> hub = Hub(use_shared_registries=False)

    """
    global _global_hub

    # Fast path: hub already initialized
    if _global_hub is not None:
        return _global_hub

    # Slow path: need to initialize hub
    with get_lock_manager().acquire("foundation.hub.init"):
        # Double-check after acquiring lock
        if _global_hub is None:
            # Global hub uses shared registries by default
            _global_hub = Hub(use_shared_registries=True)

            # Auto-initialize Foundation on first hub access
            _global_hub.initialize_foundation()

            # Bootstrap foundation components now that hub is ready (skip in test mode)
            from provide.foundation.testmode.detection import is_in_test_mode

            if not is_in_test_mode():
                try:
                    from provide.foundation.hub.components import bootstrap_foundation

                    bootstrap_foundation()
                except ImportError:
                    # Bootstrap function might not exist yet, that's okay
                    pass

    return _global_hub

injectable

injectable(cls: type[T]) -> type[T]

Mark a class as injectable for dependency injection.

This decorator indicates that the class follows DI patterns and can be automatically instantiated by the Hub's resolve() method.

The decorator: - Validates that init has type hints for all parameters - Marks the class as injectable for documentation purposes - Does not modify the class behavior (zero runtime overhead)

Parameters:

Name	Type	Description	Default
cls	type[T]	Class to mark as injectable	required

Returns:

Type	Description
type[T]	The same class, marked as injectable

Raises:

Type	Description
ValidationError	If init lacks proper type hints

Example

@injectable class UserService: ... def init(self, db: Database, cache: Cache): ... self.db = db ... self.cache = cache

Source code in provide/foundation/hub/injection.py

def injectable(cls: type[T]) -> type[T]:
    """Mark a class as injectable for dependency injection.

    This decorator indicates that the class follows DI patterns and
    can be automatically instantiated by the Hub's resolve() method.

    The decorator:
    - Validates that __init__ has type hints for all parameters
    - Marks the class as injectable for documentation purposes
    - Does not modify the class behavior (zero runtime overhead)

    Args:
        cls: Class to mark as injectable

    Returns:
        The same class, marked as injectable

    Raises:
        ValidationError: If __init__ lacks proper type hints

    Example:
        >>> @injectable
        >>> class UserService:
        ...     def __init__(self, db: Database, cache: Cache):
        ...         self.db = db
        ...         self.cache = cache
    """
    # Check if class has custom __init__ (not inherited from object)
    if "__init__" not in cls.__dict__:
        raise ValidationError(
            f"Injectable class {cls.__name__} must define its own __init__ method",
            code="INJECTABLE_NO_INIT",
            class_name=cls.__name__,
        )

    # Get type hints for __init__
    try:
        # Use localns to resolve forward references within the class's module
        import sys

        module = sys.modules.get(cls.__module__)
        localns = vars(module) if module else {}
        get_type_hints(cls.__init__, globalns=None, localns=localns)
    except NameError:
        # Forward reference couldn't be resolved - that's okay for now
        # The actual resolution will happen at runtime
        pass
    except Exception as e:
        raise ValidationError(
            f"Failed to get type hints for {cls.__name__}.__init__: {e}",
            code="INJECTABLE_TYPE_HINT_ERROR",
            class_name=cls.__name__,
            cause=e,
        ) from e

    # Get signature to check for untyped parameters
    sig = inspect.signature(cls.__init__)
    params = list(sig.parameters.values())[1:]  # Skip 'self'

    # Check each parameter has a type hint (excluding *args, **kwargs)
    untyped_params = []
    for param in params:
        if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
            continue  # *args and **kwargs are fine without hints
        # Check if parameter has annotation in signature
        if param.annotation == inspect.Parameter.empty:
            untyped_params.append(param.name)

    if untyped_params:
        raise ValidationError(
            f"Injectable class {cls.__name__}.__init__ has untyped parameters: {untyped_params}. "
            "All constructor parameters must have type hints for dependency injection.",
            code="INJECTABLE_UNTYPED_PARAMS",
            class_name=cls.__name__,
            untyped_params=untyped_params,
        )

    # Mark as injectable (for introspection/documentation)
    setattr(cls, _INJECTABLE_MARKER, True)

    return cls

is_injectable

is_injectable(cls: type[Any]) -> bool

Check if a class is marked as injectable.

Parameters:

Name	Type	Description	Default
cls	type[Any]	Class to check	required

Returns:

Type	Description
bool	True if class is marked with @injectable decorator

Source code in provide/foundation/hub/injection.py

def is_injectable(cls: type[Any]) -> bool:
    """Check if a class is marked as injectable.

    Args:
        cls: Class to check

    Returns:
        True if class is marked with @injectable decorator
    """
    return getattr(cls, _INJECTABLE_MARKER, False)

register_command

register_command(
    name: str | None = None,
    *,
    description: str | None = None,
    aliases: list[str] | None = None,
    hidden: bool = False,
    category: str | None = None,
    group: bool = False,
    replace: bool = False,
    registry: Registry | None = None,
    **metadata: Any
) -> Callable[[F], F]

register_command(func: F) -> F

register_command(
    name_or_func: str | F | None = None,
    *,
    description: str | None = None,
    aliases: list[str] | None = None,
    hidden: bool = False,
    category: str | None = None,
    group: bool = False,
    replace: bool = False,
    registry: Registry | None = None,
    **metadata: Any
) -> Any

Register a CLI command in the hub.

Can be used as a decorator with or without arguments:

@register_command
def my_command():
    pass

@register_command("custom-name", aliases=["cn"], category="utils")
def my_command():
    pass

# Nested commands using dot notation - groups are auto-created
@register_command("container.status")
def container_status():
    pass

@register_command("container.volumes.backup")
def container_volumes_backup():
    pass

# Explicit group with custom description (optional)
@register_command("container", group=True, description="Container management")
def container_group():
    pass

Parameters:

Name	Type	Description	Default
name_or_func	str | F | None	Command name using dot notation for nesting (e.g., "container.status")	None
description	str | None	Command description (defaults to docstring)	None
aliases	list[str] | None	Alternative names for the command	None
hidden	bool	Whether to hide from help listing	False
category	str | None	Command category for grouping	None
group	bool	Whether this is a command group (not a command)	False
replace	bool	Whether to replace existing registration	False
registry	Registry | None	Custom registry (defaults to global)	None
**metadata	Any	Additional metadata stored in CommandInfo.metadata	{}

Returns:

Type	Description
Any	Decorator function or decorated function

Source code in provide/foundation/hub/decorators.py

def register_command(  # type: ignore[misc]
    name_or_func: str | F | None = None,
    *,
    description: str | None = None,
    aliases: list[str] | None = None,
    hidden: bool = False,
    category: str | None = None,
    group: bool = False,
    replace: bool = False,
    registry: Registry | None = None,
    **metadata: Any,
) -> Any:
    """Register a CLI command in the hub.

    Can be used as a decorator with or without arguments:

        @register_command
        def my_command():
            pass

        @register_command("custom-name", aliases=["cn"], category="utils")
        def my_command():
            pass

        # Nested commands using dot notation - groups are auto-created
        @register_command("container.status")
        def container_status():
            pass

        @register_command("container.volumes.backup")
        def container_volumes_backup():
            pass

        # Explicit group with custom description (optional)
        @register_command("container", group=True, description="Container management")
        def container_group():
            pass

    Args:
        name_or_func: Command name using dot notation for nesting (e.g., "container.status")
        description: Command description (defaults to docstring)
        aliases: Alternative names for the command
        hidden: Whether to hide from help listing
        category: Command category for grouping
        group: Whether this is a command group (not a command)
        replace: Whether to replace existing registration
        registry: Custom registry (defaults to global)
        **metadata: Additional metadata stored in CommandInfo.metadata

    Returns:
        Decorator function or decorated function

    """
    # Handle @register_command (without parens)
    if callable(name_or_func) and not isinstance(name_or_func, str):
        func = name_or_func
        return _register_command_func(
            func,
            name=None,
            description=description,
            aliases=aliases,
            hidden=hidden,
            category=category,
            group=group,
            replace=replace,
            registry=registry,
            **metadata,
        )

    # Handle @register_command(...) (with arguments)
    # At this point, name_or_func must be str | None (not F)
    name_str: str | None = name_or_func if isinstance(name_or_func, str) or name_or_func is None else None

    def decorator(func: F) -> F:
        return _register_command_func(
            func,
            name=name_str,
            description=description,
            aliases=aliases,
            hidden=hidden,
            category=category,
            group=group,
            replace=replace,
            registry=registry,
            **metadata,
        )

    return decorator