Resources¶

Resources are shared, injectable dependencies (database connections, API clients, etc.) available across assets, tasks, schedules, and sensors.

Defining a resource¶

Extend rivers.Resource (a pydantic_settings.BaseSettings subclass) with optional lifecycle hooks:

import rivers as rs

class DatabaseResource(rs.Resource):
    connection_string: str  # explicit or from CONNECTION_STRING env var
    pool_size: int = 5

    def setup(self):
        self._pool = create_pool(self.connection_string, self.pool_size)

    def teardown(self):
        self._pool.close()

setup() and teardown() are optional — only override them if your resource needs initialization or cleanup.

Secrets¶

Declare credentials as pydantic.SecretStr (or SecretBytes). Reprs and logs show **********; your code reads the value with get_secret_value():

from pydantic import SecretStr

class WarehouseResource(rs.Resource):
    user: str
    password: SecretStr  # explicit or from the PASSWORD env var

    def setup(self):
        self._conn = connect(self.user, self.password.get_secret_value())

A secret keeps its value in other processes: a parallel worker and a sensor or schedule evaluated in a subprocess rebuild the resource from its field values, secrets included. The UI never gets a secret: in the launch dialog, a field that holds one at any depth (SecretStr | None, a nested model's field, a list's items) has no current value, and a launch that leaves it out keeps the resource's value.

Registering resources¶

Pass resources to CodeRepository as a dict:

repo = rs.CodeRepository(
    assets=[my_asset],
    resources={
        "db": DatabaseResource(connection_string="postgresql://..."),
        "api": APIClient(base_url="https://api.example.com"),
    },
)

Injection¶

Resources are injected by matching parameter names to resource keys:

@rs.Asset
def my_asset(context: rs.AssetExecutionContext, db: DatabaseResource, api: APIClient):
    data = api.fetch("/data")
    db._pool.execute("INSERT ...", data)
    return data

The injection resolution order is:

1. Context (first param, if it matches a context type)
2. self — self-dependency
3. Upstream asset name match (in-memory result)
4. Upstream asset name match (via io_handler)
5. Resource key match
6. Error — unknown parameter

Asset names take precedence over resource keys. rivers warns at resolve time if a resource key shadows an asset name.

Schedule and sensor injection¶

Schedules and sensors also support resource injection:

@rs.Schedule(cron_schedule="0 * * * *", job_name="hourly_job")
def my_schedule(context: rs.ScheduleEvaluationContext, db: DatabaseResource):
    if db.check_condition():
        return rs.RunRequest()
    return rs.SkipReason("No new data")

Lifecycle¶

CodeRepository.__init__()
  └─ stores raw resource instances

CodeRepository.resolve()
  └─ calls resource.setup() for each resource

materialize() / evaluate_schedule() / evaluate_sensor()
  └─ injects resource instances into function calls

CodeRepository shutdown (context manager __exit__ or shutdown())
  └─ calls resource.teardown() for each resource

Resources are initialized once at resolve time and shared across all executions. They are not re-created per run, unless a launch overrides one (below).

Per-run overrides¶

A launch can rebuild a resource with different field values for one run through the launch document, from Python or the UI's launch dialogs:

repo.materialize(
    selection=["my_asset"],
    config={"resources": {"db": {"connection_string": "postgres://replica/app"}}},
)

The named resource is built again through its class, from the instance's current fields with these values on top (so validation runs, and a secret keeps its value). A nested model takes the values field by field: {"login": {"user": "etl"}} keeps login.password. A list or dict field is replaced whole. The run record stores the document as written, so set a secret on the resource or in its environment, not in the document. A Resource built this way gets setup() before the run's first step and teardown() after its last; the repository's own instance is untouched and keeps serving other runs. A class given as a resource is built once for the run with the values. An IO handler cannot be overridden this way: assets hold it through the registry. An unknown key, or a value the class rejects, is refused before a run exists.

IOHandler as resource reference¶

io_handler on an asset can reference a resource by string key:

class S3Handler(rs.BaseIOHandler):
    bucket: str
    prefix: str

    def handle_output(self, context, obj): ...
    def load_input(self, context): ...

@rs.Asset(io_handler="s3")  # references resources["s3"]
def my_asset():
    return data

repo = rs.CodeRepository(
    assets=[my_asset],
    resources={"s3": S3Handler(bucket="my-bucket", prefix="assets/")},
)

String references are resolved to the actual handler at resolve() time, against that repository's resources. Two repositories built from the same assets each use their own handler.

Testing¶

Override resources in tests by passing a different dict:

def test_my_pipeline():
    mock_db = DatabaseResource(connection_string="sqlite:///:memory:")
    repo = rs.CodeRepository(
        assets=[my_asset],
        resources={"db": mock_db},
    )
    result = repo.materialize()
    assert result.success

Validation¶

rivers validates all resource references at resolve time:

• Asset/task function parameters that don't match an upstream asset or resource key raise ConfigurationError
• Schedule/sensor evaluation function parameters that don't match a resource key raise ConfigurationError
• IOHandler string references that don't match a resource key, or that name a resource which does not implement the IOHandler protocol, raise AssetDefinitionError. It is not a subclass of ConfigurationError, so catch it separately