LaminDB Triggers: event-driven scheduling¶

The provider's event triggers let DAGs run when something changes in a LaminDB instance, using Airflow's event-driven scheduling: attach a trigger to an Asset through an AssetWatcher, and schedule DAGs on that asset.

How it works¶

LaminHub logs every INSERT, UPDATE and DELETE on an instance's database in the hubmodule.dbwrite registry, shown as Changes → Database writes in the LaminHub UI. The triggers poll this log through the LaminHub REST API. Each log entry has a strictly increasing id that serves as a cursor, and each entry records which record was written on which branch, along with the previous values of the changed fields. From this, the triggers derive semantic events such as "artifact created on main" or "branch moved from review to merged".

Only hosted instances. The database write log is a LaminHub feature. Instances that are not managed by LaminHub are not supported.
Only lamindb 2.8.1 or newer. A trigger never polls an instance on an older lamindb: it logs a LaminDBVersionError and retries with backoff, like after any other error, so it starts once the instance is migrated.
Resuming after restarts. The cursor is stored (see Where the cursor is stored), so a restarted triggerer resumes where it left off. It's only saved when the trigger processed new writes that match its filter, at most once per poll. On the very first start, a trigger only reports changes from then on.
At-least-once delivery. The cursor is committed past events only once they were handed to Airflow at least the store's commit_delay ago (by default 5 seconds), checked at the start of each poll, so at least one poll cycle later. Events are not lost if the triggerer crashes, but after a crash an event may be delivered twice. Each payload contains the dbwrite entry, whose id/uid identify the change, so DAGs can deduplicate if needed.
Batching. Every change is a separate trigger event (and asset event). Airflow may combine several asset events into one DAG run; iterate over triggering_asset_events.
Changing a trigger resets its cursor. The state key is derived from the trigger's arguments except the polling settings (poll_interval, batch_size, max_batches_per_poll) and cursor_store, so changing a filter, the connection or the instance starts a fresh cursor. Changing cursor_store moves the cursor to another place; see Where the cursor is stored.

Note

Define a watched asset in a single DAG file. Airflow registers watchers per DAG file. If another DAG file defines the same asset without watchers, the DAG processor keeps removing the watcher again. In other DAG files, refer to the asset with Asset.ref(name=...).

Common arguments¶

All event triggers accept:

lamindb_conn_id: The LaminDB connection (default lamindb_default).
instance: Instance owner/name; overrides the connection's instance.
poll_interval: Seconds between polls (default 30).
batch_size / max_batches_per_poll: Paging of the write log (defaults 200 and 10).
cursor_store: Where and how the cursor is stored: AssetCursorStore() (default) or RecordCursorStore(); see Where the cursor is stored.

Where the cursor is stored¶

Pass a trigger a cursor store from airflow.providers.lamindb.triggers.cursors as cursor_store. Both take commit_delay: the minimum seconds between handing events to Airflow and committing the cursor past them (default 5). Longer delays save the cursor less often, but redeliver more events after a crash.

AssetCursorStore() (default)

In the watched asset's asset state store, which Airflow 3.3 added for event-driven scheduling. The cursor lives in Airflow's metadata database next to the asset events, so restoring that database also restores matching cursors. A trigger outside an AssetWatcher has no asset state store and keeps its cursor in memory only.

RecordCursorStore()

In a record of the watched instance, one per trigger. By default it's a core.record on main, named after the trigger's state key and grouped under the record type Airflow trigger cursors. The cursor is in the record's extra_data, and the description says which trigger it belongs to. Use it for triggers outside an AssetWatcher, or to keep cursors when Airflow's metadata database is reset. Keep in mind:

The connection's API key needs write access to the instance, in the cursor records' space. The record type and record are created on the first save.
Each save is a database write, so it appears in the instance's history and write log. Triggers for artifacts, branches and other registries never see it. LaminDBRecordEventTrigger skips cursor records in the registry it watches: it doesn't report them, and it doesn't save its cursor after a batch of only cursor writes, so triggers never react to each other's saves.
The cursor outlives Airflow's metadata database. After restoring an older backup of that database, the trigger resumes from the newer cursor, and the asset events in between are not delivered again.
To reset the cursor, delete the record (or move it to the trash) and restart the triggerer. The trigger then starts again from the latest write.

Configure the cursor record with the arguments of RecordCursorStore, all keyword-only and optional:

registry (default core.record): Registry of the cursor records. It needs the fields name, branch, space and the field below, plus reference_type and type/is_type unless those are turned off.
field (default extra_data): JSON field that holds the cursor, {"dbwrite_id": ..., "instance_id": ...}.
record_type (default Airflow trigger cursors): Name of the record type that groups the cursor records. None for no type, for registries without types.
reference_type (default airflow_trigger_cursor): Marker that identifies cursor records. None for no marker; then record triggers on registry report the cursor records like any other record. A record trigger that watches registry itself needs a marker to recognize its own saves, so it rejects None.
branch (default main): Branch of the cursor records and their type, by name or id.
space (default None: LaminHub's default space): Space of the cursor records and their type, by name or id, for example a restricted space that only the Airflow API key can write to.
name (default None: the trigger's state key): Name of the cursor record. A fixed name must be unique per trigger, or triggers overwrite each other's cursors.
description (default None: describes the trigger): Description of the cursor record.

from airflow.providers.lamindb.triggers.cursors import RecordCursorStore

LaminDBArtifactEventTrigger(
    key_prefix="raw/",
    cursor_store=RecordCursorStore(space="Airflow", name="raw-artifacts-watcher", commit_delay=30),
)

A trigger serializes its cursor store as a dict with a type key (asset or record) and the settings, and accepts that dict as cursor_store too.

A record trigger recognizes cursor records by the registry and reference_type of its own RecordCursorStore, and by the defaults if it stores its cursor in the asset state store. Cursor records with other settings are ordinary records to it:

A record trigger with the asset state store reports them as events (but never writes to LaminDB).
Two record triggers with a RecordCursorStore that watch the same registry, and store their cursors there with different reference_type markers, report each other's saves and save again in response, every poll. Give them the same registry and reference_type.

Changing branch, space, name, registry or the markers points the trigger at another record, so its cursor starts over from the latest write. The old record stays until you delete it.

Every event payload contains instance, instance_id, changed_by (id, handle and name of the LaminDB user) and dbwrite: the write log entry with id, uid, event_type, table_name, sqlrecord_id, created_at, branch_id, space_id and run_id. In a DAG, the payload is available as event.extra["payload"].

Artifacts¶

LaminDBArtifactEventTrigger reports artifacts that are created, updated or deleted (argument events, default ["created"]). Filter them with key, key_prefix, suffix, kind and a general filter.

With wait_for_upload=True (default), created fires only once the artifact's upload to storage has completed. LaminDB creates the artifact record before it uploads the file and deletes the record again if the upload fails. Downstream tasks can load the artifact right away.
Artifacts that LaminDB creates internally (run logs, environments) are ignored unless include_internal=True.

tests/system/lamindb/example_lamindb_event_scheduling.py

new_fastqs = Asset(
    "lamindb_new_fastqs",
    watchers=[
        AssetWatcher(
            name="lamindb_new_fastqs_watcher",
            trigger=LaminDBArtifactEventTrigger(
                lamindb_conn_id="lamindb_default",
                key_prefix="raw/",
                suffix=".fastq.gz",
                poll_interval=30,
            ),
        )
    ],
)


@dag(schedule=[new_fastqs], tags=["example", "lamindb"])
def example_lamindb_artifact_events():
    @task
    def process_new_fastqs(triggering_asset_events=None):
        # several LaminDB events can be batched into one DAG run
        for event in triggering_asset_events[new_fastqs]:
            payload = event.extra["payload"]
            artifact = payload["record"]
            print(f"New artifact {artifact['key']} ({artifact['uid']}) in {payload['instance']}")

    process_new_fastqs()

Records of any registry¶

LaminDBRecordEventTrigger watches any registry, e.g. core.collection, core.run, core.transform, core.record, core.ulabel or bionty.celltype.

Branch-aware semantics (branch, default main): created covers records inserted on the branch and records that arrive on it by merging a contribution branch or by restoring them from the trash or archive. updated covers changes of records on the branch. deleted covers records deleted from the branch and records moved off it, e.g. to the trash by record.delete(). With branch=None, the trigger reports raw inserts, updates and deletes on all branches. For registries without branches, branch is ignored.
Filters: filter is a filter such as F(CollectionField.KEY).startswith("raw/") or F(CollectionField.CREATED_BY, UserField.HANDLE) == "alice", or its LaminHub REST form such as {"created_by.handle": {"eq": "alice"}}. The API evaluates it against the current record. Hard-deleted records can't be queried anymore, so deleted events only support filters on direct fields of the record, such as F(CollectionField.CREATED_BY_ID) == 7.
Updates: changed_fields restricts updated events to changes of certain fields, e.g. [CollectionField.KEY].
Payload: event, registry, table_name, record_id, record (the current record or None for hard deletes), previous (the previous values of the changed fields, or the full deleted row) and changed_fields.

tests/system/lamindb/example_lamindb_event_scheduling.py

cell_type_changes = Asset(
    "lamindb_cell_type_changes",
    watchers=[
        AssetWatcher(
            name="lamindb_cell_type_changes_watcher",
            trigger=LaminDBRecordEventTrigger(
                "bionty.celltype",
                events=["created", "updated", "deleted"],
                changed_fields=["name", "ontology_id"],
            ),
        )
    ],
)


@dag(schedule=[cell_type_changes], tags=["example", "lamindb"])
def example_lamindb_record_events():
    @task
    def sync_cell_types(triggering_asset_events=None):
        for event in triggering_asset_events[cell_type_changes]:
            payload = event.extra["payload"]
            print(payload["event"], payload["record_id"], payload["changed_fields"], payload["previous"])

    sync_cell_types()

Branches: status changes, comments and readmes¶

LaminDBBranchStatusEventTrigger reports status changes of branches (Change Requests): standalone, draft, review, merged and closed. Filter with from_status, to_status and branch_name (glob patterns such as "release-*"). The payload contains from_status, to_status and the branch with its current status.

LaminDBBranchBlockEventTrigger reports new comments and readmes on branches (kinds, branch_name). The payload contains the kind, the block with its Markdown content, and the branch.

tests/system/lamindb/example_lamindb_event_scheduling.py

review_requested = Asset(
    "lamindb_review_requested",
    watchers=[
        AssetWatcher(
            name="lamindb_review_requested_watcher",
            trigger=LaminDBBranchStatusEventTrigger(to_status="review"),
        )
    ],
)
branch_comments = Asset(
    "lamindb_branch_comments",
    watchers=[
        AssetWatcher(
            name="lamindb_branch_comments_watcher",
            trigger=LaminDBBranchBlockEventTrigger(kinds="comment", branch_name="release-*"),
        )
    ],
)


@dag(schedule=[review_requested], tags=["example", "lamindb"])
def example_lamindb_branch_review():
    @task
    def validate_change_request(triggering_asset_events=None):
        for event in triggering_asset_events[review_requested]:
            payload = event.extra["payload"]
            branch = payload["branch"]
            print(f"Validate branch {branch['name']}: {payload['from_status']} -> {payload['to_status']}")

    validate_change_request()


@dag(schedule=[branch_comments], tags=["example", "lamindb"])
def example_lamindb_branch_comments():
    @task
    def notify(triggering_asset_events=None):
        for event in triggering_asset_events[branch_comments]:
            payload = event.extra["payload"]
            author = (payload["changed_by"] or {}).get("handle")
            print(f"{author} commented on {payload['branch']['name']}: {payload['block']['content']}")

    notify()