Transformation¶

Pydantic models driving declarative bronze→silver / silver→gold transformation steps.

class scamander.models.transformation_config.transformation.AggregationSpec(*, column: str, function: Literal['sum', 'count', 'avg', 'mean', 'min', 'max', 'first', 'last', 'count_distinct', 'stddev', 'variance', 'collect_list', 'collect_set'], alias: str)¶

Bases: BaseModel

Defines one aggregation to apply in an aggregate transformation step.

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class scamander.models.transformation_config.transformation.ColumnMapping(*, source: str, target: str, type: str | None = None, nullable: bool = True, precision: int | None = None, scale: int | None = None)¶

Bases: BaseModel

Maps a source column to a target column with optional type casting.

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class scamander.models.transformation_config.transformation.JoinSpec(*, table: str, on: str | list[str], how: Literal['inner', 'left', 'right', 'full', 'left_semi', 'left_anti', 'cross'] = 'inner', broadcast: bool = False)¶

Bases: BaseModel

Defines a join to another Delta table.

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class scamander.models.transformation_config.transformation.TransformationConfig(*, source_table: str, target_table: str, column_mappings: list[ColumnMapping] | None = None, trim_strings: bool = False, replace_empty_strings: bool = False, filter_null_columns: list[str] | None = None, drop_duplicates: bool = False, duplicate_subset: list[str] | None = None, add_audit_columns: bool = True, write_mode: Literal['overwrite', 'append', 'error', 'ignore'] = 'overwrite', track_history: bool = False, key_columns: list[str] | None = None, joins: list[JoinSpec] | None = None, group_by: list[str] | None = None, aggregations: list[AggregationSpec] | None = None)¶

Bases: BaseModel

Drives a single bronze→silver or silver→gold transformation step.

Intended to be parsed from the same JSON/YAML/CSV config files used throughout the framework via platform.read_config().

Example JSON:

{
  "source_table": "bronze.customer",
  "target_table": "silver.customer",
  "column_mappings": [
    {"source": "klant_id", "target": "customer_id", "type": "INT", "nullable": false},
    {"source": "klant_naam", "target": "customer_name", "type": "VARCHAR"}
  ],
  "trim_strings": true,
  "replace_empty_strings": true,
  "filter_null_columns": ["customer_id"],
  "drop_duplicates": true,
  "duplicate_subset": ["customer_id"],
  "add_audit_columns": true,
  "write_mode": "overwrite"
}

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].