DataKnobs Config

A modular, reusable configuration system for composable settings with environment variable overrides, file loading, and optional object construction helpers.

Features

• Modular Configuration: Organize configurations by type with atomic configuration units
• Multiple Input Formats: Load from YAML, JSON files, or Python dictionaries
• Composable: Reference other configurations and compose complex setups
• Environment Overrides: Override any configuration value via environment variables
• Path Resolution: Automatically resolve relative paths to absolute
• Object Construction: Optional helpers to build objects from configurations
• Defaults Management: Global and type-specific default values
• Caching: Cache constructed objects for efficiency

Installation

pip install dataknobs-config

Quick Start

from dataknobs_config import Config

# Load from dictionary
config = Config({
    "database": [
        {"name": "primary", "host": "localhost", "port": 5432},
        {"name": "secondary", "host": "backup.local", "port": 5433}
    ],
    "cache": [
        {"name": "redis", "host": "localhost", "port": 6379}
    ]
})

# Access configurations
primary_db = config.get("database", "primary")
print(primary_db["host"])  # localhost

# Load from file
config = Config.from_file("config.yaml")

# Load from multiple sources
config = Config("base.yaml", "overrides.json", {"extra": [...]})

Core Concepts

Atomic Configurations

Each configuration is an "atomic" unit - a dictionary of settings for a single object:

{
    "name": "primary",      # Optional, auto-generated if not provided
    "type": "database",     # Optional, inferred from parent key
    "host": "localhost",
    "port": 5432,
    # ... any other attributes
}

Configuration Structure

Internally, configurations are organized by type:

{
    "database": [           # Type name
        {...},              # Atomic config 1
        {...}               # Atomic config 2
    ],
    "cache": [
        {...}               # Atomic config
    ],
    "settings": {           # Special type for global settings
        "config_root": "/app/config",
        "default_timeout": 30
    }
}

String References (xref)

Reference other configurations using the xref format:

config = Config({
    "database": [
        {"name": "primary", "host": "db.example.com"}
    ],
    "api": [
        {
            "name": "main",
            "database": "xref:database[primary]"  # Reference
        }
    ]
})

# Resolve references
api = config.resolve_reference("xref:api[main]")
print(api["database"]["host"])  # db.example.com

Reference Formats

• xref:type[name] - Reference by name
• xref:type[0] - Reference by index
• xref:type[-1] - Reference last item
• xref:type - Reference first/only item

Environment Variable Overrides

Override any configuration value using environment variables:

export DATAKNOBS_DATABASE__PRIMARY__HOST=prod.example.com
export DATAKNOBS_DATABASE__PRIMARY__PORT=5433
export DATAKNOBS_CACHE__REDIS__TTL=7200

config = Config({
    "database": [{"name": "primary", "host": "localhost", "port": 5432}],
    "cache": [{"name": "redis", "ttl": 3600}]
})

# Environment variables automatically override values
db = config.get("database", "primary")
print(db["host"])  # prod.example.com
print(db["port"])  # 5433 (converted to int)

Environment Variable Format

• Pattern: DATAKNOBS_<TYPE>__<NAME_OR_INDEX>__<ATTRIBUTE>
• Nested attributes: DATAKNOBS_DATABASE__0__CONNECTION__TIMEOUT
• Automatic type conversion for integers, floats, and booleans

File References

Reference external configuration files using the @ prefix:

# main.yaml
database:
  - "@database/primary.yaml"    # Load from file
  - "@database/secondary.yaml"

settings:
  config_root: /app/config       # Base path for relative references

Global Settings and Defaults

Configure global settings and defaults in the special settings section:

config = Config({
    "database": [{"name": "db1"}],
    "settings": {
        # Paths
        "config_root": "/app/config",           # Base path for "@"-prefixed config file references
        "global_root": "/app",                   # Base for path resolution (settings.path_resolution_attributes)
        "database.global_root": "/app/db",       # Type-specific base for path resolution
        
        # Path resolution (supports exact names and regex patterns)
        "path_resolution_attributes": [
            "config_path",                       # Exact match for all types
            "database.data_dir",                 # Exact match for database type only
            "/.*_path$/",                        # Regex: all attributes ending with "_path"
            "cache./.*_dir$/"                    # Regex: cache type attributes ending with "_dir"
        ],
        
        # Defaults
        "default_timeout": 30,                   # Global default
        "database.default_pool_size": 10        # Type-specific default
    }
})

Path Resolution

Automatically resolve relative paths to absolute:

config = Config({
    "database": [{
        "name": "db1",
        "data_dir": "./data",              # Relative path
        "backup_dir": "/abs/path"          # Absolute path unchanged
    }],
    "settings": {
        "global_root": "/app",              # Base for path resolution
        "path_resolution_attributes": ["data_dir", "backup_dir"]
    }
})

db = config.get("database", "db1")
print(db["data_dir"])     # /app/data (resolved)
print(db["backup_dir"])   # /abs/path (unchanged)

Object Construction (Optional)

Build objects directly from configurations:

# Using class attribute
config = Config({
    "database": [{
        "name": "primary",
        "class": "myapp.database.PostgreSQL",
        "host": "localhost",
        "port": 5432
    }]
})

# Build object
db = config.build_object("xref:database[primary]")
# Returns instance of myapp.database.PostgreSQL

# Using factory pattern
config = Config({
    "cache": [{
        "name": "redis",
        "factory": "myapp.cache.CacheFactory",
        "type": "redis",
        "host": "localhost"
    }]
})

cache = config.build_object("xref:cache[redis]")

Implementing Configurable Classes

build_object calls from_config(config) on the target class when it has one and falls back to cls(**config) otherwise. It dispatches on the method, not on a base class, so any class providing from_config is buildable from configuration.

StructuredConfigConsumer provides one, driven by a typed config dataclass:

from dataclasses import dataclass
from typing import ClassVar, Literal

from dataknobs_common.structured_config import (
    StructuredConfig,
    StructuredConfigConsumer,
)


@dataclass(frozen=True)
class MyDatabaseConfig(StructuredConfig):
    host: str = "localhost"
    port: int = 5432

    _UNKNOWN_KEYS: ClassVar[Literal["ignore", "raise"]] = "raise"


class MyDatabase(StructuredConfigConsumer[MyDatabaseConfig]):
    CONFIG_CLS: ClassVar[type[MyDatabaseConfig]] = MyDatabaseConfig

    def _setup(self) -> None:
        self.dsn = f"postgresql://{self.config.host}:{self.config.port}"

The dataclass is the schema, so self.config is typed and a misspelling is caught rather than defaulted:

MyDatabase.from_config({"hosst": "db"})
# ValueError: MyDatabaseConfig does not accept 'hosst' (did you mean 'host'?).
#             Accepted keys: host, port.

The class is also constructible directly from a config object (MyDatabase(MyDatabaseConfig(host="db"))), which the config-driven path above does not have to know about. See Structured Configuration for the full API.

dataknobs_config.ConfigurableBase is the deprecated predecessor of this pattern: it splats the mapping into the constructor with no schema to check it against. Existing users keep working and no runtime warning is raised, but new code should use the typed base above. See ConfigurableBase (deprecated) for what changes and why.

Implementing Factories

from dataknobs_config import FactoryBase

class DatabaseFactory(FactoryBase):
    def create(self, **config):
        db_type = config.pop("type", "postgresql")
        if db_type == "postgresql":
            return PostgreSQL(**config)
        elif db_type == "mysql":
            return MySQL(**config)

Lazy Factory Access

# Configuration with factory
config = Config({
    "database": [{
        "name": "primary",
        "factory": "myapp.db.DatabaseFactory",
        "type": "postgresql",
        "host": "localhost"
    }]
})

# Get the factory instance (cached)
factory = config.get_factory("database", "primary")
db1 = factory.create(database="app1")
db2 = factory.create(database="app2")

# Or get an instance directly
db = config.get_instance("database", "primary", database="myapp")

API Reference

Config Class

class Config:
    def __init__(self, *sources, use_env=True)
    def from_file(cls, path) -> Config
    def from_dict(cls, data) -> Config
    
    # Access
    def get_types() -> List[str]
    def get_count(type_name: str) -> int
    def get_names(type_name: str) -> List[str]
    def get(type_name: str, name_or_index: Union[str, int] = 0) -> dict
    def set(type_name: str, name_or_index: Union[str, int], config: dict)
    
    # References
    def resolve_reference(ref: str) -> dict
    def build_reference(type_name: str, name_or_index: Union[str, int]) -> str
    
    # Merging
    def merge(other: Config, precedence: str = "first")
    
    # Export
    def to_dict() -> dict
    def to_file(path: Path, format: str = None)
    
    # Object Construction
    def build_object(ref: str, cache: bool = True, **kwargs) -> Any
    def clear_object_cache(ref: str = None)
    
    # Lazy Factory Access
    def get_factory(type_name: str, name_or_index: Union[str, int] = 0) -> Any
    def get_instance(type_name: str, name_or_index: Union[str, int] = 0, **kwargs) -> Any

Examples

Multi-Environment Configuration

# base.yaml
database:
  - name: primary
    host: localhost
    port: 5432

# production.yaml  
database:
  - name: primary
    host: prod.db.example.com
    pool_size: 50

# Load with overrides
config = Config("base.yaml", "production.yaml")

Service Discovery Integration

config = Config({
    "services": [
        {"name": "auth", "url": "http://auth:8000"},
        {"name": "api", "url": "http://api:8080"}
    ],
    "app": [{
        "name": "main",
        "auth_service": "xref:services[auth]",
        "api_service": "xref:services[api]"
    }]
})

app = config.resolve_reference("xref:app[main]")
# app["auth_service"]["url"] = "http://auth:8000"

Dynamic Configuration with Environment

# Development: export DATAKNOBS_DATABASE__PRIMARY__HOST=localhost
# Production:  export DATAKNOBS_DATABASE__PRIMARY__HOST=prod.db.aws.com

config = Config.from_file("config.yaml")
db = config.get("database", "primary")
# Automatically uses environment-appropriate host

Configuration Inheritance

For simple YAML/JSON configuration files with inheritance support, use InheritableConfigLoader:

from dataknobs_config import InheritableConfigLoader, load_config_with_inheritance

# Create a loader
loader = InheritableConfigLoader("./configs")

# Load configuration with inheritance
config = loader.load("my-domain")

Base Configuration

# configs/base.yaml
llm:
  provider: openai
  model: gpt-4
  temperature: 0.7

knowledge_base:
  chunk_size: 500
  overlap: 50

Child Configuration

# configs/domain.yaml
extends: base

llm:
  model: gpt-4-turbo  # Override just this field

domain_specific:
  feature_enabled: true

Environment Variable Substitution

# configs/production.yaml
extends: base

llm:
  api_key: ${OPENAI_API_KEY}
  model: ${LLM_MODEL:gpt-4}  # With default value

paths:
  data_dir: ${DATA_DIR:~/data}  # Supports ~ expansion

InheritableConfigLoader API

class InheritableConfigLoader:
    def __init__(
        self,
        config_dir: str | Path | None = None,
        *,
        resolver: ResourceResolver[str, str] | None = None,
    )

    # Map a config name to a location under config_dir. Applied to the
    # requested config AND to every `extends:` target; identity by default.
    # Not applied under load_from_file. A name -- requested, resolved, or
    # read from an `extends:` -- may address a subdirectory of config_dir
    # but may not land outside it: `..` or an absolute path that leaves
    # config_dir raises InheritanceError (containment is judged on where
    # the name lands, not on how it is spelled -- an absolute name
    # pointing back inside config_dir is fine).
    # load_from_file is unaffected; the path is yours.
    # A layout that genuinely spans sibling trees lifts the bound with
    # InheritableConfigLoader(config_dir, allow_outside=True).
    def resolve_name(self, name: str) -> str

    # The names load() accepts. Defaults to the stems directly under
    # config_dir, which is the loadable set only while resolve_name is
    # identity -- the mapping is one-way, so override this alongside it.
    def available_names(self) -> list[str]

    # The default's body, taking a directory: an override is this pointed
    # somewhere else. Globs the extensions load() probes, from the one
    # shared list, so enumeration cannot fall behind loading.
    @staticmethod
    def stems_in(directory: Path) -> list[str]

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

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

    # List available configurations (delegates to available_names)
    def list_available(self) -> list[str]

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

    # Clear cache. Pass the name you passed load() -- this resolves it
    # the same way, so an already-resolved name is mapped a second time.
    # The debug log reports how many entries that removed; zero is the
    # sign it missed.
    def clear_cache(self, name: str | None = None) -> None

Name Resolution

resolve_name governs how a config name maps to a location, including for extends: targets — so a tree whose children name their parents bare still loads. Two modes, which are alternatives, not layers: an override replaces the default, so a loader given both ignores the injected resolver (or applies both mappings, if the override calls super()). Constructing that combination warns, since the first outcome is otherwise silent.

from dataknobs_common import CallableResolver, MappingResolver

# Inject a shipped resolver -- no consumer class needed
loader = InheritableConfigLoader(
    "./configs", resolver=CallableResolver(lambda n: f"domains/{n}")
)
loader = InheritableConfigLoader(
    "./configs", resolver=MappingResolver({"tutor": "domains/bio-tutor"})
)

# Or override the method, when the mapping needs loader state
class DomainAwareLoader(InheritableConfigLoader):
    def resolve_name(self, name: str) -> str:
        return f"{self.domain_root}/{name}"

The resolved name is what keys the cache, the cycle-detection set, the extends: invalidation edges, and clear_cache, so two spellings of one config are one entry. load_from_file suppresses resolution for the file and its whole extends: subtree, since it rebinds config_dir.

The mapping is one-way — nothing runs a resolver backwards — so a deployment that governs it also has to say which names exist. Override available_names alongside resolve_name; leaving it alone under a resolver does not raise, it reports the wrong thing quietly ([], for a layout one directory down). Build the override out of stems_in, which globs the extensions load probes — hand-rolling it against *.yaml alone silently omits every .json config while leaving them perfectly loadable.

class DomainLoader(InheritableConfigLoader):
    def resolve_name(self, name: str) -> str:
        return f"domains/{name}"

    def available_names(self) -> list[str]:
        return self.stems_in(self.config_dir / "domains")

Convenience Function

from dataknobs_config import load_config_with_inheritance

# Quick one-liner for loading a config file
config = load_config_with_inheritance("configs/my-domain.yaml")

Utility Functions

from dataknobs_config import deep_merge, substitute_env_vars

# Deep merge two dictionaries
merged = deep_merge(base_dict, override_dict)

# Substitute environment variables in any data structure
result = substitute_env_vars({"key": "${MY_VAR:default}"})

Best Practices

1. Use Type Organization: Group related configurations by type
2. Leverage Defaults: Define common values in settings to avoid repetition
3. Environment Overrides: Use for deployment-specific values (hosts, ports, credentials)
4. File References: Split large configurations into manageable files
5. Path Resolution: Use relative paths in configs for portability
6. Object Caching: Enable caching for expensive object construction
7. Use Inheritance: Create base configs and extend them for specific environments/domains

Testing

Run tests with pytest:

pytest tests/

License

MIT License - see LICENSE file for details.