Skip to content

Scheduled Triggers

This tutorial shows how to set up a DAG that runs automatically on a schedule using lythonic's trigger system.

Define the DAG

Start with a simple DAG that fetches and processes data:

from lythonic.compose.namespace import Dag, Namespace

async def fetch_prices() -> dict[str, float]:
    """Simulate fetching latest prices."""
    return {"AAPL": 185.0, "GOOG": 140.0}

async def summarize(prices: dict[str, float]) -> str:
    total = sum(prices.values())
    return f"Portfolio value: ${total:.2f}"

ns = Namespace()
ns.register(fetch_prices, nsref="market:fetch")
ns.register(summarize, nsref="market:summarize")

dag = Dag()
dag.node(ns.get("market:fetch")) >> dag.node(ns.get("market:summarize"))

ns.register(dag, nsref="pipelines:market_summary")

The DAG must be registered in the namespace so the trigger system can look it up by nsref.

Register a Trigger Definition

A trigger definition is declarative metadata — it describes what should happen but doesn't start anything:

ns.register_trigger(
    name="hourly_market",
    dag_nsref="pipelines:market_summary",
    trigger_type="poll",
    schedule="0 * * * *",
)

The schedule is a standard cron expression. Common patterns:

  • "0 * * * *" — every hour
  • "*/5 * * * *" — every 5 minutes
  • "0 9 * * MON-FRI" — weekdays at 9am
  • "0 0 * * *" — daily at midnight
  • "0 0 * * 0" — weekly on Sunday

Activate and Run

Activating a trigger creates a persistent record in the trigger database. The TriggerManager coordinates everything:

from pathlib import Path
from lythonic.compose.dag_provenance import DagProvenance
from lythonic.compose.trigger import TriggerManager, TriggerStore

store = TriggerStore(Path("triggers.db"))
provenance = DagProvenance(Path("provenance.db"))

manager = TriggerManager(
    namespace=ns,
    store=store,
    provenance=provenance,
)

manager.activate("hourly_market")

At this point the trigger definition is stored in the database with status active, but nothing is running yet.

Start the Poll Loop

Call start() to begin the background polling loop. It checks all active poll triggers every second and fires them when their interval elapses:

import asyncio

async def main():
    manager.start()

    # Your application runs here...
    # The trigger fires the DAG every hour in the background.
    await asyncio.sleep(3600)

    manager.stop()

asyncio.run(main())

Each time the trigger fires, the trigger system:

  1. Runs the DAG via the namespace node's async wrapper
  2. Records the event in trigger_events (trigger name, payload, run ID)
  3. Updates last_run_at and last_run_id in the activation record
  4. The DAG run itself is recorded in the provenance database

Check Trigger History

Query the trigger store for past events:

events = store.get_events("hourly_market")
for event in events:
    print(f"Fired at: {event['fired_at']}, status: {event['status']}, run: {event['run_id']}")

Deactivate

To stop a trigger from firing without stopping the manager:

manager.deactivate("hourly_market")

The trigger stays in the database with status disabled. Reactivate it later with manager.activate("hourly_market").

Restarts and Missed Fires

Call activate_configured() on every process start. It activates every trigger the node configs define, except those stored as disabled, so a deliberate deactivate() stays in force across restarts:

result = manager.activate_configured()
print(result.activated, result.left_disabled)

Re-activating a trigger that is already active keeps its last_run_at, even if its schedule changed, so if one or more scheduled fires fell due while the process was down, the trigger fires once (not once per missed fire) as soon as the poll loop starts.

An explicit activate() re-enables a disabled trigger and resets its last_run_at to now, so the time it spent paused is not caught up.

A run that takes longer than the gap to the next scheduled fire skips the fire times it overran and logs a warning naming the trigger.

Who Started a Run

Every run records initiated_by. Runs the poll loop fires record "scheduler" (SCHEDULER). For runs you start yourself, set the initiator with initiated_by, e.g. to the user id of the request being served:

from lythonic.compose.dag_provenance import initiated_by

with initiated_by("usr_1234"):
    result = await manager.fire("hourly_market")
    manager.deactivate("hourly_market")

provenance.get_run(result.run_id)["initiated_by"]  # "usr_1234"
store.get_activation("hourly_market")["changed_by"]  # "usr_1234"

activate() and deactivate() record the initiator as changed_by, with changed_at. activate_configured() at start-up runs with no initiator, so it leaves the last deliberate change in place.

Without Provenance

If you don't need DAG run tracking, skip the provenance parameter. The manager uses NullProvenance by default:

manager = TriggerManager(namespace=ns, store=store)

Custom Poll Functions

For triggers that check an external source instead of running on a fixed schedule, provide a poll_fn. It's called on each schedule tick — return data to fire the DAG, or None to skip:

def check_queue() -> dict[str, str] | None:
    """Check for new messages. Returns payload or None."""
    # ... check external source ...
    return {"message": "new order"} if has_messages else None

ns.register_trigger(
    name="queue_check",
    dag_nsref="pipelines:process_order",
    trigger_type="poll",
    poll_fn=check_queue,
    schedule="*/10 * * * * *",  # every 10 seconds
)

When poll_fn returns data, it becomes the DAG's input payload — the dict keys are matched to the DAG's source node parameters by name.

A check that returns None starts no run and records no event, and the next check waits for the next scheduled time, so a poll_fn that calls an external API is called once per schedule tick, not once per poll-loop second.