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=(), keyed=(), 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.
keyed (Sequence[str]) – Dotted paths of the collections merged by key within an includes composition.
binder (StructBinder | None) – Optional StructBinder override.

Returns:

A ready-to-use ConfigContext.

Raises:

ConfigError – If any file cannot be read or parsed, or a keyed collection holds a duplicate key or mixes list and mapping forms.

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>, adopt_host_id_generator=False, sampler='parentbased_always_on', sampler_ratio=1.0, max_span_links=128)[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.
adopt_host_id_generator (bool) – Installs Loom’s trace-id generator on the already-installed TracerProvider so streaming messages get end-to-end traces. Host spans are unaffected — the generator delegates to the default random generator whenever no Loom trace id is active. Only meaningful when endpoint is empty; Loom’s own provider always gets the generator.
sampler (str) – Sampler name, one of SAMPLERS. Applied to Loom’s own provider only: with an empty endpoint the host sampler decides.
sampler_ratio (float) – Sampled fraction for the ratio-based samplers. Because the OTEL trace id is the message trace id, the decision is stable from ingestion to death and across services, so a sampled message yields a complete trace and an unsampled one yields none.
max_span_links (int) – Upper bound on the links a batch span carries. When the bound is hit the span is marked with loom.links_truncated.

validate()[source]¶

Validate the transport protocol, the sampler, and the link bound.

Raises:

ValueError – If protocol is unsupported.
ValueError – If the sampler name is unknown, the sampled ratio is outside [0.0, 1.0], or the link bound is not positive.

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.

Keys are trusted deployment configuration: they come from the deployment’s own config files, never from request input. The resolver logs and reports the key as written (with its %VAR% tokens), never the expanded path. Two channels remain for a caller that lets untrusted input reach a resolver key: a missing variable raises ConfigError("Env var 'X' not found ..."), a deterministic existence oracle for any attacker-chosen variable name, and the chained cause (__cause__) of a fetch error may carry the expanded path echoed by the AWS client. The client is also called with the expanded path, so botocore’s DEBUG request logging reproduces it — keep botocore at INFO or above in production.

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.

Splits the dot-notation tail off key as written, expands %VAR_NAME% tokens in the remaining base path 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"]. Dots are read from the key as written, so an expanded value containing a dot is part of the path and never a navigation separator.
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.

Splits the dot-notation tail off key as written, expands %VAR_NAME% tokens in the remaining base path 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"]. Dots are read from the key as written, so an expanded value containing a dot is part of the path and never a navigation separator.
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.default_resolvers()[source]¶

Return loom’s built-in resolvers: secrets and ssm.

Both use the AWS SDK’s default region and credential chain and create their client lazily on the first resolution.

Returns:: A SecretsManagerResolver followed by a SsmResolver, both freshly built.
Return type:: tuple[ConfigResolver, …]

loom.core.config.expand_config_glob(pattern)[source]¶

Expand a configuration file pattern on the local or a cloud filesystem.

Parameters:: pattern (str) – Resolved local path or cloud URI, optionally containing glob characters (*, ?, [).
Returns:: [pattern] for a plain entry; for a glob, the matching regular .yaml/.yml files sorted lexicographically by resolved path.
Raises:: ConfigError – When a glob matches no configuration file or the cloud backend fails to list it.
Return type:: list[str]

loom.core.config.is_cloud_uri(uri)[source]¶

Return True when uri uses a cloud storage scheme.

Parameters:: uri (str) – Local path or URI.
Returns:: Whether the scheme belongs to a cloud storage backend.
Return type:: bool

loom.core.config.load_config(*config_files, resolvers=(), keyed=())[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.

Any file may declare a top-level includes list to pull in additional files before its own values. Entries are relative to the declaring file, may live on any scheme, and may be globs. Circular includes and entries matching no file raise ConfigError.

keyed names collections whose mapping form merges by key within an includes composition: a key declared by two files of the same composition raises ConfigError, as does mixing a list form with a mapping form. Explicit config_files layers override keys left-to-right; only the form mismatch is reported across them.

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).
keyed (Sequence[str]) – Dotted paths (e.g. "storage.tables") of the collections merged by key.

Returns:

Merged omegaconf.DictConfig with interpolation support.

Raises:

ConfigError – If no files are provided, a file is not found, parsing fails, an include matches no file, a circular include is detected, a keyed collection holds a duplicate key or mixes list and mapping forms, 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")])

Example — keyed collections:

cfg = load_config("config/app.yaml", keyed=("storage.tables", "storage.files"))

loom.core.config.merge_resolvers(explicit, defaults)[source]¶

Return explicit followed by the defaults whose names are still free.

A default is dropped when an explicit resolver takes its name or when a resolver with that name is already registered in OmegaConf; the latter is logged at DEBUG level.

Parameters:

explicit (Sequence[ConfigResolver]) – Resolvers that keep their position and always win their name.
defaults (Sequence[ConfigResolver]) – Candidate resolvers appended after explicit when their name is not taken.

Returns:

explicit in order, followed by the kept defaults in order.

Return type:

tuple[ConfigResolver, …]

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)

loom.core.config.with_default_resolvers(explicit=())[source]¶

Return explicit followed by loom’s built-in resolvers whose names are free.

Equivalent to merge_resolvers(explicit, default_resolvers()); the factories use it to register secrets and ssm behind user resolvers.

Parameters:: explicit (Sequence[ConfigResolver]) – User resolvers that keep their position and always win their name.
Returns:: explicit in order, followed by the built-in resolvers whose name is neither taken by explicit nor already registered.
Return type:: tuple[ConfigResolver, …]