wiring.build_asset_check_specs()

Return the asset check specs for the schema’s rules, plus the column-schema check.

Usage

wiring.build_asset_check_specs(
    schema,
    *,
    asset,
    partitions_def=None,
    check_granularity=None,
    filter=False
)

Parameters

schema: type[dy.Schema]: The schema whose rules become check specs.
asset: str | dg.AssetKey: The key of the asset the checks report on.
partitions_def: dg.PartitionsDefinition | None = None: The asset’s partitions, so each check evaluation records its partition. Dagster does not check that it matches the asset’s.
check_granularity: Granularity | None = None: How many checks report the schema’s rules. Changing it on an asset that has already run starts a new check history. None uses DAGSTER_DATAFRAMELY_CHECK_GRANULARITY if set, else rule.
filter: bool = False: Pass True when the asset writes its error rows instead of failing, as dd.asset(filter=True) does. With False, a check that reports for an "error" rule is blocking. With True, no rule’s check is blocking, because a blocking check that fails at ERROR fails the step after the table is written.

Returns

list[dg.AssetCheckSpec]: The column-schema check’s spec first, then one spec per rule set, to pass to @dg.asset(check_specs=...).

Raises

InvalidSettingError: A setting’s argument or environment variable has a value the setting does not allow.
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.

Examples

import dagster_dataframely as dd
from dagster_dataframely.data import OrdersSchema

specs = dd.wiring.build_asset_check_specs(OrdersSchema, asset="orders")
[(spec.name, spec.blocking) for spec in specs][:3]

[('dy_schema__columns', True),
 ('dy_rule__paid_orders_have_amount', True),
 ('dy_rule__line_numbers_are_dense', True)]