medallantic API¶
Status: Available in ETLantic 0.41.0. Medallion facade + SparkForge migrate. Install narrative: package README. Hub: Optional packages API.
Setup¶
Failure modes¶
| Topic | Behavior |
|---|---|
| IR convertibility | MDL210 manual; MDL220 unsupported |
Public API¶
Medallantic — engine-agnostic medallion pipelines built on ETLantic.
dataclass
¶
AdaptationResult(pipeline_cls: type[Pipeline], profile: Profile, validation_policy: ValidationPolicy, write_intents: tuple[WriteIntent, ...] = (), step_map: dict[str, str] = dict(), layer_by_node: dict[str, str] = dict(), diagnostics: tuple[Diagnostic, ...] = (), metadata: dict[str, Any] = dict(), required_delta_operations: tuple[str, ...] = ())
Result of adapting a SparkForge pipeline IR.
Bases: Exception
Raised when SparkForge → ETLantic adaptation fails closed.
Bronze(*, asset: str | None = None, source: str | None = None, write_mode: str | None = None, transform_ref: str | None = None, rules: dict[str, Any] | None = None, description: str | None = None, tags: tuple[str, ...] | list[str] | None = None, kind: str | None = None, metadata: dict[str, Any] | None = None, name: str | None = None)
Class-style native medallion authoring.
Subclass and declare Bronze / Silver / Gold attributes, then call
to_definition() or lower().
classmethod
¶
Return the facade-owned medallion document for this class.
classmethod
¶
Build an anonymous MedallionPipeline subclass from a document dict.
classmethod
¶
Lower this pipeline onto ETLantic Pipeline / definition surfaces.
classmethod
¶
Return a sealed public PipelineDefinition.
Bases: StrEnum
SparkForge medallion layer (adapter-only vocabulary).
dataclass
¶
SparkForgePipelineSpec(name: str, schema: str = 'default', steps: tuple[SparkForgeStepSpec, ...] = (), min_bronze_rate: float = 90.0, min_silver_rate: float = 95.0, min_gold_rate: float = 98.0, engine: str = 'spark', legacy_engine_extensions: tuple[str, ...] = (), metadata: dict[str, Any] = dict())
Complete SparkForge pipeline description (fixture-friendly).
classmethod
¶
Parse IR and return any coercion / hygiene diagnostics.
dataclass
¶
SparkForgeStepSpec(name: str, kind: StepKind, layer: LayerKind, source: str | None = None, table_name: str | None = None, transform_ref: str | None = None, rules: dict[str, Any] = dict(), write_mode: str | None = None, metadata: dict[str, Any] = dict())
One SparkForge step in secret-free form.
Bases: StrEnum
SparkForge step kinds represented in the compatibility IR.
Bases: Exception
Raised when medallion → ETLantic lowering fails closed.
dataclass
¶
LoweringResult(pipeline_cls: type[Pipeline], profile: Profile, validation_policy: ValidationPolicy, write_intents: tuple[WriteIntent, ...] = (), step_map: dict[str, str] = dict(), layer_by_node: dict[str, str] = dict(), diagnostics: tuple[Diagnostic, ...] = (), metadata: dict[str, Any] = dict(), required_delta_operations: tuple[str, ...] = ())
Bases: Data
Generic row contract for medallion planning/parity (passthrough).
Bases: ValueError
Raised when a Medallantic rules shorthand cannot be parsed.
dataclass
¶
MedallionDocument(name: str, schema: str = 'default', steps: tuple[MedallionStep, ...] = (), min_bronze_rate: float = 90.0, min_silver_rate: float = 95.0, min_gold_rate: float = 98.0, engine: str = 'local', description: str | None = None, tags: tuple[str, ...] = (), metadata: dict[str, Any] = dict())
Declarative medallion pipeline document owned by Medallantic.
dataclass
¶
MedallionStep(name: str, layer: str, kind: str, source: str | None = None, asset: str | None = None, transform_ref: str | None = None, rules: dict[str, Any] = dict(), write_mode: str | None = None, description: str | None = None, tags: tuple[str, ...] = (), metadata: dict[str, Any] = dict())
One bronze, silver, or gold step in a medallion document.
adapt_pipeline(spec: SparkForgePipelineSpec, *, capabilities: PluginCapabilities | None = None, strict_delta: bool = True) -> AdaptationResult
Map a SparkForge pipeline IR to a concrete ETLantic Pipeline subclass.
adapt_profile(spec: SparkForgePipelineSpec, *, name: str | None = None, bindings: dict[str, str] | None = None) -> Profile
Build an ETLantic Profile from SparkForge builder config.
Map layer thresholds onto a named ValidationPolicy (metadata only).
Place serialized write intents under plan.intents['write_intents'].
Create an anonymous MedallionPipeline subclass from a document.
Raises:
| Type | Description |
|---|---|
ValueError
|
When a step declares a layer outside bronze/silver/gold. |
assert_delta_capabilities(operations: list[str], *, capabilities: PluginCapabilities | None = None, strict: bool = True) -> list[Diagnostic]
Fail closed when declared Delta ops are unsupported by capabilities.
When capabilities is None and strict is True (default), emit errors
so callers must supply a plugin that supports spark_delta before
execution. When strict is False (plan-only), emit warnings instead.
Normalize SparkForge retry config into ETLantic RetryPolicy.
Map a SparkForge write-mode string to ETLantic WriteMode.
Extra write-mode metadata preserved when modes collapse (e.g. partitions).
explain_medallion_plan(plan: PipelinePlan, *, definition: PipelineDefinition | None = None) -> dict[str, Any]
Merge core plan explain with medallion layer and lifecycle metadata.
enrich_lifecycle_event(event: LifecycleEvent | SecurityEvent, layer_by_node: dict[str, str]) -> LifecycleEvent | SecurityEvent
Attach medallion layer annotation when step is known.
Summarize run report steps grouped by medallion layer.
lower_document(doc: MedallionDocument, *, required_delta_operations: tuple[str, ...] = (), diagnostic_phase: str = 'medallion_authoring', allowed_transform_refs: tuple[str, ...] | None = None, transform_refs_fail_closed: bool = False, profile: Profile | None = None) -> LoweringResult
Map a medallion document to a concrete ETLantic Pipeline subclass.
Bronze/silver/gold remain facade metadata on layer_by_node; ETLantic
core never sees medallion enums.
Importable transform_ref values are deny-by-default when
transform_refs_fail_closed is True or when profile.security_mode is
production. Pass allowed_transform_refs (or profile metadata
medallantic.allowed_transform_refs) to permit specific callables.
Development profile with medallion observability defaults.
Production template — requires explicit plugin allowlist before deploy.
Test profile with strict validation and in-memory history.
adapt_run_result(payload: dict[str, Any], *, pipeline_id: str | None = None, profile: str = 'medallantic') -> PipelineRunReport
Convert a SparkForge-shaped result dict into PipelineRunReport.
Never retains secret-like keys or free-text credentials from the source
payload. Unknown run statuses fail closed to failed with diagnostic
PMSF500.
enforce_accept_rates(report: PipelineRunReport, *, policy_metadata: dict[str, Any], layer_by_node: dict[str, str] | None = None) -> PipelineRunReport
Fail a successful report when accept-rate thresholds are violated.
Returns the original report when thresholds pass or are absent. On
violation, returns a replaced report with FAILED status and
MDL120 diagnostics.
evaluate_accept_rates(*, policy_metadata: dict[str, Any], validations: list[ValidationResult] | list[dict[str, Any]], layer: str | None = None) -> list[dict[str, Any]]
Compare validation accept rates against Medallantic layer thresholds.
Threshold keys live on ValidationPolicy.metadata as
min_accept_rate_{ingest,clean,publish} (bronze/silver/gold).
Returns a list of findings; empty when thresholds are satisfied or absent.
Dual-reporting helper: ETLantic report → SparkForge-shaped explain dict.
Parse column→shorthand lists into a portable quality ruleset.
Supported shorthands (strings)::
not_null
unique / uniqueness
regex:<pattern>
length:min:max | min_length:N | max_length:N
in:a|b|c
range:min:max
ge:N / le:N / gt:N / lt:N / eq:N / ne:N
custom:<name>
Structured dict forms are also accepted per entry.
Return an ETLantic DebugSession for an adapted pipeline class.
debug_request_from_sparkforge(*, mode: str | None = 'standard', run_until: str | None = None, run_one: str | None = None, run_from: str | None = None, skip_writes: bool = False, materialization: MaterializationPolicy | None = None, retry: dict[str, Any] | RetryPolicy | None = None, parameter_overrides: dict[str, dict[str, Any]] | None = None, implementation_overrides: dict[str, str] | None = None, invalidation: str | InvalidationMode | None = None) -> RunRequest
Build a RunRequest from SparkForge debug/session options.
skip_writes sets no_write=True only. Materialization stays
DEFAULT unless the caller passes materialization= explicitly.
VALIDATE intent also sets no_write=True.
Map SparkForge run-mode strings to RunIntent.
selection_from_sparkforge(*, run_until: str | None = None, run_one: str | None = None, run_from: str | None = None) -> RunSelection
Map SparkForge selective-execution flags to RunSelection.
Lower SparkForge IR onto ETLantic via shared Medallantic lowering.
The public migrate namespace is medallantic.migrate.sparkforge. Top-level
medallantic.adapt / medallantic.ir remain compatibility re-exports.
Bases: Data
Generic row contract for medallion planning/parity (passthrough).
Bases: Exception
Raised when SparkForge → ETLantic adaptation fails closed.
dataclass
¶
AdaptationResult(pipeline_cls: type[Pipeline], profile: Profile, validation_policy: ValidationPolicy, write_intents: tuple[WriteIntent, ...] = (), step_map: dict[str, str] = dict(), layer_by_node: dict[str, str] = dict(), diagnostics: tuple[Diagnostic, ...] = (), metadata: dict[str, Any] = dict(), required_delta_operations: tuple[str, ...] = ())
Result of adapting a SparkForge pipeline IR.
Place serialized write intents under plan.intents['write_intents'].
adapt_profile(spec: SparkForgePipelineSpec, *, name: str | None = None, bindings: dict[str, str] | None = None) -> Profile
Build an ETLantic Profile from SparkForge builder config.
Map layer thresholds onto a named ValidationPolicy (metadata only).
Convert SparkForge IR into the shared MedallionDocument.
adapt_pipeline(spec: SparkForgePipelineSpec, *, capabilities: PluginCapabilities | None = None, strict_delta: bool = True) -> AdaptationResult
Map a SparkForge pipeline IR to a concrete ETLantic Pipeline subclass.
Native MedallionPipeline class authoring surfaces.
LayerStep(*, asset: str | None = None, source: str | None = None, write_mode: str | None = None, transform_ref: str | None = None, rules: dict[str, Any] | None = None, description: str | None = None, tags: tuple[str, ...] | list[str] | None = None, kind: str | None = None, metadata: dict[str, Any] | None = None, name: str | None = None)
Base descriptor for bronze/silver/gold class attributes.
Bronze(*, asset: str | None = None, source: str | None = None, write_mode: str | None = None, transform_ref: str | None = None, rules: dict[str, Any] | None = None, description: str | None = None, tags: tuple[str, ...] | list[str] | None = None, kind: str | None = None, metadata: dict[str, Any] | None = None, name: str | None = None)
Class-style native medallion authoring.
Subclass and declare Bronze / Silver / Gold attributes, then call
to_definition() or lower().
classmethod
¶
Return the facade-owned medallion document for this class.
classmethod
¶
Build an anonymous MedallionPipeline subclass from a document dict.
classmethod
¶
Lower this pipeline onto ETLantic Pipeline / definition surfaces.
classmethod
¶
Return a sealed public PipelineDefinition.
Create an anonymous MedallionPipeline subclass from a document.
Raises:
| Type | Description |
|---|---|
ValueError
|
When a step declares a layer outside bronze/silver/gold. |
Fluent MedallionBuilder for native medallion authoring.
Resolve Medallantic transform_ref callables into ETLantic Transformations.
resolve_transform_callable(transform_ref: str, *, allowed_refs: Collection[str] | None = None, fail_closed: bool = False) -> Callable[..., Any]
Import module:attr or module.attr and return the callable.
When fail_closed is True, imports are denied unless allowed_refs
explicitly lists the reference (deny-by-default). When allowed_refs is
provided without fail_closed, only listed refs are permitted.
make_callable_transformation(name: str, *, transform_ref: str, fn: Callable[..., Any] | None = None, row_type: type[Any], allowed_refs: Collection[str] | None = None, fail_closed: bool = False) -> type[Transformation]
Build a Transformation that executes transform_ref on local records.
Non-portable PySpark Column / callable quality rules (Medallantic-only).
These rules are not part of etlantic.quality/1. They require the
quality.pyspark_column engine capability and fail closed with MDL130
when the selected engine cannot execute them.
dataclass
¶
NativeColumnRule(field: str, kind: str, expr_ref: str | None = None, required: bool = True, metadata: dict[str, Any] = field(default_factory=dict))
One engine-native Column / callable validator (secret-free metadata).
Return True when a rule entry is a PySpark Column / native callable.
split_portable_and_native_rules(rules: dict[str, Any] | None) -> tuple[dict[str, Any], list[NativeColumnRule]]
Partition a rules mapping into portable shorthand and native Column rules.
SQL / Spark / Delta / write-policy compatibility mappings.
Map a SparkForge write-mode string to ETLantic WriteMode.
Extra write-mode metadata preserved when modes collapse (e.g. partitions).
Normalize SparkForge retry config into ETLantic RetryPolicy.
assert_delta_capabilities(operations: list[str], *, capabilities: PluginCapabilities | None = None, strict: bool = True) -> list[Diagnostic]
Fail closed when declared Delta ops are unsupported by capabilities.
When capabilities is None and strict is True (default), emit errors
so callers must supply a plugin that supports spark_delta before
execution. When strict is False (plan-only), emit warnings instead.
Stable Medallantic construction and graph diagnostics (MDL1xx).
mdl_diagnostic(code: str, message: str, *, severity: Severity = Severity.ERROR, path: tuple[str, ...] = (), phase: str = 'medallion_authoring') -> Diagnostic
Build a Medallantic MDL* diagnostic.
Medallion-oriented plan explain output.
explain_medallion_plan(plan: PipelinePlan, *, definition: PipelineDefinition | None = None) -> dict[str, Any]
Merge core plan explain with medallion layer and lifecycle metadata.
SparkForge-independent intermediate IR for migration mapping.
Medallion layer names live only in this adapter package — never in ETLantic core.
Bases: StrEnum
SparkForge medallion layer (adapter-only vocabulary).
Bases: StrEnum
SparkForge step kinds represented in the compatibility IR.
dataclass
¶
SparkForgeStepSpec(name: str, kind: StepKind, layer: LayerKind, source: str | None = None, table_name: str | None = None, transform_ref: str | None = None, rules: dict[str, Any] = dict(), write_mode: str | None = None, metadata: dict[str, Any] = dict())
One SparkForge step in secret-free form.
dataclass
¶
SparkForgePipelineSpec(name: str, schema: str = 'default', steps: tuple[SparkForgeStepSpec, ...] = (), min_bronze_rate: float = 90.0, min_silver_rate: float = 95.0, min_gold_rate: float = 98.0, engine: str = 'spark', legacy_engine_extensions: tuple[str, ...] = (), metadata: dict[str, Any] = dict())
Complete SparkForge pipeline description (fixture-friendly).
classmethod
¶
Parse IR and return any coercion / hygiene diagnostics.
Medallantic layer lifecycle defaults (facade-only; not core wire vocabulary).
lifecycle_policy_for_layer(*, subject_id: str, layer: str, incremental_field: str | None = None, metadata: dict[str, Any] | None = None) -> LifecyclePolicy
Return the default lifecycle policy for a medallion layer.
incremental_strategy_for_step(*, subject_id: str, layer: str, incremental_column: str | None = None, watermark_column: str | None = None) -> IncrementalStrategy | None
Map Medallantic incremental/watermark columns onto IncrementalStrategy.
Layer-aware lifecycle views over normalized ETLantic events.
Lower MedallionDocument onto ETLantic Pipeline / PipelineDefinition.
Bases: Exception
Raised when medallion → ETLantic lowering fails closed.
Bases: Data
Generic row contract for medallion planning/parity (passthrough).
dataclass
¶
LoweringResult(pipeline_cls: type[Pipeline], profile: Profile, validation_policy: ValidationPolicy, write_intents: tuple[WriteIntent, ...] = (), step_map: dict[str, str] = dict(), layer_by_node: dict[str, str] = dict(), diagnostics: tuple[Diagnostic, ...] = (), metadata: dict[str, Any] = dict(), required_delta_operations: tuple[str, ...] = ())
build_profile(doc: MedallionDocument, *, name: str | None = None, bindings: dict[str, str] | None = None) -> Profile
Build an ETLantic Profile from a medallion document.
Map layer thresholds onto a named ValidationPolicy (metadata only).
lower_document(doc: MedallionDocument, *, required_delta_operations: tuple[str, ...] = (), diagnostic_phase: str = 'medallion_authoring', allowed_transform_refs: tuple[str, ...] | None = None, transform_refs_fail_closed: bool = False, profile: Profile | None = None) -> LoweringResult
Map a medallion document to a concrete ETLantic Pipeline subclass.
Bronze/silver/gold remain facade metadata on layer_by_node; ETLantic
core never sees medallion enums.
Importable transform_ref values are deny-by-default when
transform_refs_fail_closed is True or when profile.security_mode is
production. Pass allowed_transform_refs (or profile metadata
medallantic.allowed_transform_refs) to permit specific callables.
Migration bridges from legacy builders into Medallantic.
dataclass
¶
GenerationResult(definition: PipelineDefinition | None, convertibility: Convertibility, diagnostics: tuple[Diagnostic, ...], source_fingerprint: str | None = None, adaptation: AdaptationResult | None = None)
Result of safe native generation from SparkForge IR.
dataclass
¶
InventoryArtifact(path: str, builder_kind: BuilderKind, step_count: int, convertibility: Convertibility, diagnostic_codes: tuple[str, ...] = (), source_fingerprint: str | None = None, notes: tuple[str, ...] = ())
One discovered migration candidate (secret-free).
dataclass
¶
MigrationInventoryReport(root: str, artifacts: tuple[InventoryArtifact, ...] = (), diagnostics: tuple[Diagnostic, ...] = ())
Secret-free project inventory for SparkForge migration.
generate_from_artifact(artifact: InventoryArtifact, *, root: str | Path, require_auto: bool = True) -> GenerationResult
Generate from an inventory artifact under root when auto-safe.
generate_from_ir(payload: dict[str, Any], *, source_path: str | None = None, require_auto: bool = True) -> GenerationResult
Generate a native PipelineDefinition from SparkForge IR JSON.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payload
|
dict[str, Any]
|
Secret-free SparkForge pipeline IR mapping. |
required |
source_path
|
str | None
|
Optional path label for diagnostics. |
None
|
require_auto
|
bool
|
When True, refuse generation unless convertibility is auto. |
True
|
Returns:
| Type | Description |
|---|---|
GenerationResult
|
|
Load JSON IR from disk and generate a native definition.
Scan a project tree for SparkForge / SQL builder migration candidates.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | Path
|
Project root or a single JSON/Python file. |
required |
Returns:
| Type | Description |
|---|---|
MigrationInventoryReport
|
Secret-free |
Safe native Medallantic definition generation from SparkForge IR (M7).
Generators never resolve secrets, never import untrusted project code, and
never read production tables. Only auto-safe inventory artifacts / IR payloads
are converted; manual and unsupported paths emit stable MDL* diagnostics.
dataclass
¶
GenerationResult(definition: PipelineDefinition | None, convertibility: Convertibility, diagnostics: tuple[Diagnostic, ...], source_fingerprint: str | None = None, adaptation: AdaptationResult | None = None)
Result of safe native generation from SparkForge IR.
generate_from_ir(payload: dict[str, Any], *, source_path: str | None = None, require_auto: bool = True) -> GenerationResult
Generate a native PipelineDefinition from SparkForge IR JSON.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payload
|
dict[str, Any]
|
Secret-free SparkForge pipeline IR mapping. |
required |
source_path
|
str | None
|
Optional path label for diagnostics. |
None
|
require_auto
|
bool
|
When True, refuse generation unless convertibility is auto. |
True
|
Returns:
| Type | Description |
|---|---|
GenerationResult
|
|
Load JSON IR from disk and generate a native definition.
generate_from_artifact(artifact: InventoryArtifact, *, root: str | Path, require_auto: bool = True) -> GenerationResult
Generate from an inventory artifact under root when auto-safe.
SparkForge / SQL builder project inventory scanner (Medallantic M7).
Analysis is static and secret-free: JSON IR parsing and text pattern matching only. Never imports untrusted project code, never resolves secrets, never reads production tables, and never mutates targets.
dataclass
¶
InventoryArtifact(path: str, builder_kind: BuilderKind, step_count: int, convertibility: Convertibility, diagnostic_codes: tuple[str, ...] = (), source_fingerprint: str | None = None, notes: tuple[str, ...] = ())
One discovered migration candidate (secret-free).
dataclass
¶
MigrationInventoryReport(root: str, artifacts: tuple[InventoryArtifact, ...] = (), diagnostics: tuple[Diagnostic, ...] = ())
Secret-free project inventory for SparkForge migration.
Scan a project tree for SparkForge / SQL builder migration candidates.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | Path
|
Project root or a single JSON/Python file. |
required |
Returns:
| Type | Description |
|---|---|
MigrationInventoryReport
|
Secret-free |
SparkForge IR migration namespace (medallantic.migrate.sparkforge).
dataclass
¶
AdaptationResult(pipeline_cls: type[Pipeline], profile: Profile, validation_policy: ValidationPolicy, write_intents: tuple[WriteIntent, ...] = (), step_map: dict[str, str] = dict(), layer_by_node: dict[str, str] = dict(), diagnostics: tuple[Diagnostic, ...] = (), metadata: dict[str, Any] = dict(), required_delta_operations: tuple[str, ...] = ())
Result of adapting a SparkForge pipeline IR.
Bases: Exception
Raised when SparkForge → ETLantic adaptation fails closed.
Bases: StrEnum
SparkForge medallion layer (adapter-only vocabulary).
dataclass
¶
SparkForgePipelineSpec(name: str, schema: str = 'default', steps: tuple[SparkForgeStepSpec, ...] = (), min_bronze_rate: float = 90.0, min_silver_rate: float = 95.0, min_gold_rate: float = 98.0, engine: str = 'spark', legacy_engine_extensions: tuple[str, ...] = (), metadata: dict[str, Any] = dict())
Complete SparkForge pipeline description (fixture-friendly).
classmethod
¶
Parse IR and return any coercion / hygiene diagnostics.
dataclass
¶
SparkForgeStepSpec(name: str, kind: StepKind, layer: LayerKind, source: str | None = None, table_name: str | None = None, transform_ref: str | None = None, rules: dict[str, Any] = dict(), write_mode: str | None = None, metadata: dict[str, Any] = dict())
One SparkForge step in secret-free form.
Bases: StrEnum
SparkForge step kinds represented in the compatibility IR.
LiveBridgeError(message: str, *, diagnostics: list[Diagnostic] | None = None, code: str = 'PMSF350')
Bases: Exception
Raised when a live PipelineBuilder cannot be extracted.
adapt_pipeline(spec: SparkForgePipelineSpec, *, capabilities: PluginCapabilities | None = None, strict_delta: bool = True) -> AdaptationResult
Map a SparkForge pipeline IR to a concrete ETLantic Pipeline subclass.
adapt_profile(spec: SparkForgePipelineSpec, *, name: str | None = None, bindings: dict[str, str] | None = None) -> Profile
Build an ETLantic Profile from SparkForge builder config.
Map layer thresholds onto a named ValidationPolicy (metadata only).
Place serialized write intents under plan.intents['write_intents'].
Convert SparkForge IR into the shared MedallionDocument.
assert_delta_capabilities(operations: list[str], *, capabilities: PluginCapabilities | None = None, strict: bool = True) -> list[Diagnostic]
Fail closed when declared Delta ops are unsupported by capabilities.
When capabilities is None and strict is True (default), emit errors
so callers must supply a plugin that supports spark_delta before
execution. When strict is False (plan-only), emit warnings instead.
Normalize SparkForge retry config into ETLantic RetryPolicy.
Map a SparkForge write-mode string to ETLantic WriteMode.
Extra write-mode metadata preserved when modes collapse (e.g. partitions).
from_pipeline_builder(builder: Any, *, name: str | None = None, engine: str = 'spark') -> tuple[SparkForgePipelineSpec, list[Diagnostic]]
Extract a secret-free SparkForgePipelineSpec from a live builder.
Accepts:
- objects with steps / to_dict / export attributes (SparkForge)
- plain mappings already shaped like IR
Return True when a SparkForge pipeline_builder module can be imported.
adapt_run_result(payload: dict[str, Any], *, pipeline_id: str | None = None, profile: str = 'medallantic') -> PipelineRunReport
Convert a SparkForge-shaped result dict into PipelineRunReport.
Never retains secret-like keys or free-text credentials from the source
payload. Unknown run statuses fail closed to failed with diagnostic
PMSF500.
Dual-reporting helper: ETLantic report → SparkForge-shaped explain dict.
Return an ETLantic DebugSession for an adapted pipeline class.
debug_request_from_sparkforge(*, mode: str | None = 'standard', run_until: str | None = None, run_one: str | None = None, run_from: str | None = None, skip_writes: bool = False, materialization: MaterializationPolicy | None = None, retry: dict[str, Any] | RetryPolicy | None = None, parameter_overrides: dict[str, dict[str, Any]] | None = None, implementation_overrides: dict[str, str] | None = None, invalidation: str | InvalidationMode | None = None) -> RunRequest
Build a RunRequest from SparkForge debug/session options.
skip_writes sets no_write=True only. Materialization stays
DEFAULT unless the caller passes materialization= explicitly.
VALIDATE intent also sets no_write=True.
Map SparkForge run-mode strings to RunIntent.
Map SparkForge rerun/invalidation strings to InvalidationMode.
selection_from_sparkforge(*, run_until: str | None = None, run_one: str | None = None, run_from: str | None = None) -> RunSelection
Map SparkForge selective-execution flags to RunSelection.
Live SparkForge PipelineBuilder → secret-free IR extraction.
SparkForge is an optional dependency. When it is not installed, callers
should use frozen IR fixtures via :func:medallantic.adapt.adapt_pipeline.
This module never embeds secrets or source rows.
LiveBridgeError(message: str, *, diagnostics: list[Diagnostic] | None = None, code: str = 'PMSF350')
Bases: Exception
Raised when a live PipelineBuilder cannot be extracted.
Return True when a SparkForge pipeline_builder module can be imported.
from_pipeline_builder(builder: Any, *, name: str | None = None, engine: str = 'spark') -> tuple[SparkForgePipelineSpec, list[Diagnostic]]
Extract a secret-free SparkForgePipelineSpec from a live builder.
Accepts:
- objects with steps / to_dict / export attributes (SparkForge)
- plain mappings already shaped like IR
SQL pipeline-builder IR migration namespace (medallantic.migrate.sql).
dataclass
¶
AdaptationResult(pipeline_cls: type[Pipeline], profile: Profile, validation_policy: ValidationPolicy, write_intents: tuple[WriteIntent, ...] = (), step_map: dict[str, str] = dict(), layer_by_node: dict[str, str] = dict(), diagnostics: tuple[Diagnostic, ...] = (), metadata: dict[str, Any] = dict(), required_delta_operations: tuple[str, ...] = ())
Result of adapting a SparkForge pipeline IR.
Bases: Exception
Raised when SparkForge → ETLantic adaptation fails closed.
Bases: StrEnum
SparkForge medallion layer (adapter-only vocabulary).
dataclass
¶
SparkForgePipelineSpec(name: str, schema: str = 'default', steps: tuple[SparkForgeStepSpec, ...] = (), min_bronze_rate: float = 90.0, min_silver_rate: float = 95.0, min_gold_rate: float = 98.0, engine: str = 'spark', legacy_engine_extensions: tuple[str, ...] = (), metadata: dict[str, Any] = dict())
Complete SparkForge pipeline description (fixture-friendly).
classmethod
¶
Parse IR and return any coercion / hygiene diagnostics.
dataclass
¶
SparkForgeStepSpec(name: str, kind: StepKind, layer: LayerKind, source: str | None = None, table_name: str | None = None, transform_ref: str | None = None, rules: dict[str, Any] = dict(), write_mode: str | None = None, metadata: dict[str, Any] = dict())
One SparkForge step in secret-free form.
Bases: StrEnum
SparkForge step kinds represented in the compatibility IR.
LiveBridgeError(message: str, *, diagnostics: list[Diagnostic] | None = None, code: str = 'PMSQ350')
Bases: Exception
Raised when a live SqlPipelineBuilder cannot be extracted.
adapt_pipeline(spec: SparkForgePipelineSpec, *, capabilities: PluginCapabilities | None = None, strict_delta: bool = True) -> AdaptationResult
Map a SparkForge pipeline IR to a concrete ETLantic Pipeline subclass.
adapt_profile(spec: SparkForgePipelineSpec, *, name: str | None = None, bindings: dict[str, str] | None = None) -> Profile
Build an ETLantic Profile from SparkForge builder config.
Map layer thresholds onto a named ValidationPolicy (metadata only).
Place serialized write intents under plan.intents['write_intents'].
Convert SparkForge IR into the shared MedallionDocument.
Map a SparkForge write-mode string to ETLantic WriteMode.
Extra write-mode metadata preserved when modes collapse (e.g. partitions).
from_sql_pipeline_builder(builder: Any, *, name: str | None = None, engine: str = 'sql') -> tuple[SparkForgePipelineSpec, list[Diagnostic]]
Extract a secret-free SparkForgePipelineSpec from a live SQL builder.
Accepts:
- objects with steps / to_dict / export attributes
- SQLAlchemy Select-like objects
- plain mappings already shaped like IR
Return True when a SQL pipeline builder module can be imported.
adapt_run_result(payload: dict[str, Any], *, pipeline_id: str | None = None, profile: str = 'medallantic') -> PipelineRunReport
Convert a SparkForge-shaped result dict into PipelineRunReport.
Never retains secret-like keys or free-text credentials from the source
payload. Unknown run statuses fail closed to failed with diagnostic
PMSF500.
Dual-reporting helper: ETLantic report → SparkForge-shaped explain dict.
Live SqlPipelineBuilder → secret-free IR extraction.
SqlPipelineBuilder / Moltres are optional dependencies. When they are not
installed, callers should use frozen IR fixtures via
:func:medallantic.adapt.adapt_pipeline. This module never embeds secrets or
source rows.
LiveBridgeError(message: str, *, diagnostics: list[Diagnostic] | None = None, code: str = 'PMSQ350')
Bases: Exception
Raised when a live SqlPipelineBuilder cannot be extracted.
Return True when a SQL pipeline builder module can be imported.
from_sql_pipeline_builder(builder: Any, *, name: str | None = None, engine: str = 'sql') -> tuple[SparkForgePipelineSpec, list[Diagnostic]]
Extract a secret-free SparkForgePipelineSpec from a live SQL builder.
Accepts:
- objects with steps / to_dict / export attributes
- SQLAlchemy Select-like objects
- plain mappings already shaped like IR
Non-portable Moltres / SQL-native quality rules (Medallantic-only).
These rules are not part of etlantic.quality/1. They require the
quality.moltres_expr engine capability and fail closed with MDL132
when the selected engine cannot execute them.
dataclass
¶
NativeMoltresRule(field: str, kind: str, expr_ref: str | None = None, required: bool = True, metadata: dict[str, Any] = field(default_factory=dict))
One engine-native Moltres / SQLAlchemy expression validator.
Return True when a rule entry is a Moltres / SQLAlchemy-native validator.
split_portable_and_moltres_rules(rules: dict[str, Any] | None) -> tuple[dict[str, Any], list[NativeMoltresRule]]
Partition a rules mapping into portable shorthand and Moltres rules.
Medallion profile templates composing core production envelopes.
Normalize SparkForge-shaped run results into PipelineRunReport.
adapt_run_result(payload: dict[str, Any], *, pipeline_id: str | None = None, profile: str = 'medallantic') -> PipelineRunReport
Convert a SparkForge-shaped result dict into PipelineRunReport.
Never retains secret-like keys or free-text credentials from the source
payload. Unknown run statuses fail closed to failed with diagnostic
PMSF500.
Dual-reporting helper: ETLantic report → SparkForge-shaped explain dict.
evaluate_accept_rates(*, policy_metadata: dict[str, Any], validations: list[ValidationResult] | list[dict[str, Any]], layer: str | None = None) -> list[dict[str, Any]]
Compare validation accept rates against Medallantic layer thresholds.
Threshold keys live on ValidationPolicy.metadata as
min_accept_rate_{ingest,clean,publish} (bronze/silver/gold).
Returns a list of findings; empty when thresholds are satisfied or absent.
enforce_accept_rates(report: PipelineRunReport, *, policy_metadata: dict[str, Any], layer_by_node: dict[str, str] | None = None) -> PipelineRunReport
Fail a successful report when accept-rate thresholds are violated.
Returns the original report when thresholds pass or are absent. On
violation, returns a replaced report with FAILED status and
MDL120 diagnostics.
Medallantic shorthand rule DSL → etlantic.quality/1.
Bases: ValueError
Raised when a Medallantic rules shorthand cannot be parsed.
Parse column→shorthand lists into a portable quality ruleset.
Supported shorthands (strings)::
not_null
unique / uniqueness
regex:<pattern>
length:min:max | min_length:N | max_length:N
in:a|b|c
range:min:max
ge:N / le:N / gt:N / lt:N / eq:N / ne:N
custom:<name>
Structured dict forms are also accepted per entry.
Map SparkForge debug / run modes onto ETLantic RunRequest surfaces.
Map SparkForge run-mode strings to RunIntent.
selection_from_sparkforge(*, run_until: str | None = None, run_one: str | None = None, run_from: str | None = None) -> RunSelection
Map SparkForge selective-execution flags to RunSelection.
Map SparkForge rerun/invalidation strings to InvalidationMode.
debug_request_from_sparkforge(*, mode: str | None = 'standard', run_until: str | None = None, run_one: str | None = None, run_from: str | None = None, skip_writes: bool = False, materialization: MaterializationPolicy | None = None, retry: dict[str, Any] | RetryPolicy | None = None, parameter_overrides: dict[str, dict[str, Any]] | None = None, implementation_overrides: dict[str, str] | None = None, invalidation: str | InvalidationMode | None = None) -> RunRequest
Build a RunRequest from SparkForge debug/session options.
skip_writes sets no_write=True only. Materialization stays
DEFAULT unless the caller passes materialization= explicitly.
VALIDATE intent also sets no_write=True.
Return an ETLantic DebugSession for an adapted pipeline class.
Facade-owned medallion document schema (not an ETLantic wire schema).
dataclass
¶
MedallionStep(name: str, layer: str, kind: str, source: str | None = None, asset: str | None = None, transform_ref: str | None = None, rules: dict[str, Any] = dict(), write_mode: str | None = None, description: str | None = None, tags: tuple[str, ...] = (), metadata: dict[str, Any] = dict())
One bronze, silver, or gold step in a medallion document.
dataclass
¶
MedallionDocument(name: str, schema: str = 'default', steps: tuple[MedallionStep, ...] = (), min_bronze_rate: float = 90.0, min_silver_rate: float = 95.0, min_gold_rate: float = 98.0, engine: str = 'local', description: str | None = None, tags: tuple[str, ...] = (), metadata: dict[str, Any] = dict())
Declarative medallion pipeline document owned by Medallantic.