Config Package API Reference¶

📖 Also see: Auto-generated API Reference - Complete documentation from source code docstrings

dataknobs_config.config ¶

Core Config class implementation.

Classes:

Name	Description
`Config`	A modular configuration system for composable settings.

Classes¶

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

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¶

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)

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)

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)

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)

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)

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

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_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

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

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_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_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())

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)}")

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))

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}")

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)

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)

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}")

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

dataknobs_config.references ¶

String reference system for cross-referencing configurations.

Classes:

Name	Description
`ReferenceResolver`	Handles parsing and resolution of string references.

Classes¶

ReferenceResolver ¶

ReferenceResolver(config_instance: Config)

Handles parsing and resolution of string references.

Reference format: xref:[]

Examples:

xref:foo[bar] - Named reference
xref:foo[0] - Index reference
xref:foo[-1] - Last item reference
xref:foo - First/only item reference

Initialize the reference resolver.

Parameters:

Name	Type	Description	Default
`config_instance`	`Config`	The Config instance to resolve references against	required

Methods:

Name	Description
`build`	Build a string reference for a configuration.
`is_reference`	Check if a value is a string reference.
`parse_reference`	Parse a string reference into its components.
`resolve`	Resolve a string reference to a configuration.

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

def __init__(self, config_instance: "Config") -> None:
    """Initialize the reference resolver.

    Args:
        config_instance: The Config instance to resolve references against
    """
    self._config = config_instance
    self._resolving: set[str] = set()  # Track references being resolved to detect cycles

Functions¶

build ¶

build(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/references.py

def build(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
    """
    if isinstance(name_or_index, int):
        if name_or_index == 0:
            # First item can be referenced without selector
            return f"xref:{type_name}"
        else:
            return f"xref:{type_name}[{name_or_index}]"
    else:
        return f"xref:{type_name}[{name_or_index}]"

is_reference ¶

is_reference(value: Any) -> bool

Check if a value is a string reference.

Parameters:

Name	Type	Description	Default
`value`	`Any`	Value to check	required

Returns:

Type	Description
`bool`	True if value is a string reference

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

def is_reference(self, value: Any) -> bool:
    """Check if a value is a string reference.

    Args:
        value: Value to check

    Returns:
        True if value is a string reference
    """
    if not isinstance(value, str):
        return False
    return value.startswith("xref:")

parse_reference ¶

parse_reference(ref: str) -> Tuple[str, Union[str, int]]

Parse a string reference into its components.

Parameters:

Name	Type	Description	Default
`ref`	`str`	String reference	required

Returns:

Type	Description
`Tuple[str, Union[str, int]]`	Tuple of (type_name, name_or_index)

Raises:

Type	Description
`InvalidReferenceError`	If reference format is invalid

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

def parse_reference(self, ref: str) -> Tuple[str, Union[str, int]]:
    """Parse a string reference into its components.

    Args:
        ref: String reference

    Returns:
        Tuple of (type_name, name_or_index)

    Raises:
        InvalidReferenceError: If reference format is invalid
    """
    if not ref.startswith("xref:"):
        raise InvalidReferenceError(f"Reference must start with 'xref:': {ref}")

    match = self.REFERENCE_PATTERN.match(ref)
    if not match:
        raise InvalidReferenceError(f"Invalid reference format: {ref}")

    type_name = match.group(1)
    selector = match.group(2)

    if not type_name:
        raise InvalidReferenceError(f"Missing type in reference: {ref}")

    # Parse selector
    name_or_index: Union[str, int]
    if selector is None:
        # No selector means first item (index 0)
        name_or_index = 0
    elif selector.isdigit() or (selector.startswith("-") and selector[1:].isdigit()):
        # Numeric selector (index)
        name_or_index = int(selector)
    else:
        # String selector (name)
        name_or_index = selector

    return type_name, name_or_index

resolve ¶

resolve(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

Raises:

Type	Description
`InvalidReferenceError`	If reference format is invalid
`ConfigNotFoundError`	If referenced configuration doesn't exist

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

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

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

    Returns:
        Referenced configuration dictionary

    Raises:
        InvalidReferenceError: If reference format is invalid
        ConfigNotFoundError: If referenced configuration doesn't exist
    """
    # Check for circular references
    if ref in self._resolving:
        raise InvalidReferenceError(f"Circular reference detected: {ref}")

    # Parse the reference
    type_name, name_or_index = self.parse_reference(ref)

    # Track that we're resolving this reference
    self._resolving.add(ref)

    try:
        # Get the configuration
        config = self._config.get(type_name, name_or_index)

        # Resolve any nested references
        config = self._resolve_nested_references(config)

        return config
    finally:
        # Remove from resolving set
        self._resolving.discard(ref)

dataknobs_config.environment ¶

Environment variable override system.

Classes:

Name	Description
`EnvironmentOverrides`	Handles environment variable overrides for configurations.

Classes¶

EnvironmentOverrides ¶

EnvironmentOverrides(prefix: str | None = None)

Handles environment variable overrides for configurations.

Environment variable format: DATAKNOBS_

Examples:

DATAKNOBS_FOO__BAR__PARAM -> xref:foo[bar].param
DATAKNOBS_FOO__0__PARAM -> xref:foo[0].param

Initialize the environment override handler.

Parameters:

Name	Type	Description	Default
`prefix`	`str \| None`	Custom environment variable prefix (default: DATAKNOBS_)	`None`

Methods:

Name	Description
`get_overrides`	Get all environment variable overrides.
`parse_env_reference`	Parse a reference string with attribute.
`reference_to_env_var`	Convert a reference string to an environment variable name.

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

def __init__(self, prefix: str | None = None) -> None:
    """Initialize the environment override handler.

    Args:
        prefix: Custom environment variable prefix (default: DATAKNOBS_)
    """
    self.prefix = prefix or self.ENV_PREFIX

Functions¶

get_overrides ¶

get_overrides() -> Dict[str, Any]

Get all environment variable overrides.

Returns:

Type	Description
`Dict[str, Any]`	Dictionary mapping references to override values

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

def get_overrides(self) -> Dict[str, Any]:
    """Get all environment variable overrides.

    Returns:
        Dictionary mapping references to override values
    """
    overrides = {}

    for key, value in os.environ.items():
        if key.startswith(self.prefix):
            try:
                # Parse the environment variable
                ref = self._env_var_to_reference(key)

                # Convert value to appropriate type
                typed_value = self._parse_value(value)

                overrides[ref] = typed_value
            except Exception:
                # Skip invalid environment variables
                continue

    return overrides

parse_env_reference ¶

parse_env_reference(ref: str) -> Tuple[str, Union[str, int], str | None]

Parse a reference string with attribute.

Parameters:

Name	Type	Description	Default
`ref`	`str`	Reference string with attribute (e.g., "xref:foo[bar].param")	required

Returns:

Type	Description
`Tuple[str, Union[str, int], str \| None]`	Tuple of (type_name, name_or_index, attribute)

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

def parse_env_reference(self, ref: str) -> Tuple[str, Union[str, int], str | None]:
    """Parse a reference string with attribute.

    Args:
        ref: Reference string with attribute (e.g., "xref:foo[bar].param")

    Returns:
        Tuple of (type_name, name_or_index, attribute)
    """
    # Check if reference includes attribute
    if "." in ref and not ref.endswith("]"):
        # Split reference and attribute
        base_ref, attribute = ref.rsplit(".", 1)
    else:
        base_ref = ref
        attribute = None

    # Parse base reference
    pattern = re.compile(r"^xref:([a-zA-Z_][a-zA-Z0-9_]*)?(?:\[([^\]]+)\])?$")
    match = pattern.match(base_ref)

    if not match:
        # Try parsing the full reference if it includes attribute
        if "." in ref:
            parts = ref.split(".")
            if parts[0].startswith("xref:"):
                # Handle xref:type.attribute format
                type_part = parts[0][5:]  # Remove 'xref:' prefix

                # Check for selector in type part
                if "[" in type_part:
                    type_name, selector = type_part.split("[", 1)
                    selector = selector.rstrip("]")
                else:
                    type_name = type_part
                    selector = None

                attribute = ".".join(parts[1:])
            else:
                raise InvalidReferenceError(f"Invalid reference format: {ref}")
        else:
            raise InvalidReferenceError(f"Invalid reference format: {ref}")
    else:
        type_name = match.group(1)
        selector = match.group(2)

    if not type_name:
        raise InvalidReferenceError(f"Missing type in reference: {ref}")

    # Parse selector
    name_or_index: Union[str, int]
    if selector is None:
        name_or_index = 0
    elif selector.isdigit() or (selector.startswith("-") and selector[1:].isdigit()):
        name_or_index = int(selector)
    else:
        name_or_index = selector

    return type_name, name_or_index, attribute

reference_to_env_var ¶

reference_to_env_var(ref: str, attribute: str | None = None) -> str

Convert a reference string to an environment variable name.

Parameters:

Name	Type	Description	Default
`ref`	`str`	Reference string (e.g., "xref:foo[bar]")	required
`attribute`	`str \| None`	Attribute name (if not included in ref)	`None`

Returns:

Type	Description
`str`	Environment variable name

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

def reference_to_env_var(self, ref: str, attribute: str | None = None) -> str:
    """Convert a reference string to an environment variable name.

    Args:
        ref: Reference string (e.g., "xref:foo[bar]")
        attribute: Attribute name (if not included in ref)

    Returns:
        Environment variable name
    """
    # Parse reference components
    type_name, name_or_index, attr = self.parse_env_reference(ref)

    if attribute:
        attr = attribute

    if not attr:
        raise InvalidReferenceError(f"Attribute required for environment variable: {ref}")

    # Build environment variable name
    type_part = type_name.upper()
    selector_part = str(name_or_index).upper()
    attr_part = attr.upper().replace(".", self.ENV_SEPARATOR)

    return f"{self.prefix}{type_part}{self.ENV_SEPARATOR}{selector_part}{self.ENV_SEPARATOR}{attr_part}"

dataknobs_config.settings ¶

Global settings and defaults management.

Classes:

Name	Description
`SettingsManager`	Manages global settings, defaults, and path resolution.

Classes¶

SettingsManager ¶

SettingsManager()

Manages global settings, defaults, and path resolution.

Settings attributes

config_root: Root directory for configuration files
global_root: Global root directory for all types
.global_root: Type-specific root directory
path_resolution_attributes: List of attributes to resolve as paths
.: Type-specific defaults
: Global defaults

Initialize the settings manager.

Methods:

Name	Description
`apply_defaults`	Apply default values to a configuration.
`get_setting`	Get a setting value.
`load_settings`	Load settings from a dictionary.
`resolve_paths`	Resolve relative paths in configuration.
`set_setting`	Set a setting value.
`to_dict`	Export settings as a dictionary.

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

def __init__(self) -> None:
    """Initialize the settings manager."""
    self._settings: Dict[str, Any] = {}

Functions¶

apply_defaults ¶

apply_defaults(config: dict, type_name: str) -> dict

Apply default values to a configuration.

Parameters:

Name	Type	Description	Default
`config`	`dict`	Atomic configuration dictionary	required
`type_name`	`str`	Type of the configuration	required

Returns:

Type	Description
`dict`	Configuration with defaults applied

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

def apply_defaults(self, config: dict, type_name: str) -> dict:
    """Apply default values to a configuration.

    Args:
        config: Atomic configuration dictionary
        type_name: Type of the configuration

    Returns:
        Configuration with defaults applied
    """
    result = copy.deepcopy(config)

    # Apply global defaults
    for key, value in self._settings.items():
        if "." not in key and key not in result:
            # Global default for any attribute
            if key not in ["config_root", "global_root", "path_resolution_attributes"]:
                result[key] = copy.deepcopy(value)

    # Apply type-specific defaults
    type_prefix = f"{type_name}."
    for key, value in self._settings.items():
        if key.startswith(type_prefix):
            attr_name = key[len(type_prefix) :]
            if attr_name not in result:
                result[attr_name] = copy.deepcopy(value)

    return result

get_setting ¶

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

Get a setting value.

Parameters:

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

Returns:

Type	Description
`Any`	Setting value or default

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

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

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

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

load_settings ¶

load_settings(settings: dict) -> None

Load settings from a dictionary.

Parameters:

Name	Type	Description	Default
`settings`	`dict`	Settings dictionary	required

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

def load_settings(self, settings: dict) -> None:
    """Load settings from a dictionary.

    Args:
        settings: Settings dictionary
    """
    # Merge with existing settings (first seen takes precedence)
    for key, value in settings.items():
        if key not in self._settings:
            self._settings[key] = value

resolve_paths ¶

resolve_paths(config: dict, type_name: str) -> dict

Resolve relative paths in configuration.

Parameters:

Name	Type	Description	Default
`config`	`dict`	Atomic configuration dictionary	required
`type_name`	`str`	Type of the configuration	required

Returns:

Type	Description
`dict`	Configuration with resolved paths

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

def resolve_paths(self, config: dict, type_name: str) -> dict:
    """Resolve relative paths in configuration.

    Args:
        config: Atomic configuration dictionary
        type_name: Type of the configuration

    Returns:
        Configuration with resolved paths
    """
    result = copy.deepcopy(config)

    # Get path resolution attributes
    path_attrs = self.get_setting("path_resolution_attributes", [])
    if not path_attrs:
        return result

    # Determine base path for resolution
    base_path = self._get_base_path(type_name)

    # Resolve paths for matching attributes
    for attr_pattern in path_attrs:
        # Check if it's a regex pattern (starts with / and ends with /)
        if (
            isinstance(attr_pattern, str)
            and attr_pattern.startswith("/")
            and attr_pattern.endswith("/")
        ):
            # Regex pattern
            pattern_str = attr_pattern[1:-1]  # Remove the / delimiters
            try:
                pattern = re.compile(pattern_str)
                # Apply to all matching attributes
                for key in list(result.keys()):
                    if pattern.match(key):
                        result[key] = self._resolve_path(result[key], base_path)
            except re.error:
                # Invalid regex, skip
                continue
        elif "." in attr_pattern:
            # Type-specific attribute
            type_prefix, attr_name = attr_pattern.split(".", 1)
            if type_prefix == type_name:
                # Check if attr_name is a regex
                if attr_name.startswith("/") and attr_name.endswith("/"):
                    pattern_str = attr_name[1:-1]
                    try:
                        pattern = re.compile(pattern_str)
                        for key in list(result.keys()):
                            if pattern.match(key):
                                result[key] = self._resolve_path(result[key], base_path)
                    except re.error:
                        continue
                elif attr_name in result:
                    result[attr_name] = self._resolve_path(result[attr_name], base_path)
        # Global attribute (exact match)
        elif attr_pattern in result:
            result[attr_pattern] = self._resolve_path(result[attr_pattern], base_path)

    return result

set_setting ¶

set_setting(key: str, value: Any) -> None

Set a setting value.

Parameters:

Name	Type	Description	Default
`key`	`str`	Setting key	required
`value`	`Any`	Setting value	required

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

def set_setting(self, key: str, value: Any) -> None:
    """Set a setting value.

    Args:
        key: Setting key
        value: Setting value
    """
    self._settings[key] = value

to_dict ¶

to_dict() -> dict

Export settings as a dictionary.

Returns:

Type	Description
`dict`	Settings dictionary

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

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

    Returns:
        Settings dictionary
    """
    return copy.deepcopy(self._settings)

dataknobs_config.builders ¶

Optional object construction and caching functionality.

Classes:

Name	Description
`ObjectBuilder`	Handles object construction from configurations.
`ConfigurableBase`	Base class for objects that can be configured.
`FactoryBase`	Base class for factory objects.

Classes¶

ObjectBuilder ¶

ObjectBuilder(config_instance: Config)

Handles object construction from configurations.

Supports

Direct class instantiation via 'class' attribute
Factory pattern via 'factory' attribute
Object caching

Initialize the object builder.

Parameters:

Name	Type	Description	Default
`config_instance`	`Config`	The Config instance to build objects from	required

Methods:

Name	Description
`build`	Build an object from a configuration reference.
`clear_cache`	Clear cached objects.
`get_cached`	Get a cached object without building.

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

def __init__(self, config_instance: "Config") -> None:
    """Initialize the object builder.

    Args:
        config_instance: The Config instance to build objects from
    """
    self._config = config_instance
    self._cache: Dict[str, Any] = {}

Functions¶

build ¶

build(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

Raises:

Type	Description
`ConfigError`	If object cannot be built

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

def build(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

    Raises:
        ConfigError: If object cannot be built
    """
    # Check cache first
    if cache and ref in self._cache:
        return self._cache[ref]

    # Resolve the configuration
    config = self._config.resolve_reference(ref)

    # Build the object
    obj = self._build_from_config(config, **kwargs)

    # Cache if requested
    if cache:
        self._cache[ref] = obj

    return obj

clear_cache ¶

clear_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/builders.py

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

    Args:
        ref: Specific reference to clear, or None to clear all
    """
    if ref:
        self._cache.pop(ref, None)
    else:
        self._cache.clear()

get_cached ¶

get_cached(ref: str) -> Any | None

Get a cached object without building.

Parameters:

Name	Type	Description	Default
`ref`	`str`	String reference	required

Returns:

Type	Description
`Any \| None`	Cached object or None

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

def get_cached(self, ref: str) -> Any | None:
    """Get a cached object without building.

    Args:
        ref: String reference

    Returns:
        Cached object or None
    """
    return self._cache.get(ref)

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

dataknobs_config.exceptions ¶

Custom exceptions for the config package.

This module defines exception types for the config package, built on the common exception framework from dataknobs_common.

Classes:

Name	Description
`ConfigNotFoundError`	Raised when a requested configuration is not found.
`InvalidReferenceError`	Raised when a configuration reference is invalid.
`FileNotFoundError`	Raised when a referenced configuration file is not found.

Attributes:

Name	Type	Description
`ConfigError`
`ValidationError`

Attributes¶

ConfigError `module-attribute` ¶

ConfigError = ConfigurationError

ValidationError `module-attribute` ¶

ValidationError = ValidationError

Classes¶

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

FileNotFoundError ¶

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

Bases: NotFoundError

Raised when a referenced configuration file 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

Environment-Aware Configuration¶

dataknobs_config.environment_config ¶

Environment-specific configuration and resource bindings.

This module provides environment-aware configuration management for deploying the same application across different environments (development, staging, production) where infrastructure differs.

Key features: - Environment detection (via DATAKNOBS_ENVIRONMENT or cloud indicators) - Resource bindings (logical names -> concrete implementations) - Environment-wide settings management

Example

# Auto-detect environment
env = EnvironmentConfig.load()

# Or specify explicitly
env = EnvironmentConfig.load("production", config_dir="config/environments")

# Get concrete config for a logical resource
db_config = env.get_resource("databases", "conversations")
# Returns: {"backend": "postgres", "connection_string": "..."}

Environment file format (config/environments/production.yaml):

name: production
description: AWS production environment

settings:
  log_level: INFO
  enable_metrics: true

resources:
  databases:
    default:
      backend: postgres
      connection_string: ${DATABASE_URL}
    conversations:
      backend: postgres
      connection_string: ${DATABASE_URL}
      table: conversations

  vector_stores:
    default:
      backend: pgvector
      connection_string: ${DATABASE_URL}

Classes:

Name	Description
`EnvironmentConfig`	Environment-specific configuration and resource bindings.
`ResourceBinding`	A binding from logical name to concrete implementation.
`EnvironmentConfigError`	Error related to environment configuration.
`ResourceNotFoundError`	Resource not found in environment configuration.

Classes¶

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.
`from_dict`	Create EnvironmentConfig from a dictionary.
`get_resource`	Get concrete config for a logical resource.
`get_resource_names`	Get all resource names for a type.
`get_resource_types`	Get all resource types in this environment.
`get_setting`	Get an environment-wide setting.
`has_resource`	Check if a resource exists.
`load`	Load environment configuration from file.
`merge`	Merge another environment config into this one.
`to_dict`	Convert to dictionary representation.

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"

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}'"
    )

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())

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_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)

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, {})

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", ""),
    )

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,
    )

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

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

EnvironmentConfigError ¶

Bases: Exception

Error related to environment configuration.

ResourceNotFoundError ¶

Bases: EnvironmentConfigError, KeyError

Resource not found in environment configuration.

dataknobs_config.environment_aware ¶

Environment-aware configuration with late-binding resource resolution.

This module provides the EnvironmentAwareConfig class that supports: - Logical resource references resolved per-environment - Late-binding of environment variables (at instantiation time, not load time) - Separation of portable app config from infrastructure bindings

Example

# Load with auto-detected environment
config = EnvironmentAwareConfig.load_app(
    "my-bot",
    app_dir="config/apps",
    env_dir="config/environments"
)

# Get resolved config for object building (late binding happens here)
resolved = config.resolve_for_build()

# Get portable config for storage (no env vars resolved)
portable = config.get_portable_config()

App config format (config/apps/my-bot.yaml):

name: my-bot
version: "1.0.0"

bot:
  llm:
    $resource: default
    type: llm_providers
    temperature: 0.7

  conversation_storage:
    $resource: conversations
    type: databases

Classes:

Name	Description
`EnvironmentAwareConfig`	Configuration with environment-aware resource resolution.
`EnvironmentAwareConfigError`	Error related to environment-aware configuration.

Classes¶

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
`__repr__`	String representation.
`from_dict`	Create from a configuration dictionary.
`get`	Get a value from the config.
`get_portable_config`	Get the portable (unresolved) configuration.
`get_resource`	Get a resolved resource configuration.
`get_setting`	Get an environment setting.
`load_app`	Load an application configuration with environment bindings.
`resolve_for_build`	Resolve configuration for object building.
`to_dict`	Get the raw configuration dictionary.
`with_environment`	Create a new instance with a different environment.

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¶

app_name `property` ¶

app_name: str | None

Get the application name.

environment `property` ¶

environment: EnvironmentConfig

Get the current environment configuration.

environment_name `property` ¶

environment_name: str

Get the current environment name.

Functions¶

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})"
    )

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)

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)

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)

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,
    )

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

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,
    )

EnvironmentAwareConfigError ¶

Bases: Exception

Error related to environment-aware configuration.

dataknobs_config.binding_resolver ¶

Configuration binding resolver for resource instantiation.

This module provides the ConfigBindingResolver class that resolves logical resource references to concrete instances using registered factories.

Example

from dataknobs_config import EnvironmentConfig, ConfigBindingResolver

# Load environment config
env = EnvironmentConfig.load("production")

# Create resolver
resolver = ConfigBindingResolver(env)

# Register factories for resource types
resolver.register_factory("databases", DatabaseFactory())
resolver.register_factory("vector_stores", VectorStoreFactory())

# Resolve a logical reference to a concrete instance
db = resolver.resolve("databases", "conversations")

# Or with async factories
vector_store = await resolver.resolve_async("vector_stores", "knowledge")

Classes:

Name	Description
`ConfigBindingResolver`	Resolves logical resource bindings to concrete instances.
`SimpleFactory`	Simple factory that creates instances of a class.
`CallableFactory`	Factory that wraps a callable.
`AsyncCallableFactory`	Factory that wraps an async callable.
`BindingResolverError`	Error during resource binding resolution.
`FactoryNotFoundError`	Factory not registered for resource type.

Classes¶

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

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¶

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

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

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))

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())

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

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

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}")

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

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}")

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)

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)

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`	Sync create is not supported for async factories.
`create_async`	Create an instance asynchronously.

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 ¶

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()."
    )

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)

BindingResolverError ¶

Bases: Exception

Error during resource binding resolution.

FactoryNotFoundError ¶

Bases: BindingResolverError

Factory not registered for resource type.

Config Package API Reference¶

dataknobs_config.config ¶

Classes¶

Config ¶

Functions¶

build_object ¶

build_reference ¶

clear_object_cache ¶

from_dict classmethod ¶

from_file classmethod ¶

get ¶

get_count ¶

get_factory ¶

get_instance ¶

get_names ¶

get_registered_factories ¶

get_types ¶

load ¶

merge ¶

register_factory ¶

resolve_reference ¶

set ¶

to_dict ¶

to_file ¶

unregister_factory ¶

dataknobs_config.references ¶

Classes¶

ReferenceResolver ¶

Functions¶

build ¶

is_reference ¶

parse_reference ¶

resolve ¶

dataknobs_config.environment ¶

Classes¶

EnvironmentOverrides ¶

Functions¶

get_overrides ¶

parse_env_reference ¶

reference_to_env_var ¶

dataknobs_config.settings ¶

Classes¶

SettingsManager ¶

Functions¶

apply_defaults ¶

get_setting ¶

load_settings ¶

resolve_paths ¶

set_setting ¶

to_dict ¶

dataknobs_config.builders ¶

Classes¶

ObjectBuilder ¶

Functions¶

build ¶

clear_cache ¶

get_cached ¶

ConfigurableBase ¶

Functions¶

from_config classmethod ¶

FactoryBase ¶

Functions¶

create ¶

dataknobs_config.exceptions ¶

Attributes¶

ConfigError module-attribute ¶

ValidationError module-attribute ¶

Classes¶

ConfigNotFoundError ¶

InvalidReferenceError ¶

FileNotFoundError ¶

Environment-Aware Configuration¶

dataknobs_config.environment_config ¶

Classes¶

EnvironmentConfig dataclass ¶

Functions¶

detect_environment classmethod ¶

from_dict classmethod ¶

get_resource ¶

get_resource_names ¶

from_dict `classmethod` ¶

from_file `classmethod` ¶

from_config `classmethod` ¶

ConfigError `module-attribute` ¶

ValidationError `module-attribute` ¶

EnvironmentConfig `dataclass` ¶

detect_environment `classmethod` ¶

from_dict `classmethod` ¶

load `classmethod` ¶

ResourceBinding `dataclass` ¶

app_name `property` ¶

environment `property` ¶

environment_name `property` ¶

repr ¶

from_dict `classmethod` ¶

load_app `classmethod` ¶

environment `property` ¶

resolve_async `async` ¶

create_async `async` ¶