LaminDB Sensors¶

The sensors wait for a condition inside a running DAG. They support deferrable=True (or [operators] default_deferrable), which moves the wait to the triggerer so that no worker slot is occupied. For scheduling DAGs on LaminDB changes, use the event triggers instead.

Network errors, rate limiting (429) and LaminHub server errors (5xx) don't end the wait: the sensors log them and check again at the next poke, until their timeout. Other errors fail the task, for example an invalid filter, a missing permission or an instance on lamindb older than 2.8.1.

LaminDBBranchStatusSensor¶

LaminDBBranchStatusSensor waits until a branch reaches a status (target_status, default merged), for example until a Change Request is merged. If the branch reaches a failure_status (default closed), the task fails without retries. The branch, with its status, is returned via XCom.

wait_for_merge = LaminDBBranchStatusSensor(
    task_id="wait_for_merge",
    branch="{{ params.branch }}",
    target_status="merged",
    failure_status="closed",
    deferrable=True,
    poke_interval=60,
)

LaminDBRecordSensor¶

LaminDBRecordSensor waits until at least min_count records of a registry match a filter on a branch (branch, default main). The matching records are returned via XCom: at most limit records, newest first.

wait_for_run = LaminDBRecordSensor(
    task_id="wait_for_run",
    registry=LaminDBRegistry.RUN,
    # completed runs of the transform (script, notebook, ...) with the key "preprocess.py"
    filter=(F(RunField.TRANSFORM, TransformField.KEY) == "preprocess.py")
    & (F(RunField.STATUS_CODE) == RunStatus.COMPLETED),
    deferrable=True,
    poke_interval=60,
)

The same filter as a LaminHub REST filter: {"and": [{"transform.key": {"eq": "preprocess.py"}}, {"_status_code": {"eq": 0}}]}.

LaminDBArtifactSensor¶

LaminDBArtifactSensor waits for artifacts by key, key_prefix, suffix or kind (e.g. ArtifactKind.DATASET) and an additional filter, and ignores artifacts whose upload to storage is still in progress.

wait_for_artifact = LaminDBArtifactSensor(
    task_id="wait_for_artifact",
    key="{{ params.key }}",
    deferrable=True,
    poke_interval=60,
)

LaminDBHook¶

LaminDBHook can be used in tasks to query the LaminHub REST API directly. Every method also has an async variant with an a prefix, such as aquery_records:

• query_records(registry, filter, order_by=..., limit=...): query any registry (paginated) with a filter
• get_records_by_ids(registry, ids)
• get_branch(name_or_id)
• query_dbwrites(filter, after_id=...): the database write log
• get_instance() and get_schema()
• insert_records(registry, records) and update_record(registry, uid, values): write records; the API key needs write access to the instance. Inserts are not retried after errors that may have reached LaminHub, so a failed request never creates a record twice.

@task
def summarize(branch, artifacts, runs):
    hook = LaminDBHook()
    recent_runs = hook.query_records(
        LaminDBRegistry.RUN, F(RunField.FINISHED_AT).is_not_null(), order_by=["-started_at"], limit=5
    )
    print(f"Branch {branch['name']} is {branch['status']}; artifact {artifacts[0]['uid']} is ready")
    print(f"Run {runs[0]['uid']} completed; recent runs: {[run['uid'] for run in recent_runs]}")