lythonic.compose.trigger

Event-driven DAG trigger system.

Trigger: Event-driven execution of namespace nodes.

Provides TriggerStore for activation state persistence and TriggerManager for runtime coordination. Trigger definitions live on NsNodeConfig.triggers as TriggerConfig instances.

Call TriggerManager.activate_configured() on every process start: it activates the triggers the node configs define, but leaves those stored as disabled, so a deliberate deactivate() survives restarts until an explicit activate(). Scheduled fires missed while the process was down run once when polling resumes: re-activating an active trigger keeps its last_run_at. A run that overruns later scheduled fire times skips them and logs a warning.

Every poll_fn call counts as the scheduled check, even when it returns None or raises: it sets last_checked_at, and the next check waits for the next scheduled time. The schedule counts from last_checked_at, falling back to last_run_at and then created_at; last_run_at moves only on a real fire, which sets both.

Runs fired by the schedule record initiated_by = SCHEDULER; runs fired with TriggerManager.fire record the caller's run_initiator (see lythonic.compose.dag_provenance).

activate and deactivate record run_initiator as changed_by, with changed_at, when it is set or when the call changes the trigger's status. So activate_configured() at process start keeps the last deliberate change, while an explicit activate() of a disabled trigger with no initiator records changed_by = NULL rather than keeping the name of whoever disabled it.

TriggerStore

SQLite-backed storage for trigger activations and events.

Source code in src/lythonic/compose/trigger.py

class TriggerStore:
    """SQLite-backed storage for trigger activations and events."""

    db_path: Path

    def __init__(self, db_path: Path) -> None:
        self.db_path = db_path
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        with open_sqlite_db(self.db_path) as conn:
            cursor = conn.cursor()
            execute_sql(cursor, _TRIGGER_ACTIVATIONS_DDL)
            add_missing_column(cursor, "trigger_activations", "changed_by", "TEXT")
            add_missing_column(cursor, "trigger_activations", "changed_at", "REAL")
            add_missing_column(cursor, "trigger_activations", "last_checked_at", "REAL")
            execute_sql(cursor, _TRIGGER_EVENTS_DDL)
            conn.commit()

    def activate(
        self, trigger_config: TriggerConfig, dag_nsref: str, *, keep_disabled: bool = False
    ) -> bool:
        """
        Create or update an activation record from a trigger config. Returns whether
        the trigger is now active: with `keep_disabled`, a trigger stored as `disabled`
        is left untouched, checked in the same transaction.

        Re-activating an active trigger (e.g. on every process start) keeps `last_run_at`,
        `last_run_id` and `created_at`, even if the config changed, so a scheduled fire
        missed while the process was down fires once when polling resumes. Activating a
        disabled trigger resets `last_run_at` and `last_checked_at` to now, so a deliberate
        pause is not caught up.
        """
        config: dict[str, Any] = {}
        if trigger_config.schedule is not None:
            config["schedule"] = trigger_config.schedule
        if trigger_config.poll_fn is not None:
            config["poll_fn"] = str(trigger_config.poll_fn)

        now = time.time()
        with open_sqlite_db(self.db_path) as conn:
            cursor = conn.cursor()
            status_before = _status(cursor, trigger_config.name)
            if keep_disabled and status_before == "disabled":
                return False
            execute_sql(
                cursor,
                "INSERT INTO trigger_activations "
                "(name, dag_nsref, trigger_type, status, last_run_at, created_at, config_json) "
                "VALUES (?, ?, ?, ?, ?, ?, ?) "
                "ON CONFLICT(name) DO UPDATE SET "
                "dag_nsref = excluded.dag_nsref, "
                "trigger_type = excluded.trigger_type, "
                "status = excluded.status, "
                "config_json = excluded.config_json, "
                "last_run_at = CASE WHEN trigger_activations.status = 'disabled' "
                "THEN excluded.last_run_at ELSE trigger_activations.last_run_at END, "
                # excluded.last_run_at is now: re-enabling restarts the check schedule too.
                "last_checked_at = CASE WHEN trigger_activations.status = 'disabled' "
                "THEN excluded.last_run_at ELSE trigger_activations.last_checked_at END",
                (
                    trigger_config.name,
                    dag_nsref,
                    trigger_config.type,
                    "active",
                    now,
                    now,
                    json.dumps(config),
                ),
            )
            _record_change(cursor, trigger_config.name, status_before, now)
            conn.commit()
        return True

    def deactivate(self, name: str) -> None:
        """Set activation status to disabled."""
        with open_sqlite_db(self.db_path) as conn:
            cursor = conn.cursor()
            status_before = _status(cursor, name)
            execute_sql(
                cursor,
                "UPDATE trigger_activations SET status = ? WHERE name = ?",
                ("disabled", name),
            )
            _record_change(cursor, name, status_before, time.time())
            conn.commit()

    def get_activation(self, name: str) -> dict[str, Any] | None:
        """Get activation record by name."""
        with open_sqlite_db(self.db_path) as conn:
            cursor = conn.cursor()
            execute_sql(
                cursor,
                "SELECT * FROM trigger_activations WHERE name = ?",
                (name,),
            )
            row = cursor.fetchone()
            if row is None:
                return None
            cols = [d[0] for d in cursor.description]
            return dict(zip(cols, row, strict=False))

    def get_active_poll_triggers(self) -> list[dict[str, Any]]:
        """Get all active poll trigger activations."""
        with open_sqlite_db(self.db_path) as conn:
            cursor = conn.cursor()
            execute_sql(
                cursor,
                "SELECT * FROM trigger_activations WHERE trigger_type = ? AND status = ?",
                ("poll", "active"),
            )
            cols = [d[0] for d in cursor.description]
            return [dict(zip(cols, row, strict=False)) for row in cursor.fetchall()]

    def record_event(
        self,
        trigger_name: str,
        payload: dict[str, Any] | None = None,
        run_id: str | None = None,
        status: str = "pending",
    ) -> str:
        """Record a trigger event. Returns the event_id."""
        event_id = str(uuid.uuid4())
        with open_sqlite_db(self.db_path) as conn:
            cursor = conn.cursor()
            execute_sql(
                cursor,
                "INSERT INTO trigger_events "
                "(event_id, trigger_name, fired_at, run_id, payload_json, status) "
                "VALUES (?, ?, ?, ?, ?, ?)",
                (
                    event_id,
                    trigger_name,
                    time.time(),
                    run_id,
                    safe_json_dumps(payload) if payload else None,
                    status,
                ),
            )
            conn.commit()
        return event_id

    def get_events(self, trigger_name: str) -> list[dict[str, Any]]:
        """Get all events for a trigger, ordered most recent first."""
        with open_sqlite_db(self.db_path) as conn:
            cursor = conn.cursor()
            execute_sql(
                cursor,
                "SELECT * FROM trigger_events WHERE trigger_name = ? ORDER BY fired_at DESC",
                (trigger_name,),
            )
            cols = [d[0] for d in cursor.description]
            return [dict(zip(cols, row, strict=False)) for row in cursor.fetchall()]

    def record_check(self, name: str) -> None:
        """
        Record a scheduled `poll_fn` check, so the next check waits for the next
        scheduled time even if this one found nothing or failed. Leaves
        `last_run_at` alone: it only moves on a real fire.
        """
        with open_sqlite_db(self.db_path) as conn:
            cursor = conn.cursor()
            execute_sql(
                cursor,
                "UPDATE trigger_activations SET last_checked_at = ? WHERE name = ?",
                (time.time(), name),
            )
            conn.commit()

    def update_last_run(self, name: str, run_id: str) -> None:
        """Update last run timestamp and run ID for an activation. A fire is also a check."""
        now = time.time()
        with open_sqlite_db(self.db_path) as conn:
            cursor = conn.cursor()
            execute_sql(
                cursor,
                "UPDATE trigger_activations SET last_run_at = ?, last_checked_at = ?, "
                "last_run_id = ? WHERE name = ?",
                (now, now, run_id, name),
            )
            conn.commit()

activate(trigger_config, dag_nsref, *, keep_disabled=False)

Create or update an activation record from a trigger config. Returns whether the trigger is now active: with keep_disabled, a trigger stored as disabled is left untouched, checked in the same transaction.

Re-activating an active trigger (e.g. on every process start) keeps last_run_at, last_run_id and created_at, even if the config changed, so a scheduled fire missed while the process was down fires once when polling resumes. Activating a disabled trigger resets last_run_at and last_checked_at to now, so a deliberate pause is not caught up.

Source code in src/lythonic/compose/trigger.py

def activate(
    self, trigger_config: TriggerConfig, dag_nsref: str, *, keep_disabled: bool = False
) -> bool:
    """
    Create or update an activation record from a trigger config. Returns whether
    the trigger is now active: with `keep_disabled`, a trigger stored as `disabled`
    is left untouched, checked in the same transaction.

    Re-activating an active trigger (e.g. on every process start) keeps `last_run_at`,
    `last_run_id` and `created_at`, even if the config changed, so a scheduled fire
    missed while the process was down fires once when polling resumes. Activating a
    disabled trigger resets `last_run_at` and `last_checked_at` to now, so a deliberate
    pause is not caught up.
    """
    config: dict[str, Any] = {}
    if trigger_config.schedule is not None:
        config["schedule"] = trigger_config.schedule
    if trigger_config.poll_fn is not None:
        config["poll_fn"] = str(trigger_config.poll_fn)

    now = time.time()
    with open_sqlite_db(self.db_path) as conn:
        cursor = conn.cursor()
        status_before = _status(cursor, trigger_config.name)
        if keep_disabled and status_before == "disabled":
            return False
        execute_sql(
            cursor,
            "INSERT INTO trigger_activations "
            "(name, dag_nsref, trigger_type, status, last_run_at, created_at, config_json) "
            "VALUES (?, ?, ?, ?, ?, ?, ?) "
            "ON CONFLICT(name) DO UPDATE SET "
            "dag_nsref = excluded.dag_nsref, "
            "trigger_type = excluded.trigger_type, "
            "status = excluded.status, "
            "config_json = excluded.config_json, "
            "last_run_at = CASE WHEN trigger_activations.status = 'disabled' "
            "THEN excluded.last_run_at ELSE trigger_activations.last_run_at END, "
            # excluded.last_run_at is now: re-enabling restarts the check schedule too.
            "last_checked_at = CASE WHEN trigger_activations.status = 'disabled' "
            "THEN excluded.last_run_at ELSE trigger_activations.last_checked_at END",
            (
                trigger_config.name,
                dag_nsref,
                trigger_config.type,
                "active",
                now,
                now,
                json.dumps(config),
            ),
        )
        _record_change(cursor, trigger_config.name, status_before, now)
        conn.commit()
    return True

deactivate(name)

Set activation status to disabled.

Source code in src/lythonic/compose/trigger.py

def deactivate(self, name: str) -> None:
    """Set activation status to disabled."""
    with open_sqlite_db(self.db_path) as conn:
        cursor = conn.cursor()
        status_before = _status(cursor, name)
        execute_sql(
            cursor,
            "UPDATE trigger_activations SET status = ? WHERE name = ?",
            ("disabled", name),
        )
        _record_change(cursor, name, status_before, time.time())
        conn.commit()

get_activation(name)

Get activation record by name.

Source code in src/lythonic/compose/trigger.py

def get_activation(self, name: str) -> dict[str, Any] | None:
    """Get activation record by name."""
    with open_sqlite_db(self.db_path) as conn:
        cursor = conn.cursor()
        execute_sql(
            cursor,
            "SELECT * FROM trigger_activations WHERE name = ?",
            (name,),
        )
        row = cursor.fetchone()
        if row is None:
            return None
        cols = [d[0] for d in cursor.description]
        return dict(zip(cols, row, strict=False))

get_active_poll_triggers()

Get all active poll trigger activations.

Source code in src/lythonic/compose/trigger.py

def get_active_poll_triggers(self) -> list[dict[str, Any]]:
    """Get all active poll trigger activations."""
    with open_sqlite_db(self.db_path) as conn:
        cursor = conn.cursor()
        execute_sql(
            cursor,
            "SELECT * FROM trigger_activations WHERE trigger_type = ? AND status = ?",
            ("poll", "active"),
        )
        cols = [d[0] for d in cursor.description]
        return [dict(zip(cols, row, strict=False)) for row in cursor.fetchall()]

record_event(trigger_name, payload=None, run_id=None, status='pending')

Record a trigger event. Returns the event_id.

Source code in src/lythonic/compose/trigger.py

def record_event(
    self,
    trigger_name: str,
    payload: dict[str, Any] | None = None,
    run_id: str | None = None,
    status: str = "pending",
) -> str:
    """Record a trigger event. Returns the event_id."""
    event_id = str(uuid.uuid4())
    with open_sqlite_db(self.db_path) as conn:
        cursor = conn.cursor()
        execute_sql(
            cursor,
            "INSERT INTO trigger_events "
            "(event_id, trigger_name, fired_at, run_id, payload_json, status) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (
                event_id,
                trigger_name,
                time.time(),
                run_id,
                safe_json_dumps(payload) if payload else None,
                status,
            ),
        )
        conn.commit()
    return event_id

get_events(trigger_name)

Get all events for a trigger, ordered most recent first.

Source code in src/lythonic/compose/trigger.py

def get_events(self, trigger_name: str) -> list[dict[str, Any]]:
    """Get all events for a trigger, ordered most recent first."""
    with open_sqlite_db(self.db_path) as conn:
        cursor = conn.cursor()
        execute_sql(
            cursor,
            "SELECT * FROM trigger_events WHERE trigger_name = ? ORDER BY fired_at DESC",
            (trigger_name,),
        )
        cols = [d[0] for d in cursor.description]
        return [dict(zip(cols, row, strict=False)) for row in cursor.fetchall()]

record_check(name)

Record a scheduled poll_fn check, so the next check waits for the next scheduled time even if this one found nothing or failed. Leaves last_run_at alone: it only moves on a real fire.

Source code in src/lythonic/compose/trigger.py

def record_check(self, name: str) -> None:
    """
    Record a scheduled `poll_fn` check, so the next check waits for the next
    scheduled time even if this one found nothing or failed. Leaves
    `last_run_at` alone: it only moves on a real fire.
    """
    with open_sqlite_db(self.db_path) as conn:
        cursor = conn.cursor()
        execute_sql(
            cursor,
            "UPDATE trigger_activations SET last_checked_at = ? WHERE name = ?",
            (time.time(), name),
        )
        conn.commit()

update_last_run(name, run_id)

Update last run timestamp and run ID for an activation. A fire is also a check.

Source code in src/lythonic/compose/trigger.py

def update_last_run(self, name: str, run_id: str) -> None:
    """Update last run timestamp and run ID for an activation. A fire is also a check."""
    now = time.time()
    with open_sqlite_db(self.db_path) as conn:
        cursor = conn.cursor()
        execute_sql(
            cursor,
            "UPDATE trigger_activations SET last_run_at = ?, last_checked_at = ?, "
            "last_run_id = ? WHERE name = ?",
            (now, now, run_id, name),
        )
        conn.commit()

TriggerManager

Runtime coordinator for triggers. Reads trigger definitions from node configs in the namespace. activate() creates DB records, fire() runs triggers, start()/stop() runs a background poll loop. Call activate_configured() on every process start.

Source code in src/lythonic/compose/trigger.py

class TriggerManager:
    """
    Runtime coordinator for triggers. Reads trigger definitions from
    node configs in the namespace. `activate()` creates DB records,
    `fire()` runs triggers, `start()`/`stop()` runs a background poll loop.
    Call `activate_configured()` on every process start.
    """

    namespace: Namespace
    store: TriggerStore
    provenance: DagProvenance | NullProvenance

    def __init__(
        self,
        namespace: Namespace,
        store: TriggerStore,
        provenance: DagProvenance | NullProvenance | None = None,
    ) -> None:
        from lythonic.compose.dag_provenance import NullProvenance

        self.namespace = namespace
        self.store = store
        self.provenance = provenance or NullProvenance()
        self._task: asyncio.Task[None] | None = None
        self._shutdown: asyncio.Event = asyncio.Event()

    def activate(self, name: str) -> None:
        """Activate a trigger by name (found in node configs), even if it is disabled."""
        node, tc = self.namespace.get_trigger(name)
        self.store.activate(tc, dag_nsref=str(node.nsref))

    def activate_configured(self) -> TriggerActivationResult:
        """
        Activate every trigger defined in node configs, except those stored as
        `disabled`: a deliberate `deactivate()` stays in force across restarts
        until an explicit `activate()`.
        """
        activated: list[str] = []
        left_disabled: list[str] = []
        for node, tc in self.namespace.triggers():
            assert tc.name is not None, "trigger name is set at registration"
            if self.store.activate(tc, dag_nsref=str(node.nsref), keep_disabled=True):
                activated.append(tc.name)
            else:
                left_disabled.append(tc.name)
        return TriggerActivationResult(activated=activated, left_disabled=left_disabled)

    def deactivate(self, name: str) -> None:
        """Deactivate a trigger."""
        self.store.deactivate(name)

    async def fire(self, name: str, payload: dict[str, Any] | None = None) -> DagRunResult:
        """
        Fire a trigger: run the associated node with payload as inputs.
        Records the event in the store.
        """
        activation = self.store.get_activation(name)
        if activation is None or activation["status"] != "active":
            raise ValueError(
                f"Trigger '{name}' is not active (status: {activation['status'] if activation else 'not found'})"
            )

        dag_nsref = activation["dag_nsref"]
        dag = self.namespace.get_as_dag(dag_nsref)

        # Resolve payload: explicit overrides config default
        _, tc = self.namespace.get_trigger(name)
        effective_payload = payload if payload is not None else (tc.payload or {})

        from lythonic.compose.dag_runner import DagRunner

        # Run with provenance so dag_runs/node_executions are recorded
        runner = DagRunner(dag, provenance=self.provenance)
        source_inputs: dict[str, dict[str, Any]] = {}
        for src_node in dag.sources():
            node_args = src_node.ns_node.method.args
            if src_node.ns_node.expects_dag_context():
                node_args = node_args[1:]
            node_kwargs = {
                a.name: effective_payload[a.name] for a in node_args if a.name in effective_payload
            }
            if node_kwargs:
                source_inputs[src_node.label] = node_kwargs
        result: DagRunResult = await runner.run(source_inputs=source_inputs, dag_nsref=dag_nsref)

        self.store.record_event(
            trigger_name=name,
            payload=effective_payload,
            run_id=result.run_id,
            status=result.status,
        )
        self.store.update_last_run(name, result.run_id)
        return result

    def start(self) -> None:
        """Start the background asyncio task for polling active poll triggers."""
        if self._task is not None and not self._task.done():
            return
        self._shutdown = asyncio.Event()
        self._task = asyncio.create_task(self._poll_loop())

    def stop(self) -> None:
        """Signal the poll loop to stop and cancel the background task."""
        if self._task is not None and not self._task.done():
            self._shutdown.set()
            self._task.cancel()

    async def _poll_loop(self) -> None:
        """Background loop that checks active poll triggers on their intervals."""
        while not self._shutdown.is_set():
            try:
                active_polls = self.store.get_active_poll_triggers()
                now = time.time()

                for activation in active_polls:
                    config = json.loads(activation.get("config_json") or "{}")
                    schedule = config.get("schedule")
                    if not schedule:
                        continue

                    schedule_from = (
                        activation.get("last_checked_at")
                        or activation.get("last_run_at")
                        or activation.get("created_at")
                        or 0
                    )
                    next_fire = _croniter_for(schedule, schedule_from).get_next(float)

                    if now < next_fire:
                        continue

                    poll_fn_gref = config.get("poll_fn")
                    payload: dict[str, Any] | None
                    if poll_fn_gref:
                        from lythonic import GlobalRef

                        fn = GlobalRef(poll_fn_gref).get_instance()
                        try:
                            result: Any = fn()
                        finally:
                            # An empty or failed check, or a failed fire after it, still
                            # waits for the next scheduled time instead of the next tick.
                            self.store.record_check(activation["name"])
                        if result is None:
                            continue
                        payload = dict(result) if isinstance(result, dict) else {"data": result}  # pyright: ignore[reportUnknownArgumentType,reportUnknownVariableType]
                    else:
                        payload = None

                    started_at = time.time()
                    try:
                        with initiated_by(SCHEDULER):
                            await self.fire(activation["name"], payload=payload)
                    except Exception:
                        _log.exception("Error firing poll trigger '%s'", activation["name"])
                    else:
                        _warn_skipped_fires(activation["name"], schedule, started_at, time.time())

                await asyncio.sleep(1)
            except asyncio.CancelledError:
                break
            except Exception:
                _log.exception("Error in poll loop")
                await asyncio.sleep(1)

activate(name)

Activate a trigger by name (found in node configs), even if it is disabled.

Source code in src/lythonic/compose/trigger.py

def activate(self, name: str) -> None:
    """Activate a trigger by name (found in node configs), even if it is disabled."""
    node, tc = self.namespace.get_trigger(name)
    self.store.activate(tc, dag_nsref=str(node.nsref))

activate_configured()

Activate every trigger defined in node configs, except those stored as disabled: a deliberate deactivate() stays in force across restarts until an explicit activate().

Source code in src/lythonic/compose/trigger.py

def activate_configured(self) -> TriggerActivationResult:
    """
    Activate every trigger defined in node configs, except those stored as
    `disabled`: a deliberate `deactivate()` stays in force across restarts
    until an explicit `activate()`.
    """
    activated: list[str] = []
    left_disabled: list[str] = []
    for node, tc in self.namespace.triggers():
        assert tc.name is not None, "trigger name is set at registration"
        if self.store.activate(tc, dag_nsref=str(node.nsref), keep_disabled=True):
            activated.append(tc.name)
        else:
            left_disabled.append(tc.name)
    return TriggerActivationResult(activated=activated, left_disabled=left_disabled)

deactivate(name)

Deactivate a trigger.

Source code in src/lythonic/compose/trigger.py

def deactivate(self, name: str) -> None:
    """Deactivate a trigger."""
    self.store.deactivate(name)

fire(name, payload=None) async

Fire a trigger: run the associated node with payload as inputs. Records the event in the store.

Source code in src/lythonic/compose/trigger.py

async def fire(self, name: str, payload: dict[str, Any] | None = None) -> DagRunResult:
    """
    Fire a trigger: run the associated node with payload as inputs.
    Records the event in the store.
    """
    activation = self.store.get_activation(name)
    if activation is None or activation["status"] != "active":
        raise ValueError(
            f"Trigger '{name}' is not active (status: {activation['status'] if activation else 'not found'})"
        )

    dag_nsref = activation["dag_nsref"]
    dag = self.namespace.get_as_dag(dag_nsref)

    # Resolve payload: explicit overrides config default
    _, tc = self.namespace.get_trigger(name)
    effective_payload = payload if payload is not None else (tc.payload or {})

    from lythonic.compose.dag_runner import DagRunner

    # Run with provenance so dag_runs/node_executions are recorded
    runner = DagRunner(dag, provenance=self.provenance)
    source_inputs: dict[str, dict[str, Any]] = {}
    for src_node in dag.sources():
        node_args = src_node.ns_node.method.args
        if src_node.ns_node.expects_dag_context():
            node_args = node_args[1:]
        node_kwargs = {
            a.name: effective_payload[a.name] for a in node_args if a.name in effective_payload
        }
        if node_kwargs:
            source_inputs[src_node.label] = node_kwargs
    result: DagRunResult = await runner.run(source_inputs=source_inputs, dag_nsref=dag_nsref)

    self.store.record_event(
        trigger_name=name,
        payload=effective_payload,
        run_id=result.run_id,
        status=result.status,
    )
    self.store.update_last_run(name, result.run_id)
    return result

start()

Start the background asyncio task for polling active poll triggers.

Source code in src/lythonic/compose/trigger.py

def start(self) -> None:
    """Start the background asyncio task for polling active poll triggers."""
    if self._task is not None and not self._task.done():
        return
    self._shutdown = asyncio.Event()
    self._task = asyncio.create_task(self._poll_loop())

stop()

Signal the poll loop to stop and cancel the background task.

Source code in src/lythonic/compose/trigger.py

def stop(self) -> None:
    """Signal the poll loop to stop and cancel the background task."""
    if self._task is not None and not self._task.done():
        self._shutdown.set()
        self._task.cancel()