Assets¶

An asset is a function that produces a data artifact. Assets form a directed acyclic graph (DAG) where dependencies are declared through function parameters.

rivers supports four asset types:

Every type can also be defined as a class — configuration as class attributes, verbs as classmethods, shared pieces inherited through base classes. Both forms desugar to the same objects; see Class-Based Assets.

Single assets¶

The most common type. A function decorated with @Asset that produces one output:

import rivers as rs

@rs.Asset
def raw_data():
    return {"key": "value"}

@rs.Asset(name="clean_data", tags=["etl"], kinds="table", group="pipeline")
def clean(raw_data: dict):
    return {k: v.upper() for k, v in raw_data.items()}

Parameters like name, tags, kinds, and group are optional. The asset name defaults to the function name.

Single assets support the deps parameter for declaring upstream dependencies with custom partition mappings, IO handler overrides, or metadata overrides — see Dependencies with deps below.

Multi assets¶

A single function that produces multiple named outputs:

@rs.Asset.from_multi(
    output_defs=[
        rs.AssetDef(name="users"),
        rs.AssetDef(name="orders"),
    ],
)
def extract():
    return {
        "users": load_users(),
        "orders": load_orders(),
    }

Each output becomes its own asset in the DAG. Downstream assets depend on individual outputs by name.

Subsetting¶

Multi-assets are implicitly subsettable — you can materialize a subset of the outputs without redefining the asset. rivers runs the function as usual, then only persists yields / dict entries whose output_name is in the requested selection:

@rs.Asset.from_multi(
    output_defs=[rs.AssetDef("users"), rs.AssetDef("orders")],
)
def extract():
    yield rs.Output(value=load_users(), output_name="users")
    yield rs.Output(value=load_orders(), output_name="orders")

repo = rs.CodeRepository(assets=[extract])
repo.materialize(selection=["users"])  # only "users" is persisted

For expensive computations, branch on context.output_selection to skip work for outputs that won't be persisted:

@rs.Asset.from_multi(
    output_defs=[rs.AssetDef("users"), rs.AssetDef("orders")],
)
def extract(context: rs.AssetExecutionContext):
    if "users" in context.output_selection:
        yield rs.Output(value=load_users(), output_name="users")
    if "orders" in context.output_selection:
        yield rs.Output(value=load_orders(), output_name="orders")

When no subsetting is requested (i.e. the whole multi-asset is materialized), context.output_selection contains all output names.

Partial failure¶

When a multi-asset raises after it yields some outputs, those outputs keep their materialization and stored value. Only the outputs it did not yield fail, with the error. An output that the function ends without yielding also fails. Every executor works this way.

Partitions on multi-assets¶

Apply partitions_def at the top level to partition all outputs uniformly:

pd = rs.PartitionsDefinition.static_(["us", "eu"])

@rs.Asset.from_multi(
    partitions_def=pd,
    output_defs=[rs.AssetDef("users"), rs.AssetDef("orders")],
)
def extract(context: rs.AssetExecutionContext):
    region = context.partition_key
    yield rs.Output(value=load_users(region), output_name="users")
    yield rs.Output(value=load_orders(region), output_name="orders")

Alternatively, set partitions_def on individual AssetDef entries when outputs have different partition spaces. Per-output definitions must share the same variant type and have overlapping keys.

Dependencies with deps¶

Use the deps parameter to declare upstream dependencies with custom partition mappings, IO handler overrides, or metadata overrides. There are two kinds:

• AssetDef.input(name) — a data dependency. The upstream's output is loaded and passed as a function parameter (matched by name).
• AssetDef.dep(name) — a lineage-only edge. Ensures ordering in the DAG without loading data. The name does not need to match a function parameter.

@rs.Asset.from_multi(
    partitions_def=pd,
    output_defs=[rs.AssetDef("report", io_handler=handler)],
    deps=[
        rs.AssetDef.input("source", partition_mapping=rs.PartitionMapping.static_({"us": "1"})),
        rs.AssetDef.dep("trigger"),
    ],
)
def report(source: int):
    yield rs.Output(value=source * 2, output_name="report")

deps can also be declared per-output on AssetDef. Input deps merge into the multi-asset's function-level input set (the function fires once for all outputs); lineage-only deps become edges to that specific output only:

@rs.Asset.from_multi(
    output_defs=[
        rs.AssetDef("users", deps=[rs.AssetDef.dep("users_source")]),
        rs.AssetDef("orders", deps=[rs.AssetDef.dep("orders_source")]),
    ],
)
def extract():
    yield rs.Output(value=..., output_name="users")
    yield rs.Output(value=..., output_name="orders")

See the API reference for the full AssetDef.input() and AssetDef.dep() signatures.

Graph assets¶

Compose Task operations into a sub-DAG treated as a single asset. Internal tasks are namespaced as {graph_name}/{task_name} and execute as independent steps:

@rs.Task
def fetch(url: str):
    return requests.get(url).json()

@rs.Task
def transform(data: dict):
    return [row for row in data if row["active"]]

@rs.Asset.from_graph()
def pipeline():
    data = fetch("https://api.example.com/users")
    return transform(data)

The return value determines the final node — its output becomes the graph asset's output. Use node_io_handler to give internal tasks a different IO handler than the graph output, and rivers/node/executor metadata to control which executor internal tasks use. See the Graph Assets guide for details.

External assets¶

An external asset represents a data source managed outside rivers — for example, a table produced by another pipeline, an S3 bucket populated by an external system, or a shared dataset. External assets participate in the dependency graph so downstream assets can depend on them, but rivers never materializes them.

# Declare an external data source
source_table = rs.Asset.external(
    name="source_table",
    io_handler=delta_handler,
    kinds="table",
    metadata={"path": "s3://bucket/source"},
)

# Downstream assets depend on it like any other asset
@rs.Asset
def enriched(source_table: pl.DataFrame) -> pl.DataFrame:
    return source_table.with_columns(pl.lit("enriched").alias("status"))

When enriched is materialized, rivers loads source_table via delta_handler.load_input() and passes the result as the function argument — but source_table itself is never executed.

Observations¶

External assets can have an optional observation function that inspects the external data source and records metadata. This is useful for tracking freshness, row counts, or schema changes without materializing anything.

Use Asset.external() as a decorator to attach an observation function:

@rs.Asset.external(io_handler=delta_handler)
def source_table(context: rs.AssetExecutionContext):
    # Query the external system for metadata
    context.add_output_metadata({
        "row_count": rs.MetadataValue.int(150_000),
        "last_modified": rs.MetadataValue.timestamp(1709913600.0),
    })

Trigger observations manually via CodeRepository.observe():

repo = rs.CodeRepository(assets=[source_table, enriched])

# Observe all external assets (or filter with asset_names=["source_table"])
result = repo.observe()
assert result.success

# The metadata lands on the run's Observation events.
events = repo.storage.get_events_for_run(result.run_id)

In production, attach an AutomationCondition.on_cron() to run observations on a schedule. When the observation records a new data_version, downstream assets with AutomationCondition.eager() will automatically materialize:

source_table = rs.Asset.external(
    name="source_table",
    io_handler=handler,
    automation_condition=rs.AutomationCondition.on_cron("0 */6 * * *"),
)

Key properties¶

• No compute function — external assets are never materialized by rivers
• io_handler is required — downstream assets need it to load data via load_input()
• Source-only nodes — external assets are leaf nodes in the DAG; they provide data to downstream assets but are never materialized themselves
• Excluded from materialization — materialize() automatically filters out external assets from the execution selection, but they remain in the dependency graph so downstream assets can load their data via io_handler.load_input()

Self-dependent assets¶

An asset can read its own previously persisted value to support incremental patterns (running counters, accumulating windows, append-only state). Declare a parameter named self typed as SelfDependency[T] — the executor loads the asset's last stored value through its IO handler before invoking the function:

@rs.Asset
def running_total(self: rs.SelfDependency[int], new_events: list) -> int:
    prev = self.get_inner()         # int | None
    if prev is None:
        return len(new_events)      # first run
    return prev + len(new_events)   # subsequent runs

How it works:

• get_inner() returns the previously stored value, or None on first run (when nothing has been persisted yet).
• No graph edge is created for self — there's no cycle. The dependency is satisfied through io_handler.load_input(), not the DAG.
• An IO handler is required. The asset's own io_handler is preferred; if unset, the default InMemoryIOHandler is used. Materialization raises ConfigurationError if neither is available.
• T is forwarded as the IO handler's type_hint (e.g. SelfDependency[pa.Table] makes DeltaIOHandler.load_input return a PyArrow Table).

This composes with normal deps: self: SelfDependency[T] only handles the self-load, leaving every other parameter free to bind to upstream assets / resources / config as usual.

Return type checks¶

rivers checks the value an asset returns against its return annotation. A mismatch fails the step with AssetOutputValidationError, and the value is not stored:

@rs.Asset
def row_count() -> int:
    return "12"  # Asset 'row_count' returned value of type 'str' but expected '<class 'int'>'

• A class is checked with isinstance. A generic such as list[int] is checked by its origin, list, not by its items.
• Any, None, and the Output, Observation and Materialization types are not checked. The value inside a returned Output is checked.
• A union such as int | None passes when the value matches one member.
• Annotated[T, ...] is checked against T.
• A pandera or dataframely schema in the annotation validates the frame itself. See pandera and dataframely.
• With from __future__ import annotations, rivers evaluates the annotation in the asset's module. A name that is imported only under TYPE_CHECKING fails the step.

A multi-asset has one return annotation for all its outputs. Give an output its own type with AssetDef(type_hint=...). rivers checks each value that the output gets, in a returned dict or a yielded Output, with the rules above:

@rs.Asset.from_multi(
    output_defs=[
        rs.AssetDef("users", type_hint=list[str]),
        rs.AssetDef("user_count", type_hint=int),
    ]
)
def load():
    users = ["ada", "grace"]
    yield rs.Output(users, output_name="users")
    yield rs.Output(len(users), output_name="user_count")

A value that fails the check fails its output with AssetOutputValidationError. The outputs that rivers stored before it keep their materialization, and the other outputs fail too (see Partial failure). A Materialization or Observation output has no value, so rivers does not check it.

Every executor runs the same check. The parallel executor runs it in the worker process.

Asset metadata¶

Attach static metadata to an asset via the metadata parameter. Metadata serves three purposes:

1. IO handler configuration — IO handlers receive metadata as context.asset_metadata and use prefixed keys to control behavior (e.g. write mode, partition columns)
2. Engine behavior — rivers itself reads rivers/-prefixed keys to change execution behavior
3. UI display — all metadata is visible in the rivers web UI for debugging and documentation

@rs.Asset(metadata={
    "delta/mode": "append",              # DeltaIOHandler: use append mode
    "delta/partition_expr": "date",       # DeltaIOHandler: partition by date column
    "rivers/executor": "in_process",      # Engine: force in-process execution
    "owner": "data-team",                 # UI: informational, displayed in asset detail
})
def events():
    ...

Reserved metadata prefixes¶

Custom IO handlers can define their own prefixed keys following the same convention.

Metadata is a dict[str, str] — all values are strings. IO handlers parse them as needed (e.g. delta/partition_expr can be a JSON dict for multi-dimensional partitions).

Per-run overrides¶

A launch can add or replace metadata keys for one run through the launch document, from Python or the UI's launch dialogs:

repo.materialize(
    selection=["events"],
    config={"assets": {"events": {"metadata": {"delta/mode": "overwrite"}}}},
)

The merged metadata reaches context.asset_metadata, the IO handlers and the engine for that run only; the definition is unchanged. rivers/executor set this way picks the executor for that asset's step.

Output metadata¶

Asset functions and IO handlers can attach runtime metadata via context.add_output_metadata(). This metadata is stored on the materialization event and can include MetadataValue instances for typed values:

context.add_output_metadata({
    "num_rows": 1000,
    "schema": rs.MetadataValue.json('{"id": "int64"}'),
})

The IO handler adds its entries in handle_output, after the asset function returns. On a key that both set, the handler's value is stored, because it describes the data that was written.

Using Output for declarative metadata¶

Instead of using the context imperatively, you can return an Output object from your asset function. This carries the value, metadata, data version, and tags together:

@rs.Asset(io_handler=handler)
def processed_data() -> rs.Output:
    df = compute_data()
    return rs.Output(
        value=df,
        metadata={
            "row_count": rs.MetadataValue.int(len(df)),
            "schema": rs.MetadataValue.json(df.schema.to_json()),
        },
        data_version="v2.0",
    )

When you return Output, the IO handler receives the unwrapped value (not the Output wrapper). You can also combine Output with context.add_output_metadata() — both are merged, with Output metadata taking precedence on key conflicts.

Using Observation for external assets¶

Similarly, observation functions on external assets can return an Observation object:

@rs.Asset.external(io_handler=handler)
def source_table(context: rs.AssetExecutionContext):
    last_mod = query_table_last_modified()
    return rs.Observation(
        metadata={"last_modified": rs.MetadataValue.timestamp(last_mod)},
        data_version=str(last_mod),
    )

Observation carries metadata and an optional data version, but no value (observations don't produce data). When returned, an Observation event is emitted to storage.

Using Materialization for self-managed persistence¶

For terminal side-effecting assets — pushing to an API, emitting a message, writing directly to an external system — return a Materialization instead of an Output:

@rs.Asset
def push_to_api(rows: list[dict]) -> rs.Materialization:
    response = requests.post(API_URL, json=rows)
    return rs.Materialization(
        metadata={"status_code": rs.MetadataValue.int(response.status_code)},
        data_version=response.headers["ETag"],
    )

Materialization is the explicit way to opt an asset out of the IO handler framework: the asset has already persisted its output, so the framework records a Materialization event with the supplied metadata and data version but never invokes handle_output. Downstream consumers cannot load_input the result — by design. See the IO handlers concept for details.