Config API¶
The weevr.config module handles YAML parsing, schema validation, configuration
inheritance, and parameter resolution across the loom/weave/thread hierarchy.
weevr.config
¶
Configuration loading and validation.
ConfigError
¶
Bases: WeevError
Base exception for configuration-related errors.
__init__(message, cause=None, file_path=None, config_key=None)
¶
Initialize ConfigError.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
Human-readable error message |
required |
cause
|
Exception | None
|
Optional underlying exception |
None
|
file_path
|
str | None
|
Path to the config file where the error occurred |
None
|
config_key
|
str | None
|
Specific config key that caused the error |
None
|
__str__()
¶
Return string representation with context.
ModelValidationError
¶
Bases: ConfigError
Raised when a fully resolved config fails to hydrate into a typed model.
This occurs after variable resolution and inheritance, when the concrete values are validated through the Pydantic domain model (Thread, Weave, or Loom). Semantic constraints that span multiple fields are checked here.
Loom
¶
Bases: FrozenBase
A deployment unit containing weave references with optional shared defaults.
Thread
¶
Bases: FrozenBase
Complete domain model for a thread configuration.
A thread is the smallest unit of work: one or more sources, a sequence of transformation steps, and a single target.
Weave
¶
Bases: FrozenBase
A collection of thread references with optional shared defaults.
apply_inheritance(loom_defaults, weave_defaults, thread_config)
¶
Apply multi-level inheritance cascade.
Cascade order (lowest to highest priority): 1. loom_defaults (lowest) 2. weave_defaults 3. thread_config (highest)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
loom_defaults
|
dict[str, Any] | None
|
Defaults from loom level |
required |
weave_defaults
|
dict[str, Any] | None
|
Defaults from weave level |
required |
thread_config
|
dict[str, Any]
|
Thread-specific config |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Fully merged config with thread values taking precedence |
expand_foreach(steps)
¶
Expand foreach macro blocks into repeated step sequences.
Each foreach block in the steps list is replaced by its template steps
repeated once per value, with {var} placeholders substituted.
Non-foreach entries pass through unchanged.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
steps
|
list[dict[str, Any]]
|
Raw step list (dicts), possibly containing foreach blocks. |
required |
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
Expanded step list with all foreach blocks replaced. |
Raises:
| Type | Description |
|---|---|
ConfigError
|
If a foreach block is missing required fields. |
detect_config_type(raw)
¶
Detect the type of config from its structure.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
raw
|
dict[str, Any]
|
Parsed config dictionary |
required |
Returns:
| Type | Description |
|---|---|
str
|
Config type: 'thread', 'weave', 'loom', or 'params' |
Raises:
| Type | Description |
|---|---|
ConfigParseError
|
If config type cannot be determined |
detect_config_type_from_extension(path)
¶
Detect config type from file extension.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | Path
|
Path to the config file. |
required |
Returns:
| Type | Description |
|---|---|
str | None
|
Config type string if the extension is a typed extension |
str | None
|
( |
str | None
|
|
Raises:
| Type | Description |
|---|---|
ConfigError
|
If the extension is |
extract_config_version(raw)
¶
Extract and parse config_version from a config dict.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
raw
|
dict[str, Any]
|
Parsed config dictionary |
required |
Returns:
| Type | Description |
|---|---|
tuple[int, int]
|
Tuple of (major, minor) version numbers |
Raises:
| Type | Description |
|---|---|
ConfigParseError
|
If config_version is missing or invalid format |
parse_yaml(path)
¶
Parse a YAML file and return its contents.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | Path
|
Path to the YAML file |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Parsed YAML content as a dictionary |
Raises:
| Type | Description |
|---|---|
ConfigParseError
|
If file not found, unreadable, or invalid YAML syntax |
validate_config_version(version, config_type)
¶
Validate that the config version is supported.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
version
|
tuple[int, int]
|
Tuple of (major, minor) version |
required |
config_type
|
str
|
Type of config (thread, weave, loom, params) |
required |
Raises:
| Type | Description |
|---|---|
ConfigVersionError
|
If the major version doesn't match supported version |
build_param_context(runtime_params=None, config_defaults=None)
¶
Build parameter context with proper priority layering.
Priority order (highest to lowest): 1. runtime_params 2. config_defaults
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
runtime_params
|
dict[str, Any] | None
|
Runtime parameter overrides. |
None
|
config_defaults
|
dict[str, Any] | None
|
Default parameters from config. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Merged parameter context dictionary with dotted key access support. |
resolve_references(config, config_type, project_root, runtime_params=None, visited=None)
¶
Resolve references to other config files.
Handles both external references (ref key) and inline definitions
(name + body keys). Recursively loads referenced configs with
circular reference detection.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any]
|
Config dict to resolve references in. |
required |
config_type
|
str
|
Type of this config ( |
required |
project_root
|
Path
|
Absolute path to the |
required |
runtime_params
|
dict[str, Any] | None
|
Runtime parameters to pass to child configs. |
None
|
visited
|
set[str] | None
|
Set of already-visited ref strings (for cycle detection). |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Config dict with resolved child configs attached under |
dict[str, Any]
|
|
Raises:
| Type | Description |
|---|---|
ReferenceResolutionError
|
If referenced file not found or circular reference detected. |
ConfigError
|
If an inline definition is missing a |
resolve_variables(config, context)
¶
Recursively resolve variable references in config.
Supports: - ${var} - simple variable reference (error if not found) - ${var:-default} - variable with fallback default
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
dict[str, Any] | list[Any] | str | Any
|
Config structure to resolve (dict, list, str, or primitive) |
required |
context
|
dict[str, Any]
|
Parameter context for variable lookup |
required |
Returns:
| Type | Description |
|---|---|
Any
|
Config with all variables resolved |
Raises:
| Type | Description |
|---|---|
VariableResolutionError
|
If variable not found and no default provided |
validate_schema(raw, config_type)
¶
Validate a raw config dict against the appropriate pre-resolution schema.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
raw
|
dict[str, Any]
|
Raw config dictionary (variables not yet resolved) |
required |
config_type
|
str
|
Type of config (thread, weave, loom, params) |
required |
Returns:
| Type | Description |
|---|---|
BaseModel
|
Validated Pydantic model instance |
Raises:
| Type | Description |
|---|---|
ConfigSchemaError
|
If validation fails |
_derive_config_name(path)
¶
Derive the component name from a file path.
Returns the filename stem — the filename without the typed extension.
For example, dim_customer.thread returns 'dim_customer'.
load_config(path, runtime_params=None, project_root=None)
¶
Load and validate a weevr configuration file.
This function orchestrates the full config loading pipeline: 1. Parse YAML file 2. Extract and validate config_version 3. Detect config type (extension-based for components, content-based for params) 4. Validate schema with Pydantic 5. Build parameter context (runtime > defaults) 6. Resolve variable references (${var} and ${var:-default}) 7. Resolve references to child configs (threads, weaves) 8. Apply inheritance cascade (loom -> weave -> thread) 9. Validate name against filename stem 10. Hydrate into typed domain model (thread, weave, loom only)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | Path
|
Path to the config file (thread, weave, or loom). |
required |
runtime_params
|
dict[str, Any] | None
|
Optional runtime parameter overrides. |
None
|
project_root
|
Path | None
|
Path to the |
None
|
Returns:
| Type | Description |
|---|---|
Thread | Weave | Loom | dict[str, Any]
|
A frozen, typed domain model instance (Thread, Weave, or Loom) for |
Thread | Weave | Loom | dict[str, Any]
|
thread/weave/loom config types. Returns a plain dict for params configs. |
Raises:
| Type | Description |
|---|---|
ConfigParseError
|
YAML syntax errors, file not found |
ConfigVersionError
|
Unsupported config version |
ConfigSchemaError
|
Schema validation failures |
ConfigError
|
Extension or name validation failures |
VariableResolutionError
|
Unresolved variables without defaults |
ReferenceResolutionError
|
Missing referenced files, circular dependencies |
ModelValidationError
|
Semantic validation failures during model hydration |