Schemas & normalization¶

A schema is an asset's data contract; a normalizer coerces the vendor's rows into it and a materialization strategy decides how they land.

schema ¶

schema(target: Any = None, /, **settings: Any) -> Any

Create a Schema subclass from a decorated class.

Can be used bare or with arguments:

@schema
class UserSchema:
    id: int
    name: str
    email: str

@schema(key="custom", name="Custom Schema")
class OtherSchema:
    value: float

The decorated class's annotations and attributes become the Schema subclass body. key overrides the auto-derived snake_cased key and name is the human-readable display name.

Schema ¶

Schema(**data: Any)

Bases: interloper.serializable.base.Serializable

Defines the expected output structure of an asset.

Subclass to declare output fields:

class UserSchema(Schema):
    id: int
    name: str
    email: str

Class methods provide schema operations on list[dict] data:

Schema.infer(rows)
UserSchema.validate_rows(rows)
UserSchema.reconcile(rows)

key ¶

key = 'schema'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

model_config ¶

model_config = {'arbitrary_types_allowed': True, 'ignored_types': (<class 'interloper.serializable.base.IgnoredDescriptor'>,)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

field_specs ¶

field_specs() -> list[FieldSpec]

Extract backend-agnostic field specs from this schema.

Fields declared by Schema itself (framework plumbing, if any) are excluded — only the subclass's data columns count. Fields come out in declaration order: Serializable normalizes model_fields so a column shadowing a parent ClassVar (name, key) is not hoisted out of place.

Returns:

Type	Description
`list[FieldSpec]`	One `FieldSpec` per declared data field, in declaration order.

infer ¶

infer(rows: list[dict[str, Any]]) -> type[Self]

Infer a Schema subclass from a list of row dicts.

Examines the values across all rows for each key and maps Python types to Pydantic field types. All fields are Optional because any key may be absent in some rows.

Parameters:

Name	Type	Description	Default
`rows`	`list[dict[str, Any]]`	Non-empty list of dicts to infer from.	required

Returns:

Type	Description
`type[Self]`	A dynamically created Schema subclass.

Raises:

Type	Description
`SchemaError`	If rows is empty.

json_schema ¶

json_schema() -> dict[str, Any]

JSON Schema for this schema's data fields only.

Like pydantic.BaseModel.model_json_schema but restricted to the subclass's data columns, in declaration order (see field_specs).

Returns:

Type	Description
`dict[str, Any]`	A JSON Schema dict whose `properties` are the data columns.

reconcile ¶

reconcile(
    rows: list[dict[str, Any]],
) -> list[dict[str, Any]]

Reconcile rows against this schema.

Fields are coerced one by one through per-field TypeAdapters — the row-wise mirror of the DataFrame representation's column-wise casts. Extra keys are dropped (their union logged as a warning), missing fields get their default (or None, which pydantic accepts only for nullable fields), and scalar str fields are stringified up front since pydantic's lax mode never coerces to str.

This is more permissive than validate_rows: it actively transforms data to match the schema rather than rejecting mismatches.

Parameters:

Name	Type	Description	Default
`rows`	`list[dict[str, Any]]`	List of row dicts.	required

Returns:

Type	Description
`list[dict[str, Any]]`	A new list of row dicts with columns aligned and types coerced.

Raises:

Type	Description
`SchemaError`	If a value cannot be coerced (e.g. `"abc"` → `int`) or a required non-nullable field is missing.

validate_rows ¶

validate_rows(
    rows: list[dict[str, Any]], *, strict: bool = False
) -> None

Validate each row against this schema.

Stops at the first row that fails validation.

Parameters:

Name	Type	Description	Default
`rows`	`list[dict[str, Any]]`	List of row dicts.	required
`strict`	`bool`	When `True`, reject rows that contain keys not defined in the schema and rows that are missing required schema fields.	`False`

Raises:

Type	Description
`SchemaError`	If any row fails validation.

Normalizer ¶

Normalizer(**data: Any)

Bases: interloper.serializable.base.Serializable

Type-native normalizer for list[dict] asset data.

Applies optional transformations to rows (column-name normalization, nested-dict flattening, missing-column fill). The shapes an asset may return besides rows (a generator, models, a lone dict) are unwrapped by the asset before normalization, so a normalizer only ever reshapes rows.

Normalizer is Serializable so instances round-trip through Spec with their concrete subclass intact — e.g. across the host → child-pod DAG-spec boundary.

Usage:

@asset(normalizer=Normalizer())
def my_asset(context):
    return [{"UserName": "alice", "Address": {"City": "NYC"}}]

Attributes:

Name	Type	Description
`normalize_columns_names`		Convert column names to snake_case.
`flatten_max_level`		Maximum nesting depth to flatten. `0` disables flattening, `None` flattens without limit, a positive `int` flattens up to that many levels.
`flatten_separator`		Separator for flattened key names.
`fill_missing`		Fill missing keys across rows with `None` so every row has the same columns.
`replace_empty_dicts`		Replace `{}` values with `None`.
`replace_empty_strings`		Replace `""` values with `None`.
`drop_na_columns`		Drop columns where every value is `None`.
`snake_case_digits`		Treat digit groups as words when snake-casing column names (`acosClicks14d` → `acos_clicks_14d` instead of `acos_clicks14d`). Off by default because the right convention depends on the upstream API's naming semantics.
`column_overrides`		Explicit raw-name → normalized-name mapping, applied before the snake_case conversion. For API quirk names no general rule can handle (`eCPAddToCart` → `ecp_add_to_cart`).

key ¶

key = 'normalizer'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

model_config ¶

model_config = {'arbitrary_types_allowed': True, 'ignored_types': (<class 'interloper.serializable.base.IgnoredDescriptor'>,)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

column_name ¶

column_name(name: str) -> str

Transform a column name according to the normalizer's convention.

Explicit column_overrides win; otherwise converts to snake_case, optionally splitting letter→digit boundaries (see snake_case_digits).

Parameters:

Name	Type	Description	Default
`name`	`str`	Original column name.	required

Returns:

Type	Description
`str`	Transformed column name.

normalize ¶

normalize(data: Any) -> list[dict[str, Any]]

Normalize rows with the configured transformations.

Applies flatten, column rename, and fill-missing in order.

Parameters:

Name	Type	Description	Default
`data`	`Any`	The rows to normalize.	required

Returns:

Type	Description
`list[dict[str, Any]]`	Normalized list of row dicts.

Raises:

Type	Description
`NormalizerError`	If data is not a list of rows.

MaterializationStrategy ¶

Bases: builtins.str, enum.Enum

How the conform step enforces an asset's schema.

Both strategies hand destinations data in the schema's canonical types; they differ in what counts as an error.

Attributes:

Name	Type	Description
`RECONCILE`		The default. Align columns to the schema (extras dropped with a warning, missing nullables filled) and coerce values. With no schema declared, infer one from the data for the destinations.
`STRICT`		Schema required. Extra columns, missing required columns and values the schema rejects fail the materialization; values that pass are still coerced to the declared types.

RECONCILE ¶

RECONCILE = <MaterializationStrategy.RECONCILE: 'reconcile'>

How the conform step enforces an asset's schema.

Both strategies hand destinations data in the schema's canonical types; they differ in what counts as an error.

Attributes:

Name	Type	Description
`RECONCILE`		The default. Align columns to the schema (extras dropped with a warning, missing nullables filled) and coerce values. With no schema declared, infer one from the data for the destinations.
`STRICT`		Schema required. Extra columns, missing required columns and values the schema rejects fail the materialization; values that pass are still coerced to the declared types.

STRICT ¶

STRICT = <MaterializationStrategy.STRICT: 'strict'>

How the conform step enforces an asset's schema.

Both strategies hand destinations data in the schema's canonical types; they differ in what counts as an error.

Attributes:

Name	Type	Description
`RECONCILE`		The default. Align columns to the schema (extras dropped with a warning, missing nullables filled) and coerce values. With no schema declared, infer one from the data for the destinations.
`STRICT`		Schema required. Extra columns, missing required columns and values the schema rejects fail the materialization; values that pass are still coerced to the declared types.