Execution Contexts¶

AssetExecutionContext¶

Execution context injected into asset functions. Provides access to asset properties, partitioning info, logging, and output metadata.

Usage¶

Add context: AssetExecutionContext as the first parameter of your asset function:

from rivers import Asset, AssetExecutionContext

@Asset(tags=["analytics"], kinds="table", group="reports")
def my_asset(context: AssetExecutionContext):
    context.log.info(f"Running {context.asset_name}")
    context.add_output_metadata({"rows": 1000})
    return compute_data()

@Asset
def downstream(context: AssetExecutionContext, my_asset: list):
    context.log.info(f"Processing {len(my_asset)} items")
    return transform(my_asset)

Context is optional — assets without it continue to work as before.

Properties¶

Methods¶

add_output_metadata(metadata)¶

Add metadata that will flow to the IO handler's OutputContext:

@Asset(io_handler=MyHandler())
def my_asset(context: AssetExecutionContext):
    result = compute()
    context.add_output_metadata({
        "rows": len(result),
        "status": "success",
    })
    return result

Values are auto-coerced to MetadataValue (str, int, float, bool, None supported); pass an explicit MetadataValue for typed variants.

The entries are stored on the materialization event. If the IO handler adds an entry with the same key in handle_output, the handler's value is stored.

register_data_version(version)¶

Register a custom data version for this materialization, overriding the auto-generated UUID:

@Asset
def my_asset(context: AssetExecutionContext):
    data = fetch_data()
    context.register_data_version(compute_hash(data))
    return data

mark_partition_failed(partition_key, error)¶

For multi-partition steps (single-run backfills, multi-key materializations), record that one specific partition inside the step failed without aborting the rest:

for key in keys:
    try:
        process(key)
    except Exception as exc:
        context.mark_partition_failed(rs.PartitionKey.single(key), str(exc))

Property: output_metadata¶

Returns accumulated metadata as dict[str, MetadataValue], or None if empty.

TaskExecutionContext¶

A lightweight execution context for tasks. Contains only task-relevant fields — no asset-specific fields like kinds, group, code_version, asset_metadata, or output metadata.

Usage¶

Add context: TaskExecutionContext as the first parameter of your task function:

from rivers import Task, TaskExecutionContext

@Task(tags=["etl"])
def my_task(context: TaskExecutionContext, source: int) -> str:
    context.log.info(f"Running task {context.task_name}")
    return f"processed: {source}"

Properties¶

parts = rs.PartitionsDefinition.static_(["a", "b"])

@rs.Task(partitions_def=parts)
def partitioned_task(context: rs.TaskExecutionContext) -> str:
    return f"key={context.partition_key}"

ActionContext¶

The context an asset action receives. An action acts on data that already exists, so this context carries no upstream inputs and no output metadata — it identifies the target and hands over the asset's resolved IO handler.

Usage¶

The first parameter is always the context (no annotation required); parameters after it are resources injected by name. Annotate it as ActionContext[Config] to receive typed config.

To unit-test an action body, build a context by hand and call the body with it: rs.ActionContext(asset_name="events", action="purge", partition=..., io_handler=...) (run_id defaults to ""; everything else to None).

@rs.action(outcome=rs.Outcome.Unchanged)
@classmethod
def compact(cls, ctx: rs.ActionContext, warehouse: DuckDB) -> None:
    warehouse.execute(f"VACUUM {ctx.asset_name}")

Properties¶

Methods¶

mark_partition_failed(partition_key, error)¶

Mirrors AssetExecutionContext.mark_partition_failed for batched action runs: record that one key of the batch failed while the rest count as succeeded, so a compact over a range can skip a corrupt partition without failing — or silently completing — the whole batch.

@rs.action(outcome=rs.Outcome.Unmaterialize)
@classmethod
def delete(cls, ctx: rs.ActionContext) -> None:
    for key in ctx.partition.keys:
        try:
            drop_partition(key)
        except Exception as exc:
            ctx.mark_partition_failed(key, str(exc))

The key must be in ctx.partition.keys. Marked keys are peeled out of the Deletion / ActionCompleted event the run emits and roll up into the backfill's failed counters — but unlike a materialization failure they do not floor the partition for automation, since a failed action is not a failed materialization attempt.

Detection rules¶

Both context types follow the same injection rules:

1. If the first parameter's type annotation is AssetExecutionContext or TaskExecutionContext, the appropriate context is injected
2. If the first parameter is named context and no asset/task with that name exists, AssetExecutionContext is injected
3. Context as a non-first parameter raises ExecutionError ("Context must be the first parameter of '<step>'")

Tasks can use either TaskExecutionContext (recommended) or AssetExecutionContext (backward compatible).

Actions differ: their first parameter is always the ActionContext, annotated or not (a zero-parameter action is allowed). Naming that parameter after a resource is an error rather than a silent context binding.

Executor support¶