Docs/Reference

Python SDK reference

The synchronoid package contains everything you need to define assets, steps, pipelines, sources, schedules and sensors. It requires Python 3.10 or newer.

terminal

$ pip install synchronoid

By convention the package is imported as sy:

pipelines/example.py

import synchronoid as sy

Decorators

@sy.asset

Declares a table-like dataset in one layer, and the function that produces it. The function receives the context and one argument per dependency. It returns or yields records.

Parameter	Type	Description
`layer`	`"raw" \| "staged" \| "published"`	Storage layer. Required.
`source`	`sy.Source`	Source this asset reads from.
`deps`	`list[Asset]`	Upstream assets.
`partitions`	`sy.Partitions`	Partition definition.
`retry`	`sy.Retry`	Retry policy for the asset.
`name`	`str`	Defaults to the function name.
`description`	`str`	Shown in the control room.
`checks`	`list[sy.Check]`	Data quality checks that run after the asset. Built-in checks: `sy.checks.schema`, `sy.checks.unique`, `sy.checks.fresh`.
`max_concurrency`	`int`	Limit for parallel partition runs.
`retention`	`sy.Retention`	How long data is kept. Defaults depend on the layer, see Storage layers.
`schema`	class	Typed schema that records must match, for example a dataclass.
`provenance`	`bool`	Record provenance per value. Default `True`.

@sy.step

Declares one unit of work inside a pipeline. A step is the smallest part that is retried and resumed.

Parameter	Type	Description
`retry`	`sy.Retry`	Overrides the pipeline policy.
`source`	`sy.Source`	Enables `ctx.source` and rate limiting.
`timeout`	`str`	For example `"15m"`. Default `"1h"`.

@sy.pipeline

Groups steps into an ordered graph.

pipelines/listings.py

@sy.pipeline(schedule=sy.Schedule(cron="0 * * * *"))
def listings_sync():
    ref = fetch_listings()
    clean = normalize(ref)
    publish(clean)

Parameters: schedule (sy.Schedule), retry (sy.Retry), name (str), tags (dict[str, str]).

@sy.sensor

Polls for an outside event and starts a run when it finds one. The function returns a list of run requests, or nothing.

pipelines/sensors.py

@sy.sensor(target=companies_current, every="5m")
def new_export_file(ctx):
    latest = ctx.source.head("/exports/latest.json")
    if latest.etag != ctx.cursor:
        ctx.cursor = latest.etag
        return sy.RunRequest(run_key=latest.etag)
    return sy.SkipRun(reason="no new file")

Classes

Class	Signature	Description
`sy.Source`	`Source(name, connector, base_url=None, auth=None, rate_limit=None, **options)`	A data provider and how to reach it. See Sources and connectors.
`sy.RateLimit`	`RateLimit(per_second=None, per_minute=None, per_hour=None, burst=None, reserve=0.0, max_wait="10m", concurrency=None, adaptive=True, retry_after_header="Retry-After")`	Shared request budget for a source.
`sy.Retry`	`Retry(max_attempts=3, backoff="exponential", initial_delay="1s", max_delay="10m", jitter=True, retry_on=None)`	Retry policy.
`sy.Schedule`	`Schedule(cron, target=None, timezone="UTC", on_missed="skip", max_concurrent_runs=1, partition=None, enabled=True)`	Time-based trigger. See Schedules and sensors.
`sy.Partitions`	`Partitions.daily(start, end=None)`, `Partitions.years(a, b)`, `Partitions.static(keys)`	Partition definitions.
`sy.Retention`	`Retention(days)`	How long an asset's data is kept.
`sy.Secret`	`Secret(name)`	Reference to a secret, for use in `sy.Source` and `sy.Destination` options.
`sy.Destination`	`Destination(connector, name=None, **options)`	A warehouse that receives published assets. Call `.publish(asset, table=...)`. See Snowflake.
`sy.dbt_project`	`dbt_project(project_dir, profiles, select=None, command="build", after=None)`	Runs dbt models after assets. See dbt.
`sy.CheckResult`	`CheckResult(passed, detail="")`	Returned by functions decorated with `@sy.check(asset=..., severity="error"\|"warn")`.
`sy.RunRequest`, `sy.SkipRun`	`RunRequest(run_key, partition=None)`, `SkipRun(reason)`	Returned by sensors. The run key prevents duplicate runs.
`sy.OnChange`, `sy.OnCreate`, `sy.OnDelete`	`OnChange(*fields)`	Event rules for `merge(events=...)`. See Change feeds.

The context object

The first argument of every asset, step and sensor is ctx.

Member	Type	Description
`ctx.log`	`Logger`	Structured logger: `debug`, `info`, `warning`, `error`. Lines appear in the run page.
`ctx.provenance`	`Provenance`	Source, fetch time and run id. Use `ctx.provenance.note(field=..., reason=...)` to add a note to a produced value.
`ctx.run_id`	`str`	Id of the current run.
`ctx.run_time`	`datetime`	Logical time of the run. Use it instead of the system clock.
`ctx.partition`	`str \| None`	Current partition key.
`ctx.attempt`	`int`	Attempt number, starting at 1.
`ctx.idempotency_key`	`str`	Stable per step and partition. Send it with outside writes.
`ctx.source`	`SourceClient`	Rate-limited client: `get`, `head`, `acquire(cost)`.
`ctx.raw`	`RawStore`	Raw layer client: `put(payload)` or `put(key, payload)` returns a reference, `exists(key)`, `ref(key)`.
`ctx.secret(name)`	`str`	Reads a secret for the current environment.
`ctx.cursor`	`str \| None`	Persistent value for sensors.
`ctx.add_metadata(dict)`	`None`	Attaches metadata, such as a row count, to the current materialization. Shown in the control room.

Provenance is automaticYou rarely call ctx.provenance yourself. Every value you yield gets its source, fetch time and run id. See Provenance.

Errors and client

Raise sy.errors.Timeout or sy.errors.HttpStatus(429) to trigger the matching retry_on rule. Raise sy.errors.CheckFailed("...") inside an asset to stop it before its output is published. To read published data from other Python code, use sy.Client().assets.get("companies_current").record(key, with_provenance=True). Provenance contains source, fetched_at, run_id, step, payload_ref, pointer and observed_since.

Custom connectors are declared with @sy.connector(name=...) and extend Connector from synchronoid.connectors.

Layer data objects

Method	Description
`raw.records(**filters)`	Iterates over decoded records from a raw payload.
`raw.payloads()`	Iterates over untouched payload bytes.
`data.merge(key, track_changes=False, events=None)`	Merges records into the published layer by key.
`data.filter(fn)`, `data.map(fn)`	Lazy transformations. Nothing loads until the step ends.
`data.count()`	Number of records, computed in the storage layer.

Testing

Use sy.testing.run_asset(asset, inputs=...) in unit tests. It runs the function with a fake context and returns the yielded records, so no network or account is needed.

tests/test_companies.py

from synchronoid.testing import run_asset

def test_names_are_trimmed():
    out = run_asset(company_profiles, raw=[{"id": "1", "name": " Acme ", "address": "x"}])
    assert out[0]["name"] == "Acme"