Docs/Concepts

Schedules and sensors

Schedules start runs at fixed times. Sensors start runs when something happens. Both are lightweight definitions that point at the assets or pipelines they should run.

Schedules

A Schedule uses a standard five-field cron expression. It names a target, which is an asset, a list of assets or a pipeline.

pipelines/schedules.py

import synchronoid as sy
from .companies import companies_current

hourly = sy.Schedule(cron="0 * * * *", target=companies_current)

nightly = sy.Schedule(
    cron="0 2 * * *",
    target=companies_current,
    timezone="Europe/Berlin",
    on_missed="run_once",
    max_concurrent_runs=1,
)

Option	Default	Meaning
`cron`	required	Five-field cron expression.
`target`	required	Asset, list of assets or pipeline to run.
`timezone`	`"UTC"`	IANA timezone name. Daylight saving changes are handled.
`on_missed`	`"skip"`	What to do if the platform was unavailable at the trigger time: `skip`, `run_once` or `run_all`.
`max_concurrent_runs`	`1`	A new run is skipped while this many runs are active.
`enabled`	`True`	Set to `False` to define a schedule without activating it.

Partitions and schedules

When the target is a partitioned asset, the schedule runs the newest partition by default. Use partition="previous" to run the one before, for example to process yesterday's data at 02:00.

pipelines/schedules.py

daily_meters = sy.Schedule(cron="0 2 * * *", target=meter_readings, partition="previous")

Sensors

A sensor is a function that Synchronoid calls at a regular interval. It decides whether to start a run. Use a sensor when the right time to run is not known in advance, for example when a file arrives or when a provider publishes a new version.

pipelines/sensors.py

import synchronoid as sy

@sy.sensor(target=company_profiles, every="5m")
def registry_updated(ctx):
    latest = ctx.source("company_registry").get("/version").json()["version"]
    if latest == ctx.cursor:
        return sy.SkipRun(reason="no new version")
    ctx.cursor = latest
    return sy.RunRequest(run_key=latest)

ctx.cursor is a small value that Synchronoid stores for you between calls. The run_key makes the request idempotent: two requests with the same key start only one run, even if the sensor is evaluated twice.

Sensors that use the rate limit

Sensor requests to a source count against the source's rate limit, like all other requests. Pick an every value that leaves room for your assets. A sensor that polls every minute against a 60 requests per minute budget would be a poor choice.

Keep sensors cheap

A sensor must finish within 30 seconds. It should only look at a version, a timestamp or a file listing. Do the real work in an asset.

Managing triggers

You can see all schedules and sensors in the control room, with the next planned run and the last result. You can also turn them off from the CLI.

terminal

$ syn schedules list
$ syn schedules disable nightly
$ syn sensors test registry_updated   # evaluates once, starts no run

Schedules, sensors and backfills

Schedules and sensors only deal with the present. To fill in past partitions, use a backfill. To react to changes in the output instead, use a change feed.