build_error_rows_spec()

Return an asset spec that adds an asset’s error rows to the asset graph.

Usage

build_error_rows_spec(
    schema,
    asset,
    *,
    partitions_def=None,
    group_name=None,
    owners=None,
    tags=None
)

The asset writes the error rows whether or not the spec exists.

Parameters

schema: type[dy.Schema]: The asset’s schema. The spec’s Columns tab has its columns, then one rule column per rule.
asset: dg.AssetsDefinition | dg.AssetKey | str | Sequence[str]: The asset the error rows came from, as its definition or its asset key. Pass a key only for an asset you cannot import. A definition also sets the spec’s partitions, group, owners and tags, and the error-rows key, IO manager key and definition metadata of its writer.
partitions_def: dg.PartitionsDefinition | None = None: The error rows’ partitions, when asset is a key.
group_name: str | None = None: The error rows’ asset group. A definition’s group is the default, and a value overrides it. Unlike partitions_def, passing it with a definition does not raise, because the group sets only where the catalog lists the error rows.
owners: Sequence[str] | None = None: The error rows’ owners. A definition’s owners are the default, and a value replaces them, as group_name does.
tags: Mapping[str, str] | None = None: The error rows’ tags. A definition’s tags are the default, and a value replaces them, as group_name does.

Returns

dg.AssetSpec: A spec keyed by the error-rows key that depends on asset, to pass to dg.Definitions(assets=[...]).

Raises

ReservedColumnError: A column name is in the reserved dy_ namespace.
InvalidColumnNameError: A column name has a character Dagster does not allow in an asset check name.
CheckNameCollisionError: Two rules produce the same asset check name.
InvalidSeverityError: The schema declares a severity this package cannot read.
dg.DagsterInvariantViolationError: asset is a definition and the call also passes partitions_def.

Examples

import polars as pl

import dagster_dataframely as dd
from dagster_dataframely.data import OrdersSchema


@dd.asset(OrdersSchema, filter=True)
def orders(raw_orders: pl.DataFrame) -> pl.DataFrame:
    return raw_orders


dd.build_error_rows_spec(OrdersSchema, orders).key

AssetKey(['orders__error_rows'])

Pass the spec, not its key, to dg.Definitions(assets=[...]) with orders. A downstream asset can then declare the error rows as an input.