Docs/Getting started

Tutorial: first pipeline

You will build a pipeline that tracks public holidays from an open JSON API. It moves data through the raw, staged and published layers, validates it with a check, runs on a schedule, and notifies a webhook when a record changes. It takes about 30 minutes.

Prerequisites

Finish the Quickstart first, so that syn is installed and you are logged in.

1. Create the project

terminal

$ syn init holidays
$ cd holidays
$ rm pipelines/example.py

2. Declare the source

The API is https://date.nager.example/api/v3 and allows 30 requests per minute. We describe it once as a Source. The rate limit is enforced for every pipeline that uses the source.

pipelines/sources.py

import synchronoid as sy

holidays_api = sy.Source(
    name="holidays_api",
    connector="http_json",
    base_url="https://date.nager.example/api/v3",
    rate_limit=sy.RateLimit(per_minute=30),
    retry=sy.Retry(max_attempts=5, backoff="exponential"),
)

3. Fetch into the raw layer

A raw asset stores the response exactly as it was received. Nothing is cleaned, so you can always replay later.

pipelines/holidays.py

from .sources import holidays_api
import synchronoid as sy

@sy.asset(layer="raw", source=holidays_api, partitions=sy.Partitions.years(2025, 2027))
def holidays_raw(ctx):
    return ctx.source.get(f"/PublicHolidays/{ctx.partition}/DE")

Each year is a partition. A partition can be run, replayed or backfilled on its own.

4. Normalize into the staged layer

pipelines/holidays.py

@sy.asset(layer="staged", deps=[holidays_raw])
def holidays_staged(ctx, holidays_raw):
    for item in holidays_raw.records():
        yield {
            "holiday_id": f"{item['countryCode']}-{item['date']}",
            "date": item["date"],
            "name": item["localName"].strip(),
            "is_global": item["global"],
        }

Records written here carry provenance automatically. You do not need to add anything.

5. Publish a current table

pipelines/holidays.py

@sy.asset(layer="published", deps=[holidays_staged])
def holidays_current(ctx, holidays_staged):
    return holidays_staged.merge(key="holiday_id", track_changes=True)

track_changes=True keeps a history of every version of each record. This is what powers the change feed in step 8.

6. Run it locally

Start syn dev, open http://localhost:4040, select holidays_current and click Materialize all. Synchronoid runs the three assets in order. Open the Lineage tab to see the chain.

localhost:4040/lineage

Assets

Lineage

Runs

holidays

holidays_raw3 partitionsSucceeded

holidays_staged71 recordsSucceeded

holidays_current71 recordsRunning

7. Add a check

A check is a function that inspects an asset after it runs. If it fails, the run is marked as failed and downstream assets do not update.

pipelines/checks.py

import synchronoid as sy
from .holidays import holidays_staged

@sy.check(asset=holidays_staged, severity="error")
def holidays_have_names(ctx, data):
    missing = data.count(where="name IS NULL OR name = ''")
    return sy.CheckResult(passed=missing == 0, detail=f"{missing} records without a name")

Use warnings first

When you add a check to existing data, start with severity="warn". Promote it to "error" when you are happy with the results.

8. Add a schedule

pipelines/schedules.py

import synchronoid as sy
from .holidays import holidays_current

weekly = sy.Schedule(cron="0 3 * * 1", target=holidays_current, timezone="Europe/Berlin")

The schedule runs the full chain every Monday at 03:00. See schedules and sensors for more options.

9. Deploy and subscribe to changes

terminal

$ syn deploy
$ syn feeds create holidays-changes \
    --asset holidays_current \
    --webhook https://example.com/hooks/holidays
Created change feed holidays-changes
Signing secret: whsec_••••••••

Whenever a record in holidays_current changes, the feed delivers an event like this:

webhook payload

{
  "id": "evt_01JA2B7XQ4",
  "type": "record.changed",
  "asset": "holidays_current",
  "key": { "holiday_id": "DE-2026-12-25" },
  "before": { "name": "Erster Weihnachtstag" },
  "after": { "name": "1. Weihnachtstag" },
  "provenance": { "source": "holidays_api", "run_id": "run_8f2c1a", "fetched_at": "2026-10-26T03:00:12Z" },
  "occurred_at": "2026-10-26T03:00:40Z"
}

The run_id in provenance links the event back to the exact run that produced it. The change feeds guide covers Kafka delivery and the pull API.

What you built

A source with a shared rate limit and retry policy.
Three assets across the storage layers.
A check that stops bad data from reaching published.
A weekly schedule and a webhook change feed.

Continue with Assets to understand the model in more depth.