Airbyte¶

rivers.integrations.airbyte runs Airbyte syncs as rivers assets. AirbyteResource calls the Airbyte API. Its connection_asset method makes a multi-asset for an Airbyte connection, with one output per enabled stream. A run starts a sync job, waits for it, and records a materialization for each stream.

Setup¶

The integration needs no extra: it calls the Airbyte API with the Python standard library.

Create an application in Airbyte (Settings → Applications) to get a client id and a client secret. AirbyteResource reads them from AIRBYTE_CLIENT_ID and AIRBYTE_CLIENT_SECRET, and connects to Airbyte Cloud:

from rivers.integrations.airbyte import AirbyteResource

airbyte = AirbyteResource()  # Airbyte Cloud, credentials from AIRBYTE_CLIENT_ID and AIRBYTE_CLIENT_SECRET

For a self-managed Airbyte, set the URLs of its two APIs:

airbyte = AirbyteResource(
    api_url="http://localhost:8000/api/public/v1",
    configuration_api_url="http://localhost:8000/api/v1",
)

With basic authentication, set username and password instead of the client credentials. Without authentication, set neither.

Fields that you do not give are read from AIRBYTE_* environment variables, for example AIRBYTE_API_URL.

Assets for a connection¶

import rivers as rs

postgres = airbyte.get_connection("9924bcd0-99be-453d-ba47-c2c9766f7da5")

repo = rs.CodeRepository(
    assets=[airbyte.connection_asset(postgres)],
    resources={"airbyte": airbyte},
)

get_connection reads the connection and its enabled streams from the Airbyte API. connection_asset makes a multi-asset named airbyte_ and the connection name in snake case, for example airbyte_postgres_to_snowflake. Each output has the name of its destination table: the connection's table prefix and the stream name, for example pg_users. The outputs have the kind airbyte.

get_connection runs each time a process loads the code location, for example each Kubernetes step pod. To load the code location without the Airbyte API, for example in tests, save the connection as JSON and read it back:

from pathlib import Path
from rivers.integrations.airbyte import AirbyteConnection

Path("airbyte/postgres.json").write_text(postgres.model_dump_json())  # once

postgres = AirbyteConnection.model_validate_json(Path("airbyte/postgres.json").read_text())

When the connection's streams change in Airbyte, load the code location again to update the outputs.

The registered resource¶

A run uses the resource registered under the key airbyte, not the instance that made the asset. So a run can override the resource's fields, from Python or from the launch dialog in the UI:

repo.materialize(config={"resources": {"airbyte": {"poll_timeout": 7200}}})

Without the resource, the code location fails to load with unresolved input 'airbyte'.

To register a resource under another key, for example for a second Airbyte, give the key as resource_key:

local = AirbyteResource(
    api_url="http://localhost:8000/api/public/v1",
    configuration_api_url="http://localhost:8000/api/v1",
)
mysql = local.get_connection("0c31738c-0b2d-4887-b506-e2cd1c39cc35")

repo = rs.CodeRepository(
    assets=[
        airbyte.connection_asset(postgres),
        local.connection_asset(mysql, resource_key="airbyte_local"),
    ],
    resources={"airbyte": airbyte, "airbyte_local": local},
)

What a run does¶

1. Reads the connection again, to find streams that were removed or added after the code location loaded.
2. Starts one sync job for the whole connection.
3. Checks the job every poll_interval seconds until it finishes.
4. Records a materialization for each selected output.

The asset's function is async, so its step waits for the job without holding a worker of the executor, and the Airbyte assets of one level wait at the same time: all of them with the parallel executor (max_async_concurrent limits them), and up to four with the in-process executor. A waiting step keeps its pool slots until the job finishes.

Airbyte syncs all the streams of a connection. A run that selects some outputs syncs all the streams too, and records materializations only for the selected outputs.

When the job fails or is cancelled in Airbyte, all the outputs fail with AirbyteError. When a selected output's stream is no longer enabled, that output fails and the others succeed. A stream that was enabled after the code location loaded is synced, and the run log names it.

When a sync of the connection is already running, Airbyte does not start another, and the run fails with AirbyteError (HTTP 409).

Each materialization has this metadata:

The counts come from the Configuration API, because the Airbyte API reports only totals for the whole job. When Airbyte has no counts for a stream, its materialization has no count keys. When the Configuration API request fails, no materialization has counts, and the run log has a warning.

Timeouts and cancel¶

Set poll_timeout to stop waiting for long jobs:

airbyte = AirbyteResource(poll_timeout=3600)

When rivers stops waiting before the job finishes, it cancels the job in Airbyte. This occurs after poll_timeout, when an API request fails, and on an interrupt such as Ctrl-C. Set cancel_on_interrupt=False to let the job continue in Airbyte. A process that is killed, for example a deleted Kubernetes pod, cannot cancel the job.

To try a failed sync again, give the asset a retry policy:

from rivers.integrations.airbyte import AirbyteError

airbyte.connection_asset(postgres, retry=rs.RetryPolicy(max_retries=2, retry_on=[AirbyteError]))

Change the outputs¶

asset_def_fn makes the rs.AssetDef of each stream. It takes the AirbyteConnection and one of its AirbyteStreams. connection.output_def(stream, ...) returns the default output with the fields that you give. The other keyword arguments of connection_asset go to rs.Asset.from_multi, for example group, hooks, automation_condition, or retry:

postgres_sync = airbyte.connection_asset(
    postgres,
    name="postgres_sync",
    asset_def_fn=lambda connection, stream: connection.output_def(
        stream, kinds=["airbyte", "snowflake"]
    ),
    group="raw",
    automation_condition=rs.AutomationCondition.on_cron("0 2 * * *"),
)

The default output sets only the name, so it gets group and the kind airbyte from connection_asset. A field that asset_def_fn sets replaces them: above, the outputs have the kinds airbyte and snowflake.

Two streams with the same name in different namespaces need an asset_def_fn that gives them different names, for example with stream.namespace.

Downstream assets¶

Airbyte writes the tables, and each run records a Materialization, so rivers stores no values for these assets. Declare the dependency with rs.AssetDef.dep, and read the table from the destination:

@rs.Asset(deps=[rs.AssetDef.dep("pg_users")])
def active_users() -> None:
    ...  # query pg_users in the destination

Start a sync in your own code¶

sync starts a sync job, waits for it, and returns the AirbyteJob. It raises AirbyteError when the job does not succeed. Take the registered resource as a parameter, so that a run's config reaches it too:

@rs.Asset
def postgres_rows(context: rs.AssetExecutionContext, airbyte: AirbyteResource) -> int:
    job = airbyte.sync("9924bcd0-99be-453d-ba47-c2c9766f7da5", log=context.log)
    return job.rows_synced  # records of all streams

In an async def asset, use async_sync. It takes the same arguments and waits without blocking the event loop.

See the API reference for all the classes.