dataknobs-config Complete API Reference

Complete auto-generated API documentation from source code docstrings.

💡 Also see: - Curated Guide - Hand-crafted tutorials and examples - Package Overview - Introduction and getting started - Source Code - View on GitHub

---

dataknobs_config

DataKnobs Config Package

A modular, reusable configuration system for composable settings.

Modules:

Name	Description
binding_resolver	Configuration binding resolver for resource instantiation.
builders	Optional object construction and caching functionality.
config	Core Config class implementation.
environment	Environment variable override system.
environment_aware	Environment-aware configuration with late-binding resource resolution.
environment_config	Environment-specific configuration and resource bindings.
examples	Example classes for dataknobs-config.
exceptions	Custom exceptions for the config package.
inheritance	Configuration inheritance utilities.
references	String reference system for cross-referencing configurations.
settings	Global settings and defaults management.
substitution	Environment variable substitution for configuration values.
template_vars	Template variable substitution for nested configuration structures.

Classes:

Name	Description
AsyncCallableFactory	Factory that wraps an async callable.
BindingResolverError	Error during resource binding resolution.
CallableFactory	Factory that wraps a callable.
ConfigBindingResolver	Resolves logical resource bindings to concrete instances.
FactoryNotFoundError	Factory not registered for resource type.
SimpleFactory	Simple factory that creates instances of a class.
ConfigurableBase	Base class for objects that can be configured.
FactoryBase	Base class for factory objects.
Config	A modular configuration system for composable settings.
EnvironmentAwareConfig	Configuration with environment-aware resource resolution.
EnvironmentAwareConfigError	Error related to environment-aware configuration.
EnvironmentConfig	Environment-specific configuration and resource bindings.
EnvironmentConfigError	Error related to environment configuration.
ResourceBinding	A binding from logical name to concrete implementation.
ResourceNotFoundError	Resource not found in environment configuration.
ConfigNotFoundError	Raised when a requested configuration is not found.
InvalidReferenceError	Raised when a configuration reference is invalid.
InheritableConfigLoader	Configuration loader with inheritance support.
InheritanceError	Error during configuration inheritance resolution.
VariableSubstitution	Handles environment variable substitution in configuration values.

Functions:

Name	Description
deep_merge	Deep merge two dictionaries.
load_config_with_inheritance	Convenience function to load a config file with inheritance.
substitute_env_vars	Recursively substitute environment variables in configuration.
substitute_template_vars	Recursively substitute {{var}} placeholders in configuration data.

Classes

AsyncCallableFactory

AsyncCallableFactory(func: Callable[..., Any], **default_kwargs: Any)

Factory that wraps an async callable.

Example

async def create_database(backend, connection_string, **kwargs):
    db = DatabaseConnection(backend, connection_string)
    await db.connect()
    return db

resolver.register_factory(
    "databases",
    AsyncCallableFactory(create_database)
)

Initialize with an async callable.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	Async callable that creates resources	required
**default_kwargs	Any	Default kwargs merged with config	{}

Methods:

Name	Description
create_async	Create an instance asynchronously.
create	Sync create is not supported for async factories.

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def __init__(self, func: Callable[..., Any], **default_kwargs: Any):
    """Initialize with an async callable.

    Args:
        func: Async callable that creates resources
        **default_kwargs: Default kwargs merged with config
    """
    self._func = func
    self._defaults = default_kwargs

Functions

create_async async

create_async(**config: Any) -> Any

Create an instance asynchronously.

Parameters:

Name	Type	Description	Default
**config	Any	Configuration merged with defaults	{}

Returns:

Type	Description
Any	Created instance

Source code in packages/config/src/dataknobs_config/binding_resolver.py

async def create_async(self, **config: Any) -> Any:
    """Create an instance asynchronously.

    Args:
        **config: Configuration merged with defaults

    Returns:
        Created instance
    """
    merged = {**self._defaults, **config}
    return await self._func(**merged)

create

create(**config: Any) -> Any

Sync create is not supported for async factories.

Raises:

Type	Description
RuntimeError	Always, use create_async instead

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def create(self, **config: Any) -> Any:
    """Sync create is not supported for async factories.

    Raises:
        RuntimeError: Always, use create_async instead
    """
    raise RuntimeError(
        "AsyncCallableFactory requires async context. "
        "Use resolve_async() instead of resolve()."
    )

BindingResolverError

Bases: Exception

Error during resource binding resolution.

CallableFactory

CallableFactory(func: Callable[..., Any], **default_kwargs: Any)

Factory that wraps a callable.

Example

def create_database(backend, connection_string, **kwargs):
    if backend == "postgres":
        return PostgresDB(connection_string, **kwargs)
    elif backend == "sqlite":
        return SQLiteDB(connection_string, **kwargs)

resolver.register_factory(
    "databases",
    CallableFactory(create_database)
)

Initialize with a callable.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	Callable that creates resources	required
**default_kwargs	Any	Default kwargs merged with config	{}

Methods:

Name	Description
create	Create an instance.

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def __init__(self, func: Callable[..., Any], **default_kwargs: Any):
    """Initialize with a callable.

    Args:
        func: Callable that creates resources
        **default_kwargs: Default kwargs merged with config
    """
    self._func = func
    self._defaults = default_kwargs

Functions

create

create(**config: Any) -> Any

Create an instance.

Parameters:

Name	Type	Description	Default
**config	Any	Configuration merged with defaults	{}

Returns:

Type	Description
Any	Created instance

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def create(self, **config: Any) -> Any:
    """Create an instance.

    Args:
        **config: Configuration merged with defaults

    Returns:
        Created instance
    """
    merged = {**self._defaults, **config}
    return self._func(**merged)

ConfigBindingResolver

ConfigBindingResolver(
    environment: EnvironmentConfig, resolve_env_vars: bool = True
)

Resolves logical resource bindings to concrete instances.

Works with EnvironmentConfig to: 1. Look up resource configurations by logical name 2. Resolve environment variables in configurations 3. Instantiate resources using registered factories 4. Cache instances for reuse

Attributes:

Name	Type	Description
environment	EnvironmentConfig	The EnvironmentConfig for resource lookup

Initialize the binding resolver.

Parameters:

Name	Type	Description	Default
environment	EnvironmentConfig	Environment configuration for resource lookup	required
resolve_env_vars	bool	Whether to resolve env vars before instantiation	True

Methods:

Name	Description
register_factory	Register a factory for a resource type.
unregister_factory	Unregister a factory for a resource type.
has_factory	Check if a factory is registered for a resource type.
get_registered_types	Get all registered resource types.
resolve	Resolve a logical resource reference to a concrete instance.
resolve_async	Async version of resolve for async factories.
get_cached	Get a cached instance if it exists.
is_cached	Check if an instance is cached.
clear_cache	Clear cached resource instances.
cache_instance	Manually cache an instance.

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def __init__(
    self,
    environment: EnvironmentConfig,
    resolve_env_vars: bool = True,
):
    """Initialize the binding resolver.

    Args:
        environment: Environment configuration for resource lookup
        resolve_env_vars: Whether to resolve env vars before instantiation
    """
    self._environment = environment
    self._resolve_env_vars = resolve_env_vars
    self._factories: dict[str, ResourceFactory | Callable[..., Any]] = {}
    self._cache: dict[tuple[str, str], Any] = {}

Attributes

environment property

environment: EnvironmentConfig

Get the environment configuration.

Functions

register_factory

register_factory(
    resource_type: str, factory: ResourceFactory | Callable[..., Any]
) -> None

Register a factory for a resource type.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource (e.g., "databases", "vector_stores")	required
factory	ResourceFactory | Callable[..., Any]	Factory instance or callable that creates resources. Must have create(**config) method or be callable.	required

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def register_factory(
    self,
    resource_type: str,
    factory: ResourceFactory | Callable[..., Any],
) -> None:
    """Register a factory for a resource type.

    Args:
        resource_type: Type of resource (e.g., "databases", "vector_stores")
        factory: Factory instance or callable that creates resources.
                Must have create(**config) method or be callable.
    """
    self._factories[resource_type] = factory
    logger.debug(f"Registered factory for resource type: {resource_type}")

unregister_factory

unregister_factory(resource_type: str) -> None

Unregister a factory for a resource type.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource to unregister	required

Raises:

Type	Description
KeyError	If factory not registered

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def unregister_factory(self, resource_type: str) -> None:
    """Unregister a factory for a resource type.

    Args:
        resource_type: Type of resource to unregister

    Raises:
        KeyError: If factory not registered
    """
    if resource_type not in self._factories:
        raise KeyError(f"No factory registered for: {resource_type}")
    del self._factories[resource_type]
    logger.debug(f"Unregistered factory for resource type: {resource_type}")

has_factory

has_factory(resource_type: str) -> bool

Check if a factory is registered for a resource type.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource	required

Returns:

Type	Description
bool	True if factory is registered

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def has_factory(self, resource_type: str) -> bool:
    """Check if a factory is registered for a resource type.

    Args:
        resource_type: Type of resource

    Returns:
        True if factory is registered
    """
    return resource_type in self._factories

get_registered_types

get_registered_types() -> list[str]

Get all registered resource types.

Returns:

Type	Description
list[str]	List of resource type names with registered factories

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def get_registered_types(self) -> list[str]:
    """Get all registered resource types.

    Returns:
        List of resource type names with registered factories
    """
    return list(self._factories.keys())

resolve

resolve(
    resource_type: str,
    logical_name: str,
    use_cache: bool = True,
    **overrides: Any,
) -> Any

Resolve a logical resource reference to a concrete instance.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource ("databases", "vector_stores", etc.)	required
logical_name	str	Logical name from app config	required
use_cache	bool	Whether to return cached instance if available	True
**overrides	Any	Config overrides for this resolution	{}

Returns:

Type	Description
Any	Instantiated resource

Raises:

Type	Description
FactoryNotFoundError	If no factory registered for resource type
BindingResolverError	If resource creation fails

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def resolve(
    self,
    resource_type: str,
    logical_name: str,
    use_cache: bool = True,
    **overrides: Any,
) -> Any:
    """Resolve a logical resource reference to a concrete instance.

    Args:
        resource_type: Type of resource ("databases", "vector_stores", etc.)
        logical_name: Logical name from app config
        use_cache: Whether to return cached instance if available
        **overrides: Config overrides for this resolution

    Returns:
        Instantiated resource

    Raises:
        FactoryNotFoundError: If no factory registered for resource type
        BindingResolverError: If resource creation fails
    """
    cache_key = (resource_type, logical_name)

    # Check cache
    if use_cache and cache_key in self._cache:
        logger.debug(f"Returning cached {resource_type}[{logical_name}]")
        return self._cache[cache_key]

    # Get config from environment
    config = self._get_resolved_config(resource_type, logical_name, overrides)

    # Get factory for this type
    factory = self._get_factory(resource_type)

    # Create instance
    try:
        instance = self._create_instance(factory, config)
    except Exception as e:
        raise BindingResolverError(
            f"Failed to create {resource_type}[{logical_name}]: {e}"
        ) from e

    # Cache the instance
    if use_cache:
        self._cache[cache_key] = instance
        logger.debug(f"Cached {resource_type}[{logical_name}]")

    return instance

resolve_async async

resolve_async(
    resource_type: str,
    logical_name: str,
    use_cache: bool = True,
    **overrides: Any,
) -> Any

Async version of resolve for async factories.

If the factory has a create_async method, it will be used. Otherwise, falls back to synchronous create.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource	required
logical_name	str	Logical name from app config	required
use_cache	bool	Whether to return cached instance	True
**overrides	Any	Config overrides for this resolution	{}

Returns:

Type	Description
Any	Instantiated resource

Raises:

Type	Description
FactoryNotFoundError	If no factory registered for resource type
BindingResolverError	If resource creation fails

Source code in packages/config/src/dataknobs_config/binding_resolver.py

async def resolve_async(
    self,
    resource_type: str,
    logical_name: str,
    use_cache: bool = True,
    **overrides: Any,
) -> Any:
    """Async version of resolve for async factories.

    If the factory has a create_async method, it will be used.
    Otherwise, falls back to synchronous create.

    Args:
        resource_type: Type of resource
        logical_name: Logical name from app config
        use_cache: Whether to return cached instance
        **overrides: Config overrides for this resolution

    Returns:
        Instantiated resource

    Raises:
        FactoryNotFoundError: If no factory registered for resource type
        BindingResolverError: If resource creation fails
    """
    cache_key = (resource_type, logical_name)

    # Check cache
    if use_cache and cache_key in self._cache:
        logger.debug(f"Returning cached {resource_type}[{logical_name}]")
        return self._cache[cache_key]

    # Get config from environment
    config = self._get_resolved_config(resource_type, logical_name, overrides)

    # Get factory for this type
    factory = self._get_factory(resource_type)

    # Create instance (async if supported)
    try:
        instance = await self._create_instance_async(factory, config)
    except Exception as e:
        raise BindingResolverError(
            f"Failed to create {resource_type}[{logical_name}]: {e}"
        ) from e

    # Cache the instance
    if use_cache:
        self._cache[cache_key] = instance
        logger.debug(f"Cached {resource_type}[{logical_name}]")

    return instance

get_cached

get_cached(resource_type: str, logical_name: str) -> Any | None

Get a cached instance if it exists.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource	required
logical_name	str	Logical name	required

Returns:

Type	Description
Any | None	Cached instance or None

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def get_cached(
    self,
    resource_type: str,
    logical_name: str,
) -> Any | None:
    """Get a cached instance if it exists.

    Args:
        resource_type: Type of resource
        logical_name: Logical name

    Returns:
        Cached instance or None
    """
    return self._cache.get((resource_type, logical_name))

is_cached

is_cached(resource_type: str, logical_name: str) -> bool

Check if an instance is cached.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource	required
logical_name	str	Logical name	required

Returns:

Type	Description
bool	True if cached

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def is_cached(self, resource_type: str, logical_name: str) -> bool:
    """Check if an instance is cached.

    Args:
        resource_type: Type of resource
        logical_name: Logical name

    Returns:
        True if cached
    """
    return (resource_type, logical_name) in self._cache

clear_cache

clear_cache(resource_type: str | None = None) -> None

Clear cached resource instances.

Parameters:

Name	Type	Description	Default
resource_type	str | None	Specific resource type to clear, or None for all	None

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def clear_cache(self, resource_type: str | None = None) -> None:
    """Clear cached resource instances.

    Args:
        resource_type: Specific resource type to clear, or None for all
    """
    if resource_type:
        # Clear only for specific type
        keys_to_remove = [k for k in self._cache if k[0] == resource_type]
        for key in keys_to_remove:
            del self._cache[key]
        logger.debug(f"Cleared cache for resource type: {resource_type}")
    else:
        self._cache.clear()
        logger.debug("Cleared all cached resources")

cache_instance

cache_instance(resource_type: str, logical_name: str, instance: Any) -> None

Manually cache an instance.

Useful for pre-populating cache or caching externally created instances.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource	required
logical_name	str	Logical name	required
instance	Any	Instance to cache	required

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def cache_instance(
    self,
    resource_type: str,
    logical_name: str,
    instance: Any,
) -> None:
    """Manually cache an instance.

    Useful for pre-populating cache or caching externally created instances.

    Args:
        resource_type: Type of resource
        logical_name: Logical name
        instance: Instance to cache
    """
    self._cache[(resource_type, logical_name)] = instance
    logger.debug(f"Manually cached {resource_type}[{logical_name}]")

FactoryNotFoundError

Bases: BindingResolverError

Factory not registered for resource type.

SimpleFactory

SimpleFactory(cls: type, **default_kwargs: Any)

Simple factory that creates instances of a class.

Example

from myapp import DatabaseConnection

resolver.register_factory(
    "databases",
    SimpleFactory(DatabaseConnection)
)

Initialize with a class to instantiate.

Parameters:

Name	Type	Description	Default
cls	type	Class to instantiate	required
**default_kwargs	Any	Default kwargs merged with config	{}

Methods:

Name	Description
create	Create an instance.

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def __init__(self, cls: type, **default_kwargs: Any):
    """Initialize with a class to instantiate.

    Args:
        cls: Class to instantiate
        **default_kwargs: Default kwargs merged with config
    """
    self._cls = cls
    self._defaults = default_kwargs

Functions

create

create(**config: Any) -> Any

Create an instance.

Parameters:

Name	Type	Description	Default
**config	Any	Configuration merged with defaults	{}

Returns:

Type	Description
Any	Created instance

Source code in packages/config/src/dataknobs_config/binding_resolver.py

def create(self, **config: Any) -> Any:
    """Create an instance.

    Args:
        **config: Configuration merged with defaults

    Returns:
        Created instance
    """
    merged = {**self._defaults, **config}
    return self._cls(**merged)

ConfigurableBase

Base class for objects that can be configured.

Classes that inherit from this can implement custom configuration loading logic.

Methods:

Name	Description
from_config	Create an instance from a configuration dictionary.

Functions

from_config classmethod

from_config(config: dict) -> ConfigurableBase

Create an instance from a configuration dictionary.

Parameters:

Name	Type	Description	Default
config	dict	Configuration dictionary	required

Returns:

Type	Description
ConfigurableBase	Instance of the class

Source code in packages/config/src/dataknobs_config/builders.py

@classmethod
def from_config(cls, config: dict) -> "ConfigurableBase":
    """Create an instance from a configuration dictionary.

    Args:
        config: Configuration dictionary

    Returns:
        Instance of the class
    """
    return cls(**config)

FactoryBase

Base class for factory objects.

Factories that inherit from this should implement the create method.

Methods:

Name	Description
create	Create an object from configuration.

Functions

create

create(**config: Any) -> Any

Create an object from configuration.

Parameters:

Name	Type	Description	Default
**config	Any	Configuration parameters	{}

Returns:

Type	Description
Any	Created object

Source code in packages/config/src/dataknobs_config/builders.py

def create(self, **config: Any) -> Any:
    """Create an object from configuration.

    Args:
        **config: Configuration parameters

    Returns:
        Created object
    """
    raise NotImplementedError("Subclasses must implement create method")

Config

Config(*sources: Union[str, Path, dict], **kwargs: Any)

A modular configuration system for composable settings.

Internally stores configurations as a dictionary of lists of atomic configuration dictionaries, organized by type.

Initialize a Config object from one or more sources.

Parameters:

Name	Type	Description	Default
*sources	Union[str, Path, dict]	Variable number of sources (file paths or dictionaries)	()
**kwargs	Any	Additional keyword arguments	{}

Methods:

Name	Description
from_file	Create a Config object from a file.
from_dict	Create a Config object from a dictionary.
load	Load configuration from a source.
get_types	Get all configuration types.
get_count	Get the count of configurations for a type.
get_names	Get all configuration names for a type.
get	Get a configuration by type and name/index.
set	Set a configuration by type and name/index.
resolve_reference	Resolve a string reference to a configuration.
build_reference	Build a string reference for a configuration.
merge	Merge another configuration into this one.
to_dict	Export configuration as a dictionary.
to_file	Save configuration to a file.
build_object	Build an object from a configuration reference.
clear_object_cache	Clear cached objects.
get_factory	Get a factory instance for a configuration.
register_factory	Register a factory instance for use in configurations.
unregister_factory	Unregister a factory.
get_registered_factories	Get all registered factories.
get_instance	Get an instance from a configuration.

Source code in packages/config/src/dataknobs_config/config.py

def __init__(self, *sources: Union[str, Path, dict], **kwargs: Any) -> None:
    """Initialize a Config object from one or more sources.

    Args:
        *sources: Variable number of sources (file paths or dictionaries)
        **kwargs: Additional keyword arguments
    """
    self._data: Dict[str, List[Dict[str, Any]]] = {}
    self._settings_manager = SettingsManager()
    self._reference_resolver = ReferenceResolver(self)
    self._environment_overrides = EnvironmentOverrides()
    self._object_builder = ObjectBuilder(self)
    self._variable_substitution = VariableSubstitution()
    self._registered_factories = Registry[Any](name="factories", enable_metrics=True)

    # Load sources
    for source in sources:
        self.load(source)

    # Apply environment overrides if enabled
    if kwargs.get("use_env", True):
        self._apply_environment_overrides()

Functions

from_file classmethod

from_file(path: Union[str, Path]) -> Config

Create a Config object from a file.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	Path to configuration file (YAML or JSON)	required

Returns:

Type	Description
Config	Config object

Source code in packages/config/src/dataknobs_config/config.py

@classmethod
def from_file(cls, path: Union[str, Path]) -> "Config":
    """Create a Config object from a file.

    Args:
        path: Path to configuration file (YAML or JSON)

    Returns:
        Config object
    """
    return cls(path)

from_dict classmethod

from_dict(data: dict) -> Config

Create a Config object from a dictionary.

Parameters:

Name	Type	Description	Default
data	dict	Configuration dictionary	required

Returns:

Type	Description
Config	Config object

Source code in packages/config/src/dataknobs_config/config.py

@classmethod
def from_dict(cls, data: dict) -> "Config":
    """Create a Config object from a dictionary.

    Args:
        data: Configuration dictionary

    Returns:
        Config object
    """
    return cls(data)

load

load(source: Union[str, Path, dict]) -> None

Load configuration from a source.

Parameters:

Name	Type	Description	Default
source	Union[str, Path, dict]	File path or dictionary	required

Source code in packages/config/src/dataknobs_config/config.py

def load(self, source: Union[str, Path, dict]) -> None:
    """Load configuration from a source.

    Args:
        source: File path or dictionary
    """
    if isinstance(source, dict):
        self._load_dict(source)
    elif isinstance(source, (str, Path)):
        self._load_file(source)
    else:
        raise ValidationError(f"Invalid source type: {type(source)}")

get_types

get_types() -> List[str]

Get all configuration types.

Returns:

Type	Description
List[str]	List of type names

Source code in packages/config/src/dataknobs_config/config.py

def get_types(self) -> List[str]:
    """Get all configuration types.

    Returns:
        List of type names
    """
    return list(self._data.keys())

get_count

get_count(type_name: str) -> int

Get the count of configurations for a type.

Parameters:

Name	Type	Description	Default
type_name	str	Type name	required

Returns:

Type	Description
int	Number of configurations

Source code in packages/config/src/dataknobs_config/config.py

def get_count(self, type_name: str) -> int:
    """Get the count of configurations for a type.

    Args:
        type_name: Type name

    Returns:
        Number of configurations
    """
    return len(self._data.get(type_name, []))

get_names

get_names(type_name: str) -> List[str]

Get all configuration names for a type.

Parameters:

Name	Type	Description	Default
type_name	str	Type name	required

Returns:

Type	Description
List[str]	List of configuration names

Source code in packages/config/src/dataknobs_config/config.py

def get_names(self, type_name: str) -> List[str]:
    """Get all configuration names for a type.

    Args:
        type_name: Type name

    Returns:
        List of configuration names
    """
    configs = self._data.get(type_name, [])
    return [config.get("name", str(i)) for i, config in enumerate(configs)]

get

get(type_name: str, name_or_index: Union[str, int] = 0) -> dict

Get a configuration by type and name/index.

Parameters:

Name	Type	Description	Default
type_name	str	Type name	required
name_or_index	Union[str, int]	Configuration name or index	0

Returns:

Type	Description
dict	Configuration dictionary

Source code in packages/config/src/dataknobs_config/config.py

def get(self, type_name: str, name_or_index: Union[str, int] = 0) -> dict:
    """Get a configuration by type and name/index.

    Args:
        type_name: Type name
        name_or_index: Configuration name or index

    Returns:
        Configuration dictionary
    """
    if type_name not in self._data:
        raise ConfigNotFoundError(f"Type not found: {type_name}")

    configs = self._data[type_name]

    if isinstance(name_or_index, int):
        # Handle index access
        try:
            return copy.deepcopy(configs[name_or_index])
        except IndexError as e:
            raise ConfigNotFoundError(f"Index out of range: {name_or_index}") from e
    else:
        # Handle name access
        for config in configs:
            if config.get("name") == name_or_index:
                return copy.deepcopy(config)
        raise ConfigNotFoundError(f"Configuration not found: {type_name}[{name_or_index}]")

set

set(type_name: str, name_or_index: Union[str, int], config: dict) -> None

Set a configuration by type and name/index.

Parameters:

Name	Type	Description	Default
type_name	str	Type name	required
name_or_index	Union[str, int]	Configuration name or index	required
config	dict	Configuration dictionary	required

Source code in packages/config/src/dataknobs_config/config.py

def set(self, type_name: str, name_or_index: Union[str, int], config: dict) -> None:
    """Set a configuration by type and name/index.

    Args:
        type_name: Type name
        name_or_index: Configuration name or index
        config: Configuration dictionary
    """
    if type_name not in self._data:
        self._data[type_name] = []

    configs = self._data[type_name]
    config = copy.deepcopy(config)

    # Normalize the configuration
    if isinstance(name_or_index, int):
        config = self._normalize_atomic_config(config, type_name, name_or_index)
        if name_or_index < len(configs):
            configs[name_or_index] = config
        elif name_or_index == len(configs):
            configs.append(config)
        else:
            raise ValidationError(f"Index out of range: {name_or_index}")
    else:
        # Find and replace by name
        config["name"] = name_or_index

        # Find the existing config to replace
        replaced = False
        for i, existing in enumerate(configs):
            if existing.get("name") == name_or_index:
                # Normalize with the correct index
                config = self._normalize_atomic_config(config, type_name, i)
                configs[i] = config
                replaced = True
                break

        if not replaced:
            # New config, add to end
            config = self._normalize_atomic_config(config, type_name, len(configs))
            configs.append(config)

resolve_reference

resolve_reference(ref: str) -> dict

Resolve a string reference to a configuration.

Parameters:

Name	Type	Description	Default
ref	str	String reference (e.g., "xref:foo[bar]")	required

Returns:

Type	Description
dict	Referenced configuration dictionary

Source code in packages/config/src/dataknobs_config/config.py

def resolve_reference(self, ref: str) -> dict:
    """Resolve a string reference to a configuration.

    Args:
        ref: String reference (e.g., "xref:foo[bar]")

    Returns:
        Referenced configuration dictionary
    """
    return self._reference_resolver.resolve(ref)

build_reference

build_reference(type_name: str, name_or_index: Union[str, int]) -> str

Build a string reference for a configuration.

Parameters:

Name	Type	Description	Default
type_name	str	Type name	required
name_or_index	Union[str, int]	Configuration name or index	required

Returns:

Type	Description
str	String reference

Source code in packages/config/src/dataknobs_config/config.py

def build_reference(self, type_name: str, name_or_index: Union[str, int]) -> str:
    """Build a string reference for a configuration.

    Args:
        type_name: Type name
        name_or_index: Configuration name or index

    Returns:
        String reference
    """
    return self._reference_resolver.build(type_name, name_or_index)

merge

merge(other: Config, precedence: str = 'first') -> None

Merge another configuration into this one.

Parameters:

Name	Type	Description	Default
other	Config	Configuration to merge	required
precedence	str	Precedence rule ("first" or "last")	'first'

Source code in packages/config/src/dataknobs_config/config.py

def merge(self, other: "Config", precedence: str = "first") -> None:
    """Merge another configuration into this one.

    Args:
        other: Configuration to merge
        precedence: Precedence rule ("first" or "last")
    """
    if precedence not in ["first", "last"]:
        raise ValidationError(f"Invalid precedence: {precedence}")

    # Merge settings first
    if precedence == "first":
        # Current settings take precedence
        other_settings = other._settings_manager._settings.copy()
        other_settings.update(self._settings_manager._settings)
        self._settings_manager._settings = other_settings
    else:
        # Other settings take precedence
        self._settings_manager._settings.update(other._settings_manager._settings)

    # Merge configurations
    for type_name in other.get_types():
        if type_name not in self._data:
            self._data[type_name] = []

        # Add all configurations from other
        for config in other._data[type_name]:
            # Check for duplicate names
            existing_names = self.get_names(type_name)
            if config.get("name") in existing_names:
                if precedence == "first":
                    # Skip if we already have this name
                    continue
                else:
                    # Replace existing with same name
                    name = config.get("name")
                    if name is not None:
                        self.set(type_name, name, config)
            else:
                # Add new configuration
                self._data[type_name].append(copy.deepcopy(config))

to_dict

to_dict() -> dict

Export configuration as a dictionary.

Returns:

Type	Description
dict	Configuration dictionary

Source code in packages/config/src/dataknobs_config/config.py

def to_dict(self) -> dict:
    """Export configuration as a dictionary.

    Returns:
        Configuration dictionary
    """
    result = copy.deepcopy(self._data)

    # Add settings if any
    settings = self._settings_manager.to_dict()
    if settings:
        result["settings"] = settings  # type: ignore[assignment]

    return result

to_file

to_file(path: Union[str, Path], format: str | None = None) -> None

Save configuration to a file.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	Output file path	required
format	str | None	Output format ("yaml" or "json"), auto-detected if not specified	None

Source code in packages/config/src/dataknobs_config/config.py

def to_file(self, path: Union[str, Path], format: str | None = None) -> None:
    """Save configuration to a file.

    Args:
        path: Output file path
        format: Output format ("yaml" or "json"), auto-detected if not specified
    """
    path = Path(path)

    # Auto-detect format from extension if not specified
    if format is None:
        suffix = path.suffix.lower()
        if suffix in [".yaml", ".yml"]:
            format = "yaml"
        elif suffix == ".json":
            format = "json"
        else:
            raise ValidationError(f"Cannot determine format from extension: {suffix}")

    data = self.to_dict()

    with open(path, "w") as f:
        if format == "yaml":
            yaml.safe_dump(data, f, default_flow_style=False, sort_keys=False)
        elif format == "json":
            json.dump(data, f, indent=2)
        else:
            raise ValidationError(f"Unsupported format: {format}")

build_object

build_object(ref: str, cache: bool = True, **kwargs: Any) -> Any

Build an object from a configuration reference.

Parameters:

Name	Type	Description	Default
ref	str	String reference to configuration	required
cache	bool	Whether to cache the built object	True
**kwargs	Any	Additional keyword arguments for construction	{}

Returns:

Type	Description
Any	Built object instance

Source code in packages/config/src/dataknobs_config/config.py

def build_object(self, ref: str, cache: bool = True, **kwargs: Any) -> Any:
    """Build an object from a configuration reference.

    Args:
        ref: String reference to configuration
        cache: Whether to cache the built object
        **kwargs: Additional keyword arguments for construction

    Returns:
        Built object instance
    """
    return self._object_builder.build(ref, cache=cache, **kwargs)

clear_object_cache

clear_object_cache(ref: str | None = None) -> None

Clear cached objects.

Parameters:

Name	Type	Description	Default
ref	str | None	Specific reference to clear, or None to clear all	None

Source code in packages/config/src/dataknobs_config/config.py

def clear_object_cache(self, ref: str | None = None) -> None:
    """Clear cached objects.

    Args:
        ref: Specific reference to clear, or None to clear all
    """
    self._object_builder.clear_cache(ref)

get_factory

get_factory(type_name: str, name_or_index: Union[str, int] = 0) -> Any

Get a factory instance for a configuration.

This method lazily instantiates and caches factory instances defined in configurations with a 'factory' attribute.

Parameters:

Name	Type	Description	Default
type_name	str	Type name	required
name_or_index	Union[str, int]	Configuration name or index	0

Returns:

Type	Description
Any	Factory instance

Raises:

Type	Description
ConfigError	If no factory is defined for the configuration

Source code in packages/config/src/dataknobs_config/config.py

def get_factory(self, type_name: str, name_or_index: Union[str, int] = 0) -> Any:
    """Get a factory instance for a configuration.

    This method lazily instantiates and caches factory instances defined
    in configurations with a 'factory' attribute.

    Args:
        type_name: Type name
        name_or_index: Configuration name or index

    Returns:
        Factory instance

    Raises:
        ConfigError: If no factory is defined for the configuration
    """
    # Get the configuration
    config = self.get(type_name, name_or_index)

    if "factory" not in config:
        raise ConfigError(
            f"No factory defined for {type_name}[{name_or_index}]. "
            f"Add 'factory' attribute to the configuration."
        )

    # Build a reference for caching
    ref = self.build_reference(type_name, name_or_index)
    factory_ref = f"{ref}.factory"

    # Check if factory is already cached
    cached = self._object_builder.get_cached(factory_ref)
    if cached is not None:
        return cached

    # Load and instantiate the factory
    factory_path = config["factory"]

    # Check if it's a registered factory first
    if self._registered_factories.has(factory_path):
        factory = self._registered_factories.get(factory_path)
    else:
        # Try to load as a module path
        factory_cls = self._object_builder._load_class(factory_path)

        # Create factory instance
        try:
            factory = factory_cls()
        except TypeError:
            # Factory might be a module-level function or callable class
            factory = factory_cls

    # Cache the factory instance
    self._object_builder._cache[factory_ref] = factory

    return factory

register_factory

register_factory(name: str, factory: Any) -> None

Register a factory instance for use in configurations.

Registered factories take precedence over module paths. This allows: 1. Cleaner configuration files (no module paths needed) 2. Runtime factory substitution (useful for testing) 3. Pre-configured factory instances

Parameters:

Name	Type	Description	Default
name	str	Name to register the factory under	required
factory	Any	Factory instance or class	required

Example

from dataknobs_config import Config

# Assuming you have a database_factory defined
config = Config()
config.register_factory("database", database_factory)
config.load({
    "databases": [{
        "name": "main",
        "factory": "database",  # Uses registered factory
        "backend": "postgres"
    }]
})

Note

If a factory name matches both a registered factory and a module path, the registered factory takes precedence.

Source code in packages/config/src/dataknobs_config/config.py

def register_factory(self, name: str, factory: Any) -> None:
    """Register a factory instance for use in configurations.

    Registered factories take precedence over module paths. This allows:
    1. Cleaner configuration files (no module paths needed)
    2. Runtime factory substitution (useful for testing)
    3. Pre-configured factory instances

    Args:
        name: Name to register the factory under
        factory: Factory instance or class

    Example:
        ```python
        from dataknobs_config import Config

        # Assuming you have a database_factory defined
        config = Config()
        config.register_factory("database", database_factory)
        config.load({
            "databases": [{
                "name": "main",
                "factory": "database",  # Uses registered factory
                "backend": "postgres"
            }]
        })
        ```

    Note:
        If a factory name matches both a registered factory and a module
        path, the registered factory takes precedence.
    """
    self._registered_factories.register(name, factory, allow_overwrite=True)
    logger.debug(f"Registered factory '{name}': {factory}")

unregister_factory

unregister_factory(name: str) -> None

Unregister a factory.

Parameters:

Name	Type	Description	Default
name	str	Name of the factory to unregister	required

Raises:

Type	Description
KeyError	If factory is not registered

Source code in packages/config/src/dataknobs_config/config.py

def unregister_factory(self, name: str) -> None:
    """Unregister a factory.

    Args:
        name: Name of the factory to unregister

    Raises:
        KeyError: If factory is not registered
    """
    try:
        self._registered_factories.unregister(name)
        logger.debug(f"Unregistered factory '{name}'")
    except NotFoundError as e:
        raise KeyError(f"Factory '{name}' is not registered") from e

get_registered_factories

get_registered_factories() -> Dict[str, Any]

Get all registered factories.

Returns:

Type	Description
Dict[str, Any]	Dictionary mapping factory names to factory instances

Source code in packages/config/src/dataknobs_config/config.py

def get_registered_factories(self) -> Dict[str, Any]:
    """Get all registered factories.

    Returns:
        Dictionary mapping factory names to factory instances
    """
    return self._registered_factories.copy()

get_instance

get_instance(
    type_name: str, name_or_index: Union[str, int] = 0, **kwargs: Any
) -> Any

Get an instance from a configuration.

This is a convenience method that combines get() and build_object(). If the configuration has a 'class' or 'factory' attribute, it will build and return an instance. Otherwise, it returns the config dict.

Parameters:

Name	Type	Description	Default
type_name	str	Type name	required
name_or_index	Union[str, int]	Configuration name or index	0
**kwargs	Any	Additional keyword arguments for construction	{}

Returns:

Type	Description
Any	Built instance or configuration dictionary

Source code in packages/config/src/dataknobs_config/config.py

def get_instance(
    self, type_name: str, name_or_index: Union[str, int] = 0, **kwargs: Any
) -> Any:
    """Get an instance from a configuration.

    This is a convenience method that combines get() and build_object().
    If the configuration has a 'class' or 'factory' attribute, it will
    build and return an instance. Otherwise, it returns the config dict.

    Args:
        type_name: Type name
        name_or_index: Configuration name or index
        **kwargs: Additional keyword arguments for construction

    Returns:
        Built instance or configuration dictionary
    """
    config = self.get(type_name, name_or_index)

    # If config has class or factory, build an object
    if "class" in config or "factory" in config:
        ref = self.build_reference(type_name, name_or_index)
        return self.build_object(ref, **kwargs)

    # Otherwise return the config itself
    return config

EnvironmentAwareConfig

EnvironmentAwareConfig(
    config: dict[str, Any],
    environment: EnvironmentConfig | None = None,
    app_name: str | None = None,
)

Configuration with environment-aware resource resolution.

Manages application configuration with support for: - Logical resource references that resolve per-environment - Late-binding of environment variables - Portable config storage (unresolved)

Attributes:

Name	Type	Description
environment	EnvironmentConfig	The EnvironmentConfig for resource resolution
app_name	str | None	Name of the loaded application (if any)

Initialize environment-aware configuration.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Application configuration dictionary	required
environment	EnvironmentConfig | None	Environment configuration for resource resolution. If None, auto-detects and loads environment.	None
app_name	str | None	Optional name for this application config	None

Methods:

Name	Description
load_app	Load an application configuration with environment bindings.
from_dict	Create from a configuration dictionary.
get	Get a value from the config.
resolve_for_build	Resolve configuration for object building.
get_portable_config	Get the portable (unresolved) configuration.
to_dict	Get the raw configuration dictionary.
with_environment	Create a new instance with a different environment.
get_resource	Get a resolved resource configuration.
get_setting	Get an environment setting.
__repr__	String representation.

Source code in packages/config/src/dataknobs_config/environment_aware.py

def __init__(
    self,
    config: dict[str, Any],
    environment: EnvironmentConfig | None = None,
    app_name: str | None = None,
):
    """Initialize environment-aware configuration.

    Args:
        config: Application configuration dictionary
        environment: Environment configuration for resource resolution.
                    If None, auto-detects and loads environment.
        app_name: Optional name for this application config
    """
    self._config = config
    self._environment = environment or EnvironmentConfig.load()
    self._app_name = app_name or config.get("name")

Attributes

environment property

environment: EnvironmentConfig

Get the current environment configuration.

environment_name property

environment_name: str

Get the current environment name.

app_name property

app_name: str | None

Get the application name.

Functions

load_app classmethod

load_app(
    app_name: str,
    app_dir: str | Path = "config/apps",
    env_dir: str | Path = "config/environments",
    environment: str | None = None,
) -> EnvironmentAwareConfig

Load an application configuration with environment bindings.

This is the primary entry point for loading configs in an environment-aware manner. Config files are loaded WITHOUT environment variable substitution (late binding).

Parameters:

Name	Type	Description	Default
app_name	str	Application/bot name (without .yaml extension)	required
app_dir	str | Path	Directory containing app configs	'config/apps'
env_dir	str | Path	Directory containing environment configs	'config/environments'
environment	str | None	Environment name, or None to auto-detect	None

Returns:

Type	Description
EnvironmentAwareConfig	EnvironmentAwareConfig with both app and environment loaded

Raises:

Type	Description
EnvironmentAwareConfigError	If app config not found or invalid

Source code in packages/config/src/dataknobs_config/environment_aware.py

@classmethod
def load_app(
    cls,
    app_name: str,
    app_dir: str | Path = "config/apps",
    env_dir: str | Path = "config/environments",
    environment: str | None = None,
) -> EnvironmentAwareConfig:
    """Load an application configuration with environment bindings.

    This is the primary entry point for loading configs in an
    environment-aware manner. Config files are loaded WITHOUT
    environment variable substitution (late binding).

    Args:
        app_name: Application/bot name (without .yaml extension)
        app_dir: Directory containing app configs
        env_dir: Directory containing environment configs
        environment: Environment name, or None to auto-detect

    Returns:
        EnvironmentAwareConfig with both app and environment loaded

    Raises:
        EnvironmentAwareConfigError: If app config not found or invalid
    """
    app_dir = Path(app_dir)
    env_config = EnvironmentConfig.load(environment, env_dir)

    # Find and load app config file
    config_path = cls._find_config_file(app_dir, app_name)
    if config_path is None:
        raise EnvironmentAwareConfigError(
            f"Application config not found: {app_name}.yaml in {app_dir}"
        )

    try:
        config = cls._load_file(config_path)
    except (yaml.YAMLError, json.JSONDecodeError) as e:
        raise EnvironmentAwareConfigError(
            f"Failed to parse app config {config_path}: {e}"
        ) from e
    except OSError as e:
        raise EnvironmentAwareConfigError(
            f"Failed to read app config {config_path}: {e}"
        ) from e

    logger.info(
        f"Loaded app config '{app_name}' for environment '{env_config.name}'"
    )

    return cls(
        config=config,
        environment=env_config,
        app_name=app_name,
    )

from_dict classmethod

from_dict(
    config: dict[str, Any],
    environment: str | None = None,
    env_dir: str | Path = "config/environments",
) -> EnvironmentAwareConfig

Create from a configuration dictionary.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Application configuration dictionary	required
environment	str | None	Environment name, or None to auto-detect	None
env_dir	str | Path	Directory containing environment configs	'config/environments'

Returns:

Type	Description
EnvironmentAwareConfig	EnvironmentAwareConfig instance

Source code in packages/config/src/dataknobs_config/environment_aware.py

@classmethod
def from_dict(
    cls,
    config: dict[str, Any],
    environment: str | None = None,
    env_dir: str | Path = "config/environments",
) -> EnvironmentAwareConfig:
    """Create from a configuration dictionary.

    Args:
        config: Application configuration dictionary
        environment: Environment name, or None to auto-detect
        env_dir: Directory containing environment configs

    Returns:
        EnvironmentAwareConfig instance
    """
    env_config = EnvironmentConfig.load(environment, env_dir)
    return cls(config=config, environment=env_config)

get

get(key: str, default: Any = None) -> Any

Get a value from the config.

Parameters:

Name	Type	Description	Default
key	str	Configuration key (supports dot notation for nested access)	required
default	Any	Default value if not found	None

Returns:

Type	Description
Any	Configuration value

Source code in packages/config/src/dataknobs_config/environment_aware.py

def get(self, key: str, default: Any = None) -> Any:
    """Get a value from the config.

    Args:
        key: Configuration key (supports dot notation for nested access)
        default: Default value if not found

    Returns:
        Configuration value
    """
    keys = key.split(".")
    value = self._config

    for k in keys:
        if isinstance(value, dict) and k in value:
            value = value[k]
        else:
            return default

    return copy.deepcopy(value)

resolve_for_build

resolve_for_build(
    config_key: str | None = None,
    resolve_resources: bool = True,
    resolve_env_vars: bool = True,
) -> dict[str, Any]

Resolve configuration for object building.

This is the late-binding resolution point where: 1. Logical resource names are resolved to concrete configs 2. Environment variables are substituted 3. Final merged configuration is returned

Call this method immediately before instantiating objects.

Parameters:

Name	Type	Description	Default
config_key	str | None	Specific config key to resolve, or None for root	None
resolve_resources	bool	Whether to resolve logical resource refs	True
resolve_env_vars	bool	Whether to substitute environment variables	True

Returns:

Type	Description
dict[str, Any]	Fully resolved configuration dictionary

Source code in packages/config/src/dataknobs_config/environment_aware.py

def resolve_for_build(
    self,
    config_key: str | None = None,
    resolve_resources: bool = True,
    resolve_env_vars: bool = True,
) -> dict[str, Any]:
    """Resolve configuration for object building.

    This is the late-binding resolution point where:
    1. Logical resource names are resolved to concrete configs
    2. Environment variables are substituted
    3. Final merged configuration is returned

    Call this method immediately before instantiating objects.

    Args:
        config_key: Specific config key to resolve, or None for root
        resolve_resources: Whether to resolve logical resource refs
        resolve_env_vars: Whether to substitute environment variables

    Returns:
        Fully resolved configuration dictionary
    """
    # Get the base configuration
    if config_key:
        config = self.get(config_key)
        if config is None:
            raise EnvironmentAwareConfigError(
                f"Config key not found: {config_key}"
            )
    else:
        config = copy.deepcopy(self._config)

    # Resolve logical resource references
    if resolve_resources:
        config = self._resolve_resource_refs(config)

    # Resolve environment variables (late binding)
    if resolve_env_vars:
        config = substitute_env_vars(config)

    return config

get_portable_config

get_portable_config() -> dict[str, Any]

Get the portable (unresolved) configuration.

Returns the configuration with: - Logical resource references intact - Environment variables as placeholders

This is the config that should be stored in databases for cross-environment portability.

Returns:

Type	Description
dict[str, Any]	Unresolved configuration dictionary

Source code in packages/config/src/dataknobs_config/environment_aware.py

def get_portable_config(self) -> dict[str, Any]:
    """Get the portable (unresolved) configuration.

    Returns the configuration with:
    - Logical resource references intact
    - Environment variables as placeholders

    This is the config that should be stored in databases
    for cross-environment portability.

    Returns:
        Unresolved configuration dictionary
    """
    return copy.deepcopy(self._config)

to_dict

to_dict() -> dict[str, Any]

Get the raw configuration dictionary.

Alias for get_portable_config().

Returns:

Type	Description
dict[str, Any]	Configuration dictionary

Source code in packages/config/src/dataknobs_config/environment_aware.py

def to_dict(self) -> dict[str, Any]:
    """Get the raw configuration dictionary.

    Alias for get_portable_config().

    Returns:
        Configuration dictionary
    """
    return self.get_portable_config()

with_environment

with_environment(
    environment: str | EnvironmentConfig,
    env_dir: str | Path = "config/environments",
) -> EnvironmentAwareConfig

Create a new instance with a different environment.

Useful for testing or multi-environment scenarios.

Parameters:

Name	Type	Description	Default
environment	str | EnvironmentConfig	Environment name or EnvironmentConfig instance	required
env_dir	str | Path	Directory containing environment configs (if name provided)	'config/environments'

Returns:

Type	Description
EnvironmentAwareConfig	New EnvironmentAwareConfig with the specified environment

Source code in packages/config/src/dataknobs_config/environment_aware.py

def with_environment(
    self,
    environment: str | EnvironmentConfig,
    env_dir: str | Path = "config/environments",
) -> EnvironmentAwareConfig:
    """Create a new instance with a different environment.

    Useful for testing or multi-environment scenarios.

    Args:
        environment: Environment name or EnvironmentConfig instance
        env_dir: Directory containing environment configs (if name provided)

    Returns:
        New EnvironmentAwareConfig with the specified environment
    """
    if isinstance(environment, str):
        env_config = EnvironmentConfig.load(environment, env_dir)
    else:
        env_config = environment

    return EnvironmentAwareConfig(
        config=copy.deepcopy(self._config),
        environment=env_config,
        app_name=self._app_name,
    )

get_resource

get_resource(
    resource_type: str,
    logical_name: str,
    defaults: dict[str, Any] | None = None,
) -> dict[str, Any]

Get a resolved resource configuration.

Convenience method to directly access environment resources.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource	required
logical_name	str	Logical name of resource	required
defaults	dict[str, Any] | None	Default values if resource not found	None

Returns:

Type	Description
dict[str, Any]	Resolved resource configuration

Source code in packages/config/src/dataknobs_config/environment_aware.py

def get_resource(
    self,
    resource_type: str,
    logical_name: str,
    defaults: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Get a resolved resource configuration.

    Convenience method to directly access environment resources.

    Args:
        resource_type: Type of resource
        logical_name: Logical name of resource
        defaults: Default values if resource not found

    Returns:
        Resolved resource configuration
    """
    return self._environment.get_resource(resource_type, logical_name, defaults)

get_setting

get_setting(key: str, default: Any = None) -> Any

Get an environment setting.

Parameters:

Name	Type	Description	Default
key	str	Setting key	required
default	Any	Default value if not found	None

Returns:

Type	Description
Any	Setting value

Source code in packages/config/src/dataknobs_config/environment_aware.py

def get_setting(self, key: str, default: Any = None) -> Any:
    """Get an environment setting.

    Args:
        key: Setting key
        default: Default value if not found

    Returns:
        Setting value
    """
    return self._environment.get_setting(key, default)

__repr__

__repr__() -> str

String representation.

Source code in packages/config/src/dataknobs_config/environment_aware.py

def __repr__(self) -> str:
    """String representation."""
    return (
        f"EnvironmentAwareConfig(app={self._app_name!r}, "
        f"environment={self._environment.name!r})"
    )

EnvironmentAwareConfigError

Bases: Exception

Error related to environment-aware configuration.

EnvironmentConfig dataclass

EnvironmentConfig(
    name: str,
    resources: dict[str, dict[str, dict[str, Any]]] = dict(),
    settings: dict[str, Any] = dict(),
    description: str = "",
)

Environment-specific configuration and resource bindings.

Manages the mapping from logical resource names to concrete implementations for a specific deployment environment.

Attributes:

Name	Type	Description
name	str	Environment name (e.g., "development", "staging", "production")
resources	dict[str, dict[str, dict[str, Any]]]	Nested dict of {resource_type: {logical_name: config}}
settings	dict[str, Any]	Environment-wide settings (log levels, feature flags, etc.)
description	str	Optional description of the environment

Methods:

Name	Description
detect_environment	Detect current environment from env vars or indicators.
load	Load environment configuration from file.
from_dict	Create EnvironmentConfig from a dictionary.
get_resource	Get concrete config for a logical resource.
has_resource	Check if a resource exists.
get_setting	Get an environment-wide setting.
get_resource_types	Get all resource types in this environment.
get_resource_names	Get all resource names for a type.
to_dict	Convert to dictionary representation.
merge	Merge another environment config into this one.

Functions

detect_environment classmethod

detect_environment() -> str

Detect current environment from env vars or indicators.

Checks in order: 1. DATAKNOBS_ENVIRONMENT env var 2. Common cloud indicators (AWS_EXECUTION_ENV, etc.) 3. Default to "development"

Returns:

Type	Description
str	Detected environment name (lowercase)

Source code in packages/config/src/dataknobs_config/environment_config.py

@classmethod
def detect_environment(cls) -> str:
    """Detect current environment from env vars or indicators.

    Checks in order:
    1. DATAKNOBS_ENVIRONMENT env var
    2. Common cloud indicators (AWS_EXECUTION_ENV, etc.)
    3. Default to "development"

    Returns:
        Detected environment name (lowercase)
    """
    # Explicit setting takes precedence
    if env := os.environ.get("DATAKNOBS_ENVIRONMENT"):
        return env.lower()

    # AWS Lambda or ECS
    if os.environ.get("AWS_EXECUTION_ENV"):
        # Could be Lambda, ECS, etc.
        env_tier = os.environ.get("ENVIRONMENT", "production")
        return env_tier.lower()

    # AWS ECS Fargate
    if os.environ.get("ECS_CONTAINER_METADATA_URI"):
        env_tier = os.environ.get("ENVIRONMENT", "production")
        return env_tier.lower()

    # Kubernetes
    if os.environ.get("KUBERNETES_SERVICE_HOST"):
        env_tier = os.environ.get("ENVIRONMENT", "production")
        return env_tier.lower()

    # Google Cloud Run
    if os.environ.get("K_SERVICE"):
        env_tier = os.environ.get("ENVIRONMENT", "production")
        return env_tier.lower()

    # Azure Functions
    if os.environ.get("FUNCTIONS_WORKER_RUNTIME"):
        env_tier = os.environ.get("ENVIRONMENT", "production")
        return env_tier.lower()

    # Default to development
    return "development"

load classmethod

load(
    environment: str | None = None,
    config_dir: str | Path = "config/environments",
) -> EnvironmentConfig

Load environment configuration from file.

Parameters:

Name	Type	Description	Default
environment	str | None	Environment name, or None to auto-detect	None
config_dir	str | Path	Directory containing environment config files	'config/environments'

Returns:

Type	Description
EnvironmentConfig	Loaded EnvironmentConfig instance

Raises:

Type	Description
EnvironmentConfigError	If config file is invalid

Source code in packages/config/src/dataknobs_config/environment_config.py

@classmethod
def load(
    cls,
    environment: str | None = None,
    config_dir: str | Path = "config/environments",
) -> EnvironmentConfig:
    """Load environment configuration from file.

    Args:
        environment: Environment name, or None to auto-detect
        config_dir: Directory containing environment config files

    Returns:
        Loaded EnvironmentConfig instance

    Raises:
        EnvironmentConfigError: If config file is invalid
    """
    if environment is None:
        environment = cls.detect_environment()

    config_dir = Path(config_dir)
    config_path = cls._find_config_file(config_dir, environment)

    if config_path is None:
        # Return empty config for environments without config files
        logger.debug(
            f"No environment config found for '{environment}' in {config_dir}, "
            "using empty configuration"
        )
        return cls(name=environment)

    try:
        data = cls._load_file(config_path)
    except (yaml.YAMLError, json.JSONDecodeError) as e:
        raise EnvironmentConfigError(
            f"Failed to parse environment config {config_path}: {e}"
        ) from e
    except OSError as e:
        raise EnvironmentConfigError(
            f"Failed to read environment config {config_path}: {e}"
        ) from e

    return cls(
        name=data.get("name", environment),
        resources=data.get("resources", {}),
        settings=data.get("settings", {}),
        description=data.get("description", ""),
    )

from_dict classmethod

from_dict(data: dict[str, Any]) -> EnvironmentConfig

Create EnvironmentConfig from a dictionary.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Configuration dictionary	required

Returns:

Type	Description
EnvironmentConfig	EnvironmentConfig instance

Source code in packages/config/src/dataknobs_config/environment_config.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> EnvironmentConfig:
    """Create EnvironmentConfig from a dictionary.

    Args:
        data: Configuration dictionary

    Returns:
        EnvironmentConfig instance
    """
    return cls(
        name=data.get("name", "unknown"),
        resources=data.get("resources", {}),
        settings=data.get("settings", {}),
        description=data.get("description", ""),
    )

get_resource

get_resource(
    resource_type: str,
    logical_name: str,
    defaults: dict[str, Any] | None = None,
) -> dict[str, Any]

Get concrete config for a logical resource.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource ("databases", "vector_stores", etc.)	required
logical_name	str	Logical name referenced in app config	required
defaults	dict[str, Any] | None	Default config values if resource not found	None

Returns:

Type	Description
dict[str, Any]	Concrete configuration for the resource

Raises:

Type	Description
ResourceNotFoundError	If resource not found and no defaults provided

Source code in packages/config/src/dataknobs_config/environment_config.py

def get_resource(
    self,
    resource_type: str,
    logical_name: str,
    defaults: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Get concrete config for a logical resource.

    Args:
        resource_type: Type of resource ("databases", "vector_stores", etc.)
        logical_name: Logical name referenced in app config
        defaults: Default config values if resource not found

    Returns:
        Concrete configuration for the resource

    Raises:
        ResourceNotFoundError: If resource not found and no defaults provided
    """
    type_resources = self.resources.get(resource_type, {})

    if logical_name in type_resources:
        # Copy to avoid mutation
        config = type_resources[logical_name].copy()

        # Apply defaults for missing keys
        if defaults:
            for key, value in defaults.items():
                config.setdefault(key, value)

        return config

    if defaults is not None:
        return defaults.copy()

    raise ResourceNotFoundError(
        f"Resource '{logical_name}' of type '{resource_type}' "
        f"not found in environment '{self.name}'"
    )

has_resource

has_resource(resource_type: str, logical_name: str) -> bool

Check if a resource exists.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource	required
logical_name	str	Logical name of resource	required

Returns:

Type	Description
bool	True if resource exists

Source code in packages/config/src/dataknobs_config/environment_config.py

def has_resource(self, resource_type: str, logical_name: str) -> bool:
    """Check if a resource exists.

    Args:
        resource_type: Type of resource
        logical_name: Logical name of resource

    Returns:
        True if resource exists
    """
    return logical_name in self.resources.get(resource_type, {})

get_setting

get_setting(key: str, default: Any = None) -> Any

Get an environment-wide setting.

Parameters:

Name	Type	Description	Default
key	str	Setting key	required
default	Any	Default value if not found	None

Returns:

Type	Description
Any	Setting value

Source code in packages/config/src/dataknobs_config/environment_config.py

def get_setting(self, key: str, default: Any = None) -> Any:
    """Get an environment-wide setting.

    Args:
        key: Setting key
        default: Default value if not found

    Returns:
        Setting value
    """
    return self.settings.get(key, default)

get_resource_types

get_resource_types() -> list[str]

Get all resource types in this environment.

Returns:

Type	Description
list[str]	List of resource type names

Source code in packages/config/src/dataknobs_config/environment_config.py

def get_resource_types(self) -> list[str]:
    """Get all resource types in this environment.

    Returns:
        List of resource type names
    """
    return list(self.resources.keys())

get_resource_names

get_resource_names(resource_type: str) -> list[str]

Get all resource names for a type.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource	required

Returns:

Type	Description
list[str]	List of logical resource names

Source code in packages/config/src/dataknobs_config/environment_config.py

def get_resource_names(self, resource_type: str) -> list[str]:
    """Get all resource names for a type.

    Args:
        resource_type: Type of resource

    Returns:
        List of logical resource names
    """
    return list(self.resources.get(resource_type, {}).keys())

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary representation.

Returns:

Type	Description
dict[str, Any]	Dictionary representation of environment config

Source code in packages/config/src/dataknobs_config/environment_config.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary representation.

    Returns:
        Dictionary representation of environment config
    """
    result: dict[str, Any] = {"name": self.name}

    if self.description:
        result["description"] = self.description

    if self.settings:
        result["settings"] = self.settings.copy()

    if self.resources:
        result["resources"] = {
            rtype: {name: config.copy() for name, config in resources.items()}
            for rtype, resources in self.resources.items()
        }

    return result

merge

merge(other: EnvironmentConfig) -> EnvironmentConfig

Merge another environment config into this one.

The other config's values take precedence.

Parameters:

Name	Type	Description	Default
other	EnvironmentConfig	Environment config to merge	required

Returns:

Type	Description
EnvironmentConfig	New merged EnvironmentConfig

Source code in packages/config/src/dataknobs_config/environment_config.py

def merge(self, other: EnvironmentConfig) -> EnvironmentConfig:
    """Merge another environment config into this one.

    The other config's values take precedence.

    Args:
        other: Environment config to merge

    Returns:
        New merged EnvironmentConfig
    """
    # Deep merge resources
    merged_resources: dict[str, dict[str, dict[str, Any]]] = {}

    # Start with self's resources
    for rtype, resources in self.resources.items():
        merged_resources[rtype] = {
            name: config.copy() for name, config in resources.items()
        }

    # Merge in other's resources
    for rtype, resources in other.resources.items():
        if rtype not in merged_resources:
            merged_resources[rtype] = {}
        for name, config in resources.items():
            if name in merged_resources[rtype]:
                # Merge configs
                merged_resources[rtype][name].update(config)
            else:
                merged_resources[rtype][name] = config.copy()

    # Merge settings
    merged_settings = self.settings.copy()
    merged_settings.update(other.settings)

    return EnvironmentConfig(
        name=other.name,
        resources=merged_resources,
        settings=merged_settings,
        description=other.description or self.description,
    )

EnvironmentConfigError

Bases: Exception

Error related to environment configuration.

ResourceBinding dataclass

ResourceBinding(name: str, resource_type: str, config: dict[str, Any])

A binding from logical name to concrete implementation.

Attributes:

Name	Type	Description
name	str	Logical name of the resource
resource_type	str	Type of resource (e.g., "databases", "vector_stores")
config	dict[str, Any]	Concrete configuration for the resource

ResourceNotFoundError

Bases: EnvironmentConfigError, KeyError

Resource not found in environment configuration.

ConfigNotFoundError

ConfigNotFoundError(
    message: str,
    context: Dict[str, Any] | None = None,
    details: Dict[str, Any] | None = None,
)

Bases: NotFoundError

Raised when a requested configuration is not found.

Source code in packages/common/src/dataknobs_common/exceptions.py

def __init__(
    self,
    message: str,
    context: Dict[str, Any] | None = None,
    details: Dict[str, Any] | None = None,
):
    """Initialize the exception with optional context.

    Args:
        message: Error message
        context: Optional context dictionary
        details: Optional details dictionary (merged with context)
    """
    super().__init__(message)
    # Support both context and details parameters
    # Details takes precedence if both are provided
    self.context = details or context or {}
    # Alias for FSM-style compatibility
    self.details = self.context

InvalidReferenceError

InvalidReferenceError(
    message: str,
    context: Dict[str, Any] | None = None,
    details: Dict[str, Any] | None = None,
)

Bases: ValidationError

Raised when a configuration reference is invalid.

Source code in packages/common/src/dataknobs_common/exceptions.py

def __init__(
    self,
    message: str,
    context: Dict[str, Any] | None = None,
    details: Dict[str, Any] | None = None,
):
    """Initialize the exception with optional context.

    Args:
        message: Error message
        context: Optional context dictionary
        details: Optional details dictionary (merged with context)
    """
    super().__init__(message)
    # Support both context and details parameters
    # Details takes precedence if both are provided
    self.context = details or context or {}
    # Alias for FSM-style compatibility
    self.details = self.context

InheritableConfigLoader

InheritableConfigLoader(config_dir: str | Path | None = None)

Configuration loader with inheritance support.

Loads YAML/JSON configuration files with support for configuration inheritance via an extends field. Child configurations override parent values through deep merge.

Attributes:

Name	Type	Description
config_dir		Directory containing configuration files
cache		Configuration cache for performance

Initialize configuration loader.

Parameters:

Name	Type	Description	Default
config_dir	str | Path | None	Directory containing configuration files. If None, uses ./configs	None

Methods:

Name	Description
load	Load and resolve configuration with inheritance.
load_from_file	Load configuration from a specific file path.
clear_cache	Clear configuration cache.
list_available	List all available configuration files.
validate	Validate a configuration file.

Source code in packages/config/src/dataknobs_config/inheritance.py

def __init__(self, config_dir: str | Path | None = None):
    """Initialize configuration loader.

    Args:
        config_dir: Directory containing configuration files.
                   If None, uses ./configs
    """
    self.config_dir = Path(config_dir) if config_dir else Path("./configs")
    self._cache: dict[str, dict[str, Any]] = {}
    self._loading: set[str] = set()  # Track configs being loaded to detect cycles

Functions

load

load(
    name: str, use_cache: bool = True, substitute_vars: bool = True
) -> dict[str, Any]

Load and resolve configuration with inheritance.

Parameters:

Name	Type	Description	Default
name	str	Configuration name (without extension)	required
use_cache	bool	Whether to use cached configuration if available	True
substitute_vars	bool	Whether to substitute environment variables	True

Returns:

Type	Description
dict[str, Any]	Resolved configuration dictionary

Raises:

Type	Description
InheritanceError	If config not found, cycle detected, or other error

Example

loader = InheritableConfigLoader("./configs")
config = loader.load("my-domain")

Source code in packages/config/src/dataknobs_config/inheritance.py

def load(
    self,
    name: str,
    use_cache: bool = True,
    substitute_vars: bool = True,
) -> dict[str, Any]:
    """Load and resolve configuration with inheritance.

    Args:
        name: Configuration name (without extension)
        use_cache: Whether to use cached configuration if available
        substitute_vars: Whether to substitute environment variables

    Returns:
        Resolved configuration dictionary

    Raises:
        InheritanceError: If config not found, cycle detected, or other error

    Example:
        ```python
        loader = InheritableConfigLoader("./configs")
        config = loader.load("my-domain")
        ```
    """
    # Check cache
    if use_cache and name in self._cache:
        logger.debug(f"Using cached config: {name}")
        return self._cache[name]

    # Detect circular inheritance
    if name in self._loading:
        raise InheritanceError(f"Circular inheritance detected: {name}")

    self._loading.add(name)

    try:
        # Load raw configuration
        raw_config = self._load_file(name)

        # Handle inheritance
        if raw_config.get("extends"):
            parent_name = raw_config["extends"]
            logger.debug(f"Config '{name}' extends '{parent_name}'")

            # Load parent configuration (recursively handles inheritance)
            parent_config = self.load(parent_name, use_cache=use_cache, substitute_vars=False)

            # Deep merge: child overrides parent
            raw_config = deep_merge(parent_config, raw_config)

            # Remove extends field from final config
            raw_config.pop("extends", None)

        # Substitute environment variables
        if substitute_vars:
            raw_config = substitute_env_vars(raw_config)

        # Cache the result
        self._cache[name] = raw_config
        logger.info(f"Loaded configuration: {name}")

        return raw_config

    finally:
        self._loading.discard(name)

load_from_file

load_from_file(
    filepath: str | Path, substitute_vars: bool = True
) -> dict[str, Any]

Load configuration from a specific file path.

This method bypasses the config_dir and loads directly from the path. Inheritance is resolved relative to the file's directory.

Parameters:

Name	Type	Description	Default
filepath	str | Path	Path to configuration file	required
substitute_vars	bool	Whether to substitute environment variables	True

Returns:

Type	Description
dict[str, Any]	Resolved configuration dictionary

Raises:

Type	Description
InheritanceError	If file not found or other error

Source code in packages/config/src/dataknobs_config/inheritance.py

def load_from_file(
    self,
    filepath: str | Path,
    substitute_vars: bool = True,
) -> dict[str, Any]:
    """Load configuration from a specific file path.

    This method bypasses the config_dir and loads directly from the path.
    Inheritance is resolved relative to the file's directory.

    Args:
        filepath: Path to configuration file
        substitute_vars: Whether to substitute environment variables

    Returns:
        Resolved configuration dictionary

    Raises:
        InheritanceError: If file not found or other error
    """
    filepath = Path(filepath)

    if not filepath.exists():
        raise InheritanceError(f"Configuration file not found: {filepath}")

    # Temporarily change config_dir to file's directory for inheritance
    old_config_dir = self.config_dir
    self.config_dir = filepath.parent

    try:
        return self.load(filepath.stem, use_cache=False, substitute_vars=substitute_vars)
    finally:
        self.config_dir = old_config_dir

clear_cache

clear_cache(name: str | None = None) -> None

Clear configuration cache.

Parameters:

Name	Type	Description	Default
name	str | None	Specific config to clear, or None to clear all	None

Source code in packages/config/src/dataknobs_config/inheritance.py

def clear_cache(self, name: str | None = None) -> None:
    """Clear configuration cache.

    Args:
        name: Specific config to clear, or None to clear all
    """
    if name:
        self._cache.pop(name, None)
        logger.debug(f"Cleared cache for: {name}")
    else:
        self._cache.clear()
        logger.debug("Cleared all cached configurations")

list_available

list_available() -> list[str]

List all available configuration files.

Returns:

Type	Description
list[str]	List of configuration names (without extensions)

Source code in packages/config/src/dataknobs_config/inheritance.py

def list_available(self) -> list[str]:
    """List all available configuration files.

    Returns:
        List of configuration names (without extensions)
    """
    if not self.config_dir.exists():
        return []

    configs = set()
    for pattern in ["*.yaml", "*.yml", "*.json"]:
        for file in self.config_dir.glob(pattern):
            if file.is_file():
                configs.add(file.stem)

    return sorted(configs)

validate

validate(name: str) -> tuple[bool, str | None]

Validate a configuration file.

Parameters:

Name	Type	Description	Default
name	str	Configuration name	required

Returns:

Type	Description
tuple[bool, str | None]	Tuple of (is_valid, error_message)

Source code in packages/config/src/dataknobs_config/inheritance.py

def validate(self, name: str) -> tuple[bool, str | None]:
    """Validate a configuration file.

    Args:
        name: Configuration name

    Returns:
        Tuple of (is_valid, error_message)
    """
    try:
        self.load(name, use_cache=False)
        return True, None
    except InheritanceError as e:
        return False, str(e)
    except ValueError as e:
        return False, str(e)

InheritanceError

Bases: Exception

Error during configuration inheritance resolution.

VariableSubstitution

Handles environment variable substitution in configuration values.

Supports patterns: - ${VAR} - Replace with environment variable VAR, error if not found - ${VAR:default} - Replace with VAR or use default if not found - ${VAR:-default} - Same as above (bash-style)

Methods:

Name	Description
substitute	Recursively substitute environment variables in a value.
has_variables	Check if a value contains environment variable references.

Functions

substitute

substitute(value: Any) -> Any

Recursively substitute environment variables in a value.

Parameters:

Name	Type	Description	Default
value	Any	Value to process (can be string, dict, list, or other)	required

Returns:

Type	Description
Any	Value with environment variables substituted

Raises:

Type	Description
ValueError	If a required environment variable is not found

Source code in packages/config/src/dataknobs_config/substitution.py

def substitute(self, value: Any) -> Any:
    """Recursively substitute environment variables in a value.

    Args:
        value: Value to process (can be string, dict, list, or other)

    Returns:
        Value with environment variables substituted

    Raises:
        ValueError: If a required environment variable is not found
    """
    if isinstance(value, str):
        return self._substitute_string(value)
    elif isinstance(value, dict):
        return self._substitute_dict(value)
    elif isinstance(value, list):
        return self._substitute_list(value)
    else:
        # Return other types unchanged
        return value

has_variables

has_variables(value: Any) -> bool

Check if a value contains environment variable references.

Parameters:

Name	Type	Description	Default
value	Any	Value to check	required

Returns:

Type	Description
bool	True if value contains ${...} patterns

Source code in packages/config/src/dataknobs_config/substitution.py

def has_variables(self, value: Any) -> bool:
    """Check if a value contains environment variable references.

    Args:
        value: Value to check

    Returns:
        True if value contains ${...} patterns
    """
    if isinstance(value, str):
        return bool(self.VAR_PATTERN.search(value))
    elif isinstance(value, dict):
        return any(self.has_variables(v) for v in value.values())
    elif isinstance(value, list):
        return any(self.has_variables(item) for item in value)
    else:
        return False

Functions

deep_merge

deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]

Deep merge two dictionaries.

Recursively merges override into base, with override values taking precedence. Nested dictionaries are merged recursively; all other types are replaced.

Parameters:

Name	Type	Description	Default
base	dict[str, Any]	Base dictionary (values used when not overridden)	required
override	dict[str, Any]	Override dictionary (takes precedence)	required

Returns:

Type	Description
dict[str, Any]	New merged dictionary

Example

base = {"a": 1, "nested": {"x": 10, "y": 20}} override = {"a": 2, "nested": {"y": 25, "z": 30}} deep_merge(base, override) {'a': 2, 'nested': {'x': 10, 'y': 25, 'z': 30}}

Source code in packages/config/src/dataknobs_config/inheritance.py

def deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
    """Deep merge two dictionaries.

    Recursively merges override into base, with override values taking precedence.
    Nested dictionaries are merged recursively; all other types are replaced.

    Args:
        base: Base dictionary (values used when not overridden)
        override: Override dictionary (takes precedence)

    Returns:
        New merged dictionary

    Example:
        >>> base = {"a": 1, "nested": {"x": 10, "y": 20}}
        >>> override = {"a": 2, "nested": {"y": 25, "z": 30}}
        >>> deep_merge(base, override)
        {'a': 2, 'nested': {'x': 10, 'y': 25, 'z': 30}}
    """
    result = base.copy()

    for key, value in override.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            # Recursively merge nested dicts
            result[key] = deep_merge(result[key], value)
        else:
            # Override takes precedence
            result[key] = value

    return result

load_config_with_inheritance

load_config_with_inheritance(
    filepath: str | Path, substitute_vars: bool = True
) -> dict[str, Any]

Convenience function to load a config file with inheritance.

Parameters:

Name	Type	Description	Default
filepath	str | Path	Path to configuration file	required
substitute_vars	bool	Whether to substitute environment variables	True

Returns:

Type	Description
dict[str, Any]	Resolved configuration dictionary

Example

config = load_config_with_inheritance("configs/my-domain.yaml")

Source code in packages/config/src/dataknobs_config/inheritance.py

def load_config_with_inheritance(
    filepath: str | Path,
    substitute_vars: bool = True,
) -> dict[str, Any]:
    """Convenience function to load a config file with inheritance.

    Args:
        filepath: Path to configuration file
        substitute_vars: Whether to substitute environment variables

    Returns:
        Resolved configuration dictionary

    Example:
        ```python
        config = load_config_with_inheritance("configs/my-domain.yaml")
        ```
    """
    loader = InheritableConfigLoader()
    return loader.load_from_file(filepath, substitute_vars=substitute_vars)

substitute_env_vars

substitute_env_vars(data: Any) -> Any

Recursively substitute environment variables in configuration.

Supports formats: - ${VAR_NAME}: Required variable, raises error if not set - ${VAR_NAME:default_value}: Optional with default

Substitution applies to both dict keys and values, list items, and top-level strings. Non-string dict keys (integers, booleans, etc.) pass through unchanged.

Also expands ~ in string keys and values after substitution using os.path.expanduser(), which leaves non-path strings (URLs, connection strings) intact.

Parameters:

Name	Type	Description	Default
data	Any	Configuration data (dict, list, string, or primitive)	required

Returns:

Type	Description
Any	Data with environment variables substituted

Raises:

Type	Description
ValueError	If required environment variable not set

Example

os.environ["MY_VAR"] = "hello" substitute_env_vars({"key": "\({MY_VAR}", "default": "\))}"

Source code in packages/config/src/dataknobs_config/inheritance.py

def substitute_env_vars(data: Any) -> Any:
    """Recursively substitute environment variables in configuration.

    Supports formats:
    - ${VAR_NAME}: Required variable, raises error if not set
    - ${VAR_NAME:default_value}: Optional with default

    Substitution applies to both dict keys and values, list items,
    and top-level strings. Non-string dict keys (integers, booleans, etc.)
    pass through unchanged.

    Also expands ~ in string keys and values after substitution using
    os.path.expanduser(), which leaves non-path strings (URLs,
    connection strings) intact.

    Args:
        data: Configuration data (dict, list, string, or primitive)

    Returns:
        Data with environment variables substituted

    Raises:
        ValueError: If required environment variable not set

    Example:
        >>> os.environ["MY_VAR"] = "hello"
        >>> substitute_env_vars({"key": "${MY_VAR}", "default": "${MISSING:world}"})
        {'key': 'hello', 'default': 'world'}
    """
    if isinstance(data, dict):
        return {
            _substitute_string(k) if isinstance(k, str) else k: substitute_env_vars(v)
            for k, v in data.items()
        }
    elif isinstance(data, list):
        return [substitute_env_vars(item) for item in data]
    elif isinstance(data, str):
        return _substitute_string(data)
    else:
        return data

substitute_template_vars

substitute_template_vars(
    data: Any,
    variables: dict[str, Any],
    *,
    preserve_missing: bool = True,
    type_cast: bool = True,
) -> Any

Recursively substitute {{var}} placeholders in configuration data.

Walks through nested dicts and lists, replacing {{var}} placeholders with values from the variables map. Supports type preservation for entire-value placeholders.

Parameters:

Name	Type	Description	Default
data	Any	Configuration data (dict, list, string, or primitive)	required
variables	dict[str, Any]	Dictionary of variable names to values	required
preserve_missing	bool	If True, leave {{var}} intact when var not in variables. If False, replace missing vars with empty string.	True
type_cast	bool	If True, preserve Python types for entire-value placeholders. "{{count}}" with count=10 becomes int 10, not string "10". If False, always return strings for substituted values.	True

Returns:

Type	Description
Any	Data with template variables substituted

Example

variables = {"name": "Math Tutor", "count": 10, "items": ["a", "b"]} data = {"title": "{{name}}", "max": "{{count}}", "desc": "Has {{count}} items"} substitute_template_vars(data, variables)

Type preservation for entire-value placeholders

substitute_template_vars({"list": "{{items}}"}, variables)

Missing variables preserved by default

substitute_template_vars("Hello {{name}}, {{unknown}}", {"name": "World"}) 'Hello World, {{unknown}}'

Source code in packages/config/src/dataknobs_config/template_vars.py

def substitute_template_vars(
    data: Any,
    variables: dict[str, Any],
    *,
    preserve_missing: bool = True,
    type_cast: bool = True,
) -> Any:
    """Recursively substitute {{var}} placeholders in configuration data.

    Walks through nested dicts and lists, replacing {{var}} placeholders
    with values from the variables map. Supports type preservation for
    entire-value placeholders.

    Args:
        data: Configuration data (dict, list, string, or primitive)
        variables: Dictionary of variable names to values
        preserve_missing: If True, leave {{var}} intact when var not in variables.
                         If False, replace missing vars with empty string.
        type_cast: If True, preserve Python types for entire-value placeholders.
                  "{{count}}" with count=10 becomes int 10, not string "10".
                  If False, always return strings for substituted values.

    Returns:
        Data with template variables substituted

    Example:
        >>> variables = {"name": "Math Tutor", "count": 10, "items": ["a", "b"]}
        >>> data = {"title": "{{name}}", "max": "{{count}}", "desc": "Has {{count}} items"}
        >>> substitute_template_vars(data, variables)
        {'title': 'Math Tutor', 'max': 10, 'desc': 'Has 10 items'}

        # Type preservation for entire-value placeholders
        >>> substitute_template_vars({"list": "{{items}}"}, variables)
        {'list': ['a', 'b']}

        # Missing variables preserved by default
        >>> substitute_template_vars("Hello {{name}}, {{unknown}}", {"name": "World"})
        'Hello World, {{unknown}}'
    """
    if isinstance(data, dict):
        return {
            k: substitute_template_vars(
                v, variables, preserve_missing=preserve_missing, type_cast=type_cast
            )
            for k, v in data.items()
        }
    elif isinstance(data, list):
        return [
            substitute_template_vars(
                item, variables, preserve_missing=preserve_missing, type_cast=type_cast
            )
            for item in data
        ]
    elif isinstance(data, str):
        return _substitute_string(data, variables, preserve_missing, type_cast)
    else:
        return data