dagster-dataframely

dataframely declares what a polars frame has to look like. Dagster has first-class surfaces for exactly that: the Columns tab and asset checks. This package connects them with one declaration.

import dagster as dg
import dataframely as dy
import polars as pl

import dagster_dataframely as dd


class Orders(dy.Schema):
    order_id = dy.String(primary_key=True)
    amount = dy.Float64(nullable=False, min=0.0)


@dd.dataframely_asset(schema=Orders)
def orders(raw_orders: pl.DataFrame) -> pl.DataFrame:
    return raw_orders.select("order_id", "amount")

That is the whole integration.

Before the asset has ever run, Orders fills the catalog's Columns tab: dtypes, descriptions, nullability, uniqueness, the primary key stated once at table level, and one pill per remaining constraint. Every run reports one asset check per dataframely rule, each with its own pass/fail history, behind a blocking gate check that compares the frame's columns and dtypes against the schema before a single row is filtered. Add quarantine=dg.AssetOut() and the rows the schema rejects land in a sibling asset instead of failing the run, as long as something survives.

The transform keeps plain polars annotations. Upstream assets bind as ordinary parameters, the return may be a DataFrame or a LazyFrame, and there is no context parameter to write: the wrapper reaches the context itself.

Installation

uv add dagster-dataframely

Requires Python 3.12+.

[!NOTE] Pre-1.0. The public surface is pinned by a test rather than held by convention, so it will not move quietly. It can still move: a 0.x minor release is where a breaking change lands. Pin dagster-dataframely>=0.1,<0.2 if that matters to you.

dagster, dataframely and polars are the dependencies. The IO managers import pydantic and universal-pathlib directly, so both are declared too; both already arrive with dagster, so nothing new lands in your environment. Writing to s3://, gs:// or az:// needs that scheme's fsspec filesystem installed alongside: s3fs, gcsfs or adlfs.

The failure policy is the asset's declared shape

There is no lenient mode and no strict flag, deliberately. Declaring a quarantine is the consent to partial data, so what a rejected row costs is visible in the definition and cannot disagree with what the asset declares.

Without a quarantine, every row has to be good. A run that rejects even one row fails and writes nothing, so the last-known-good table stays in place. Landing the survivors and dropping the rest is the failure this package exists to make visible, so it is not reachable by configuration. To drop rows anyway, drop them in your own asset body, where the drop is a line you wrote:

@dd.dataframely_asset(schema=Orders)
def orders(raw_orders: pl.DataFrame) -> pl.DataFrame:
    good, _ = Orders.filter(raw_orders)
    return good

The quarantine table carries the rejected rows with the original columns, then one String column per rule reading valid, invalid or unknown, named exactly as that rule's asset check. Its materialization also carries a cooccurrence table, so one broken upstream field tripping three rules reads as one row rather than three unrelated counts. It inherits the good asset's key prefix, group and IO manager, and its own dg.AssetOut can override any of them, which is how rejected rows reach a separate storage and ownership domain. Three settings raise QuarantineSettingError instead of being honoured: automation_condition, freshness_policy and code_version cannot differ between two outs of one step, so they belong on the decorator, where they cover both.

The package never casts

The gate compares the frame's dtypes against the schema's and aborts on a mismatch; the filter runs with cast=False. A Duration('ns') arriving where the schema declares Duration('us') is a pipeline defect, and silently widening it is how a thousandfold error reaches a table nobody re-reads.

If you do want conformance, write the cast yourself, in your own asset body, as a line you can see:

@dd.dataframely_asset(schema=Orders)
def orders(raw_orders: pl.DataFrame) -> pl.DataFrame:
    return Orders.cast(raw_orders)

The one cast the package makes is on columns it generated itself: the quarantine's outcome columns go from Enum to String, because a raw Enum panics the Delta writer.

Storage

defs = dg.Definitions(
    assets=[orders],
    resources={
        "io_manager": dd.DataframelyParquetIOManager(
            base_dir="s3://my-bucket/warehouse"
        )
    },
)

DataframelyParquetIOManager and DataframelyCSVIOManager write under a universal-pathlib base_dir, so a local directory and s3://, gs:// or az:// are the same call on credentials from the ambient environment. A dtype the format cannot hold raises UnwritableDtypeError before the write rather than from inside polars.

These are the supported path. They record path, bytes_written and dagster/storage_kind on each materialization and nothing else. No column schema, in particular: the asset definition owns what the data is, and leaving the materialization bucket empty is what keeps the Columns tab showing the schema as declared.

What degrades on someone else's manager is one surface. A stock polars IO manager writes its own dagster/column_schema onto the materialization, and with both metadata buckets populated Dagster merges them in the catalog Columns tab: column names come out lowercased and the constraint pills disappear, because the materialization's constraint-free schema becomes the base. Dtypes, descriptions and column tags survive, and the Lineage Metadata accordion is definition-only and never merged, so full fidelity is still one click away. This is documented, not designed around. Parity is a preference, never a constraint.

CSV without the losses

A CSV cell holds text, so five dtypes have nowhere to land. Each is encoded on the way out and decoded on the way back by its declared inverse, so the frame read back compares equal to the frame written.

A run log line names the encoded columns on both paths, because that is the only surface that can carry it.

Two cases are refused rather than encoded: Binary and Duration inside a List, Array or Struct. Polars cannot write the first to JSON and cannot read the second back, so encoding either would land a file that no longer round-trips. Object is refused by both managers.

The read needs the schema. The decode reads each column's declared dtype off the carrier the decorator puts in the asset's definition metadata, which dd.schema_metadata also builds for a plain @dg.asset. With no schema in reach the read is an ordinary inferred CSV read, and an encoded column arrives as text. The schema never comes from a sidecar file and never from the data, so it costs no round trip and cannot drift. The carrier holds the live class rather than a copy of it, and a live object does not cross a process boundary, so a manager that only reaches a deserialized definition falls back to the inferred read too.

Prefer parquet unless something downstream needs text. Parquet is self-describing, keeps every dtype natively, and needs no schema to read.

Validation materializes

Schema.filter collects, so a validated frame is a frame in memory. A LazyFrame return is accepted and collected, and the IO managers collect before the write with a warning in the run log. Sinking lazily is not on the supported path; it is tracked in issue #27.

The habitat is post-landing transformation: bronze to silver to gold, where the data is already on your side and the question is whether it is fit to publish. Ingestion-scale and larger-than-memory work belongs to other tools.

Settings

Every knob resolves through three tiers, each overriding the one before: the package default, then an environment variable, then the argument on the asset. A platform engineer sets a house style once for a whole code location, and an asset overrides it where that style is wrong. Each variable is DAGSTER_DATAFRAMELY_ plus the setting's name, upper-cased.

The chain validates on resolve, at every tier including the package's own, so a typo raises InvalidSettingError naming the value and the tier that supplied it rather than quietly becoming something else three modules later.

Changing check_granularity orphans check history

rule gives every rule its own check and its own timeline. column gives one check per rule-bearing column, dy_col__<column>, which is what makes a 40-column schema's check list readable. schema gives a single dy_schema__rules for all of them.

Changing it on an asset that has already run orphans that asset's check history. The old check names stop being reported and their timelines end where the change landed, while the new ones start empty. Nothing migrates them, so choose it before the asset ships rather than after.

Statistics and both samples are on by default

Each materialization carries a skimr-style profile of what it wrote: one table per dtype family present, under stats/numeric, stats/temporal, stats/string and stats/boolean, on the quarantine as well as the good table.

[!IMPORTANT] Two of the settings write real rows of your data into the Dagster event log, and both ship on. The event log is shared across a deployment, it is exportable, and nothing here is redacted. If a column holds an email address, a name or an account number, that value lands in the log and stays there.

The quarantine carries no row sample, deliberately. Its rows already reach the event log once, through the check that rejected each of them and with the rule attached, so a second copy would carry less and cost the same.

dataframely's own comparable setting defaults to 0, so this package is deliberately the more generous of the two. The reason is that a red check raises exactly one question the counts cannot answer: not that amount|min rejected 43 rows, but what three of those rows held. Paying for that in the event log should be a decision, which is what this section is for.

Setting either to 0 turns it off entirely, and the metadata key is then absent rather than empty. Per asset:

@dd.dataframely_asset(schema=Orders, max_failure_samples=0, row_sample=0)
def orders(raw_orders: pl.DataFrame) -> pl.DataFrame:
    return raw_orders.select("order_id", "amount")

Or once for a whole code location, in the deployment's environment:

DAGSTER_DATAFRAMELY_MAX_FAILURE_SAMPLES=0
DAGSTER_DATAFRAMELY_ROW_SAMPLE=0

Turning the samples off leaves statistics on. The string family deliberately carries no value-bearing statistic at any setting, only lengths and cardinality: consenting to summary statistics is not consenting to raw values.

The kit

The decorator is one arrangement of parts the package also exports on their own: check_specs, schema_metadata, table_schema, quarantine_table_schema, quarantine_frame, process and check_name. Reach for them when the decorator's shape is not the shape you need: a schema attached to an asset you did not declare, or an out arrangement the door does not offer. Then the @dg.multi_asset is yours to wire, out of the same parts.

@dg.multi_asset(
    outs={
        "orders": dg.AssetOut(metadata=dd.schema_metadata(Orders), is_required=False)
    },
    check_specs=dd.check_specs(Orders, asset="orders"),
)
def orders():
    yield from dd.process(
        Orders,
        transform(),
        context=dg.AssetExecutionContext.get(),
        good_out="orders",
    )

is_required=False matters: the gate and both abort paths end the step without yielding an out.

This is not a route to dy.Collection support. Hand-wiring a Collection means reimplementing the hardest part of the package rather than assembling it, because process is single-schema by signature. Declare one asset per member instead, each with the member's own schema. Passing a Collection to the decorator raises CollectionNotSupportedError at decoration time.

Two user-side traps

A from __future__ import annotations in your own module breaks a context parameter. Under PEP 563 every annotation reaches Dagster as a string, and its check on the context parameter compares against the real classes, so it rejects context: dg.AssetExecutionContext and context: AssetExecutionContext alike:

DagsterInvalidDefinitionError: Cannot annotate `context` parameter with type dg.AssetExecutionContext.
`context` must be annotated with AssetExecutionContext, AssetCheckExecutionContext, OpExecutionContext, or left blank.

Only the last option in that message survives PEP 563: leave the parameter blank, or take no context at all and reach it with dg.AssetExecutionContext.get(). Decorator call sites avoid this structurally, because the wrapper fetches the context and your transform never takes one. Kit call sites are yours to write, so this one is yours to hit.

A @dy.rule() body needs its class parameter. Written without one, the class still builds and the asset still defines; the run then fails when this package reads the rule's expression for the check metadata:

TypeError: Orders.amount_is_positive() takes 0 positional arguments but 1 was given

@dy.rule() is a classmethod-style decorator, so the body takes cls:

class Orders(dy.Schema):
    status = dy.String(nullable=False)
    amount = dy.Float64(nullable=False)

    @dy.rule()
    def paid_orders_have_amount(cls) -> pl.Expr:
        """Paid orders must carry a positive amount."""
        return (cls.status.col != "paid") | (cls.amount.col > 0)

The docstring is not decoration: it becomes that check's description in the catalog.

The reserved namespace

Every check name sits under dy_, and so does every quarantine outcome column and every key in check metadata: the gate check dy_schema__dtypes, the rule checks dy_rule__<rule>, the collapsed checks dy_col__<column> and dy_schema__rules. The materialization keys are deliberately outside it, because sample, cooccurrence and stats/* are for a data consumer rather than for this package's bookkeeping. A schema with a column of its own inside the namespace raises ReservedColumnError at definition time, and two rules that rewrite to one check name raise CheckNameCollisionError. The prefix is hardcoded rather than configurable: its whole value is being the same string in every project.

License

Apache-2.0