CodeRepository¶

CodeRepository¶

Central registry of assets, tasks, jobs, schedules, and sensors. Resolves the full dependency graph and validates each user-defined Job. Calls to materialize() build an ephemeral execution plan over the selection — there is no auto-generated job, and ad-hoc runs have RunRecord.job_name = None.

import rivers as rs

repo = rs.CodeRepository(
    assets=[asset_a, asset_b, asset_c],
    tasks=[my_task],
    jobs=[
        rs.Job(name="pipeline", assets=[asset_a, asset_b]),
    ],
    schedules=[nightly_schedule],
    sensors=[file_sensor],
)

Constructor:

Properties:

Methods¶

resolve()¶

def resolve(self, storage: Storage | None = None) -> None

Validate the graph and persist topology / metadata into storage. Must be called before materialize(), backfill(), or daemon start. Without an explicit storage, an embedded SurrealDB+RocksDB backend is created at .rivers/storage/.

validate()¶

def validate(self) -> None

Run the storage-independent validation pipeline: graph composition, partition / external / resource-reference checks, and per-job plan building. Raises the same errors resolve() would for graph problems, but does not initialize storage, invoke Resource.setup(), resolve IO handler ResourceRefs, register assets/pools, or persist topology.

Intended for CLI / IDE / UI tools that want fast feedback on whether a repository is well-formed without the side effects of a full resolve(). Always re-runs (no idempotency guard) so it can be called repeatedly while the user edits code.

get_job()¶

def get_job(self, name: str) -> Job

Retrieve a validated job by name. Raises NodeNotFoundError if no job with that name exists.

The returned job is this repository's own resolved copy: its plan, IO handlers, resources, executor, and retry policy come from this repository. The Job passed to CodeRepository(jobs=...) keeps only its declaration, so several repositories can share one Job. Run jobs through get_job().

materialize()¶

def materialize(
    self,
    selection: list[str] | None = None,
    partition_key: PartitionKey | None = None,
    tags: list[tuple[str, str]] | None = None,
    raise_on_error: bool = True,
    config: dict[str, Any] | None = None,
    run_id_override: str | None = None,
    include_upstream: bool = False,
    resume: bool = False,
    retry: RetryPolicy | str | None = None,
) -> RunResult

Materialize assets synchronously over an ephemeral execution plan. When selection is provided, only the selected assets are materialized; pass include_upstream=True to also materialize their transitive deps. The resulting RunRecord.job_name is None.

When a RunQueueConfig is configured, the run queue applies to daemon-submitted runs (sensors, schedules, automation conditions, UI). materialize() always executes directly regardless of queue configuration.

Returns: RunResult.

backfill()¶

def backfill(
    self,
    selection: list[str] | None = None,
    partition_keys: list[PartitionKey] | None = None,
    partition_range: PartitionKeyRange | None = None,
    strategy: BackfillStrategy | None = None,
    failure_policy: str = "continue",
    max_concurrency: int = 4,
    tags: list[tuple[str, str]] | None = None,
    config: dict[str, Any] | None = None,
    block: bool = True,
    dry_run: bool = False,
    action: str | None = None,
) -> BackfillResult

Backfill partitions for the selected assets. action runs that verb on every partition instead of materializing (every selected asset must define it). config is kept on the backfill record and applied to every child run, also when a daemon runs a block=False backfill later. See Backfills for the full reference.

cancel_backfill()¶

def cancel_backfill(self, backfill_id: str) -> bool

Cancel a running backfill. Returns True if the in-process coordinator was signalled (the backfill has live state in this process); returns False and falls back to a storage-level cancel marker otherwise.

get_backfill()¶

def get_backfill(self, backfill_id: str) -> BackfillStatus | None

Look up a backfill's current status by ID, or None if not found.

rerun_backfill()¶

def rerun_backfill(
    self, backfill_id: str, block: bool = True, dry_run: bool = False
) -> BackfillResult

Launch a previous backfill again. The rerun replays every partition of the original backfill, not only the failed or canceled ones; keys that are no longer valid are dropped. It keeps the original verb, so a rerun of a delete backfill deletes every partition again, including partitions materialized again after the first run. For a backfill of a job, it raises ExecutionError if the job now runs a different verb. To retry only some partitions, call backfill() with those keys and the original action. See Backfills for the full reference.

observe()¶

def observe(self, asset_names: list[str] | None = None) -> RunResult

Run observation functions on external assets. Only external assets with an observe_fn (set via the @Asset.external() decorator) are observed; names that don't match one are skipped. observe is the built-in action, so this executes as a run — the observation metadata lands on the run's Observation events.

Returns: RunResult. With nothing observable to run, a successful no-op with an empty run_id.

@rs.Asset.external(io_handler=handler)
def source(context: rs.AssetExecutionContext):
    context.add_output_metadata({"row_count": rs.MetadataValue.int(1000)})

repo = rs.CodeRepository(assets=[source])
result = repo.observe()
assert result.success

events = repo.storage.get_events_for_run(result.run_id)

run_action()¶

def run_action(
    self,
    action: str,
    selection: list[str] | None = None,
    partition_key: PartitionKey | None = None,
    tags: list[tuple[str, str]] | None = None,
    raise_on_error: bool = True,
    config: dict[str, Any] | None = None,
    run_id_override: str | None = None,
    resume: bool = False,
) -> RunResult

Run a named asset action over a selection. The plan has one step per target — upstream is never pulled in and the producing function never runs. The run record carries the verb, and each step emits ActionCompleted (or Deletion, for an Unmaterialize outcome) instead of Materialization.

result = repo.run_action("optimize", selection=["events"])
assert result.success
assert repo.storage.get_run(result.run_id).action == "optimize"

load_node()¶

def load_node(
    self,
    name: str,
    *,
    partition_key: PartitionKey | None = None,
    type_hint: type[T] | None = None,
) -> T | Any

Load a node's persisted output via its IO handler. Returns the value stored by the most recent materialization.

repo.materialize()
value = repo.load_node("my_asset")
df = repo.load_node("delta_asset", type_hint=pa.Table)  # typed as pa.Table

io_handler_for_output()¶

def io_handler_for_output(self, name: str) -> BaseIOHandler

Resolve the IO handler this repository would use to write name's output. Walks the registry chain node.io_handler() → default — the same chain the executor uses at materialize time. Returns the resolved handler instance (the configured one, an instance derived from a ResourceRef, or the shared default InMemoryIOHandler).

Useful for debugging "which handler does this asset actually use?" without running execution.

Raises: NodeNotFoundError if name is not in the resolved repository.

get_schedule() / get_sensor()¶

def get_schedule(self, name: str) -> Schedule
def get_sensor(self, name: str) -> Sensor

Look up a registered schedule or sensor by name.

evaluate_schedule()¶

def evaluate_schedule(
    self, name: str, execution_time: str | None = None
) -> ScheduleTickResult

Run a schedule's evaluation function once and return the tick result.

evaluate_sensor()¶

def evaluate_sensor(
    self, name: str, cursor: str | None = None, last_tick_time: float | None = None
) -> SensorTickResult

Run a sensor's evaluation function once and return the tick result.

shutdown()¶

def shutdown(self) -> None

Tear down resources, daemons, and gRPC/UI servers started by this repository. Calls teardown() on every registered resource.

Context manager¶

CodeRepository supports the context manager protocol; __exit__ calls shutdown():

with rs.CodeRepository(assets=[a, b]) as repo:
    repo.materialize()
# storage and resources are cleaned up automatically

RunResult¶

Outcome of a synchronous materialize() call.

To read output values, use repo.load_node(name).

RunHandle¶

Async handle to a submitted run. Returned by internal submission paths (daemon, gRPC/UI) when a RunQueueConfig is configured. Can be used to poll status, wait for completion, or cancel queued runs.

wait()¶

def wait(self, timeout: float | None = None) -> RunResult

Block until the run reaches a terminal state (or timeout seconds elapse). Raises on timeout.

cancel()¶

def cancel(self) -> None

Request cancellation of the run. No-op if already in a terminal state.

Ad-hoc materialization¶

repo.materialize() runs over all non-external assets without requiring a Job. Each call builds an ephemeral execution plan over the selection — the resulting RunRecord has job_name = None. Use a named Job when you want to surface the run target in the UI, share it across schedules, or pin executor/partition behavior.

repo = rs.CodeRepository(assets=[a, b, c])
result = repo.materialize()  # all non-external assets, ad-hoc
result = repo.materialize(selection=["b"], include_upstream=True)