# 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](<https://danplischke.github.io/apache-airflow-providers-lamindb/triggers/index.md>) 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.

tests/system/lamindb/example\_lamindb\_sensors.py

```python
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](<https://danplischke.github.io/apache-airflow-providers-lamindb/filters/index.md>) on a branch (`branch`, default `main`). The matching records are returned via XCom: at most `limit` records, newest first.

tests/system/lamindb/example\_lamindb\_sensors.py

```python
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](<https://docs.lamin.ai/rest>): `{"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](<https://danplischke.github.io/apache-airflow-providers-lamindb/filters/index.md>), and ignores artifacts whose upload to storage is still in progress.

tests/system/lamindb/example\_lamindb\_sensors.py

```python
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](<https://danplischke.github.io/apache-airflow-providers-lamindb/filters/index.md>)
- `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.

tests/system/lamindb/example\_lamindb\_sensors.py

```python
@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]}")
```