loom.core.config¶

Configuration loading utilities for Loom applications.

Provides an omegaConf-backed loader that supports multiple YAML files and typed section extraction. The framework imposes no shape on the config object — the user owns the structure.

Example:

from loom.core.config import load_config, section

cfg = load_config("config/base.yaml", "config/local.yaml")
db = section(cfg, "database", DatabaseConfig)

class loom.core.config.ConfigBinding(target, config_path='', overrides=<factory>)[source]¶

Bases: LoomFrozenStruct

Deferred binding between a class and a config section.

The binding is a declaration only. It does not read files, access global config state, or instantiate target. Compiler/bootstrap code resolves it later using an explicit runtime config object.

Parameters:

target (type[object]) – Runtime behaviour class to instantiate later.
config_path (str) – Dot-separated config path. Empty means no YAML section is required and only overrides are applied.
overrides (dict[str, object]) – Explicit keyword overrides. These win over YAML values when the binding is resolved.

class loom.core.config.ConfigContext(config, *, binder=None)[source]¶

Bases: object

Runtime config accessor backed by a resolved DictConfig.

Combines typed section extraction with constructor injection so that runner and bootstrap code has a single, consistent entry-point for reading config — no scattered section() calls or bespoke binding loops across layers.

Prefer the factory methods for construction:

from_yaml() — load from one or more YAML files.
from_dict() — build from a plain Python mapping (tests, code-defined config).
Direct ConfigContext(DictConfig) — when a DictConfig is already available.

Parameters:

config (DictConfig) – Resolved OmegaConf DictConfig.
binder (StructBinder | None) – Optional StructBinder. Defaults to a non-strict binder. Pass StructBinder(strict=True) to disallow implicit coercion.

Example:

ctx = ConfigContext.from_yaml("config.yaml")
db   = ctx.section("database", DatabaseConfig)
step = ctx.bind(ReadOrdersStep, path="pipeline.steps.orders")
dep  = ctx.resolve(MyDep.from_config("services.dep", label="prod"))

classmethod from_yaml(*paths, resolvers=(), binder=None)[source]¶

Build a ConfigContext from one or more YAML files.

Files are merged left-to-right; later files override earlier ones. Accepts local paths and cloud URIs (s3://, gs://, …).

Parameters:

*paths (str) – One or more local paths or cloud URIs.
resolvers (Sequence[ConfigResolver]) – Optional custom resolvers for ${name:key} placeholders.
binder (StructBinder | None) – Optional StructBinder override.

Returns:

A ready-to-use ConfigContext.

Raises:

ConfigError – If any file cannot be read or parsed.

Return type:

ConfigContext

classmethod from_dict(raw, *, binder=None)[source]¶

Build a ConfigContext from a plain Python mapping.

Primarily used in tests and inline config construction. The mapping must resolve to a top-level dictionary (not a list or scalar).

Parameters:

raw (Mapping[str, Any]) – Plain Python mapping of config keys to values.
binder (StructBinder | None) – Optional StructBinder override.

Returns:

A ready-to-use ConfigContext.

Raises:

TypeError – If raw does not produce a DictConfig.

Return type:

ConfigContext

has(key)[source]¶

Return True if the dot-separated key exists in the config.

Parameters:: key (str | ConfigKey) – Dot-separated path (e.g. "streaming.runtime").
Returns:: True when the key resolves to a non-null value.
Return type:: bool

section_optional(key, target)[source]¶

Extract key as target when present, otherwise return None.

Parameters:

key (str | ConfigKey) – Dot-separated path to the section.
target (type[T]) – Type to convert the section into.

Returns:

Validated instance of target, or None if the key is absent.

Return type:

T | None

section_or_default(key, target, default)[source]¶

Extract key as target when present, otherwise return default.

Parameters:

key (str | ConfigKey) – Dot-separated path to the section.
target (type[T]) – Type to convert the section into.
default (T) – Fallback value when the section is absent.

Returns:

Validated instance of target, or default when absent.

Return type:

section(key, target)[source]¶

Extract and validate a typed config section by dot-path.

Parameters:

key (str | ConfigKey) – Dot-separated path (e.g. "database" or "services.cache").
target (type[T]) – Type to convert the section into.

Returns:

Validated instance of target.

Raises:

ConfigError – If the key is absent or the section fails validation.

Return type:

bind(target, *, path='', **overrides)[source]¶

Instantiate target from a config section merged with overrides.

overrides take precedence over YAML values. Omit path to rely entirely on overrides.

Parameters:

target (type[T]) – Class to instantiate.
path (str | ConfigKey) – Dot-separated config path. Omit for override-only bindings.
**overrides (object) – Explicit keyword values applied after YAML values.

Returns:

Fully constructed instance of target.

Raises:

ConfigError – If a required field is missing or type conversion fails.

Return type:

resolve(binding)[source]¶

Materialize a ConfigBinding into a live object.

Delegates to bind() using the binding’s declared path and overrides. Compilers and bootstrap code that operate on pre-declared bindings use this method rather than calling bind() with unpacked fields.

Parameters:: binding (ConfigBinding) – Deferred binding declaration (from from_config() or configure()).
Returns:: Instantiated binding.target.
Raises:: ConfigError – If resolution or type conversion fails.
Return type:: object

exception loom.core.config.ConfigError[source]¶

Bases: Exception

Raised when configuration loading or validation fails.

Parameters:: message – Human-readable description of the failure.

Example:

raise ConfigError("Missing required field: db_url")

class loom.core.config.ConfigKey(value)[source]¶

Bases: StrEnum

Top-level YAML section keys consumed by section().

Using StrEnum keeps references typo-proof while remaining compatible with any API that expects plain str section identifiers.

Example:

cfg = section(raw, ConfigKey.CELERY, CeleryConfig)

class loom.core.config.ConfigResolver(*args, **kwargs)[source]¶

Bases: Protocol

Protocol for pluggable config value resolvers.

Implementors provide a name (used as the OmegaConf placeholder prefix) and a resolve callable that fetches the actual value at parse time.

See SsmResolver for the bundled AWS SSM implementation. Custom resolvers only need to satisfy this two-member protocol:

class VaultResolver:
    @property
    def name(self) -> str:
        return "vault"

    def resolve(self, key: str) -> object:
        return vault_client.read_secret(key)

cfg = load_config("config/prod.yaml", resolvers=[VaultResolver()])

property name: str¶

OmegaConf resolver prefix.

Used as the placeholder prefix in YAML: ${<name>:key}. Must be unique across all registered resolvers.

Returns:: Resolver name string (e.g. "ssm", "keyvault").

resolve(key)[source]¶

Resolve key to its string value.

Called by OmegaConf when materialising ${<name>:key} placeholders. Runs at config parse time (job startup), so the returned value reflects the current state of the backing store.

Parameters:: key (str) – Key portion of the placeholder after the prefix separator (e.g. "/prod/token" for ${ssm:/prod/token}).
Returns:: Resolved value. Typically a string, but may be a structured type when the resolver supports JSON navigation.
Return type:: object

class loom.core.config.Configurable[source]¶

Bases: object

Mixin for classes that can be declared from config paths.

The mixin returns ConfigBinding declarations. It does not impose a constructor shape and does not resolve configuration itself.

classmethod from_config(config_path, **overrides)[source]¶

Declare this class as configured from a YAML section.

Parameters:

config_path (str) – Dot-separated config path resolved later by compiler or bootstrap code.
**overrides (object) – Explicit keyword overrides applied after YAML values.

Returns:

Deferred config binding for this class.

Raises:

ValueError – If config_path is empty or blank.

Return type:

ConfigBinding

classmethod configure(**overrides)[source]¶

Declare this class with explicit keyword overrides only.

Parameters:: **overrides (object) – Explicit keyword values used during resolution.
Returns:: Deferred config binding for this class.
Return type:: ConfigBinding

class loom.core.config.OtelConfig(service_name='loom', tracer_name='loom', tracer_version='', protocol='http/protobuf', endpoint='', insecure=False, headers=<factory>, resource_attributes=<factory>, span_attributes=<factory>, exporter_kwargs=<factory>, span_processor_kwargs=<factory>)[source]¶

Bases: LoomFrozenStruct

OpenTelemetry SDK/exporter configuration.

Parameters:

service_name (str) – Resource attribute service.name.
tracer_name (str) – Tracer instrumentation name.
tracer_version (str) – Optional tracer instrumentation version.
protocol (str) – OTLP protocol (http/protobuf or grpc).
endpoint (str) – OTLP endpoint URI. When empty, uses global OTel runtime defaults.
insecure (bool) – Exporter transport mode when supported by protocol/exporter.
headers (dict[str, str]) – Exporter request headers (vendor auth/tags).
resource_attributes (dict[str, str]) – Additional OTel resource attributes.
span_attributes (dict[str, str]) – Static span attributes added to all spans emitted by this observer.
exporter_kwargs (dict[str, Any]) – Extra keyword args passed through to OTLP exporter.
span_processor_kwargs (dict[str, Any]) – Extra keyword args passed through to BatchSpanProcessor.

validate()[source]¶

Validate the configured transport protocol.

Raises:: ValueError – If protocol is unsupported.
Return type:: None

class loom.core.config.SecretsManagerResolver(region=None)[source]¶

Bases: object

Resolves AWS Secrets Manager paths for use with load_config().

Fetches secret values from AWS Secrets Manager. The boto3 client is created lazily on first use and reused across calls.

Env-var tokens in the form %VAR_NAME% (uppercase letters, digits, and underscores only) are expanded from os.environ before the request is made.

Parameters:: region (str | None) – AWS region name. Passed directly to boto3.client. Defaults to None, which lets boto3 use its own resolution chain (env vars, instance metadata, etc.).

Example:

resolver = SecretsManagerResolver("eu-west-1")
value = resolver.resolve("/myapp/%ENV%/db_password")

property name: str¶

OmegaConf resolver prefix.

Returns:: The string "secrets".

resolve(key)[source]¶

Resolve an AWS Secrets Manager path to its stored value.

Expands %VAR_NAME% tokens in key from the environment, then fetches the secret from AWS Secrets Manager.

Parameters:: key (str) – Secret path, optionally containing %VAR_NAME% placeholders that are replaced with environment variable values. Supports dot-notation for JSON key navigation: /path/secret.key fetches /path/secret and returns secret["key"].
Returns:: Resolved value. A plain string for secrets without dot-notation; a structured value (string, int, dict, etc.) when dot-notation navigates into a JSON secret.
Raises:: ConfigError – When key is empty, an env-var placeholder is missing, boto3 is not installed, the secret is binary, or the API call fails.
Return type:: object

class loom.core.config.SsmResolver(region=None, *, with_decryption=True)[source]¶

Bases: object

Resolves SSM Parameter Store paths for use with load_config().

Fetches parameter values from AWS Systems Manager Parameter Store. The boto3 client is created lazily on first use and reused across calls.

Env-var tokens in the form %VAR_NAME% (uppercase letters, digits, and underscores only) are expanded from os.environ before the SSM request is made.

Parameters:

region (str | None) – AWS region name. Passed directly to boto3.client. Defaults to None, which lets boto3 use its own resolution chain (env vars, instance metadata, etc.).
with_decryption (bool) – Whether to decrypt SecureString parameters. Defaults to True.

Example:

resolver = SsmResolver("eu-west-1")
value = resolver.resolve("/myapp/%ENV%/db_password")

property name: str¶

Resolver name used as the OmegaConf placeholder prefix.

Returns:: The string "ssm".

resolve(key)[source]¶

Resolve an SSM parameter path to its stored value.

Expands %VAR_NAME% tokens in key from the environment, then fetches the parameter from AWS SSM Parameter Store.

Parameters:: key (str) – SSM parameter path, optionally containing %VAR_NAME% placeholders that are replaced with environment variable values. Supports dot-notation for JSON key navigation: /path/param.key fetches /path/param and returns param["key"].
Returns:: Resolved value. A plain string for parameters without dot-notation; a structured value (string, int, dict, etc.) when dot-notation navigates into a JSON parameter.
Raises:: ConfigError – When key is empty, an env-var placeholder is missing, boto3 is not installed, or the SSM API call fails.
Return type:: object

class loom.core.config.StructBinder(strict=False)[source]¶

Bases: object

Strategy: constructor injection from a resolved config mapping.

Converts every annotated __init__ parameter present in raw via msgspec.convert. Supports primitives (int, str, bool), Literal constraints, and LoomFrozenStruct subclasses uniformly. All reflection runs once per bind call at compile time — never on the message-processing hot path.

Parameters:: strict (bool) – When True, msgspec.convert uses strict mode (no implicit coercion from string to int, etc.).

Example:

binder = StructBinder()
step = binder.bind(
    ReadOrdersStep,
    {"db": {"host": "localhost", "port": 5432}, "mode": "batch"},
)

bind(target, raw)[source]¶

Instantiate target injecting and converting values from raw.

For each annotated __init__ parameter that appears in raw, the value is converted to the declared type via msgspec.convert. Parameters absent from raw use their default; required parameters absent from raw raise ConfigError.

Parameters:

target (type[T]) – Class to instantiate.
raw (Mapping[str, object]) – Flat mapping of parameter names to raw values.

Returns:

A fully constructed instance of target.

Raises:

ConfigError – If a required parameter is absent from raw or a value fails type conversion.

Return type:

loom.core.config.load_config(*config_files, resolvers=())[source]¶

Load and merge one or more YAML config files into a DictConfig.

Accepts local filesystem paths and cloud storage URIs (s3://, gs://, abfss://, r2:// …). Cloud files are fetched via fsspec at call time.

Files are merged left-to-right: values in later files override those in earlier ones. ${oc.env:VAR} interpolations are resolved by OmegaConf. Custom resolvers are registered before parsing so their ${name:key} placeholders resolve during the same pass.

Local files may declare a top-level includes list to pull in additional files before their own values. Included paths are relative to the declaring file. Circular includes raise ConfigError. The includes directive is not supported for cloud URIs.

The framework does not impose any shape on the resulting config — the user owns the structure entirely. Use section() to extract typed sub-objects where desired.

Parameters:

*config_files (str) – One or more local paths or cloud URIs.
resolvers (Sequence[ConfigResolver]) – Optional sequence of ConfigResolver instances. Each resolver registers a ${name:key} placeholder resolved at parse time (e.g. from AWS SSM or Azure Key Vault).

Returns:

Merged omegaconf.DictConfig with interpolation support.

Raises:

ConfigError – If no files are provided, a file is not found, parsing fails, a circular include is detected, omegaconf is not installed, or a cloud URI fetch fails.

Return type:

DictConfig

Example — single local file with inline includes:

cfg = load_config("config.yaml")

Example — explicit multi-file composition:

cfg = load_config("config/base.yaml", "config/production.yaml")
db_url = cfg.database.url

Example — cloud URI:

cfg = load_config("s3://my-bucket/config/prod.yaml")

Example — with custom resolver:

cfg = load_config("config/prod.yaml", resolvers=[SsmResolver("eu-west-1")])

loom.core.config.section(cfg, key, target_type)[source]¶

Extract and validate a config section as a typed object.

Navigates cfg by key (dot-notation supported, e.g. "database.primary"), resolves omegaConf interpolations, and converts the result to target_type via msgspec.convert.

Works with any type supported by msgspec.convert: msgspec.Struct subclasses, dataclasses, TypedDict, plain dicts, etc.

Parameters:

cfg (DictConfig) – Root omegaconf.DictConfig returned by load_config().
key (str) – Dot-separated path to the desired section (e.g. "database" or "services.cache").
target_type (type[T]) – Type to convert the section into.

Returns:

Validated instance of target_type.

Raises:

ConfigError – If the key is absent, the section cannot be resolved, or msgspec validation fails.

Return type:

Example:

class DatabaseConfig(msgspec.Struct, kw_only=True):
    url: str
    pool_size: int = 5

db = section(cfg, "database", DatabaseConfig)