# Authoring pipelines (/pipelines/authoring)

Write an Airflow 3 DAG that reads and writes lakehouse data through databaas-utils.

This page covers writing pipeline code: the shape of a DAG, the `databaas-utils` helpers that reach
the catalog and object storage, and which libraries you can use. To ship the code, see
[develop and ship a pipeline](/pipelines/development-workflow). If you only operate pipelines,
[monitoring and running](/pipelines/monitoring) is the page you want to look at.

## A pipeline is Python

A pipeline is an Airflow 3 DAG written with the TaskFlow API: a function decorated with `@dag` that
calls `@task` functions. Each task does one step; passing one task's return value into another sets
the order they run in. Databaas runs Airflow 3.1, so import from `airflow.sdk`, not the Airflow 2
`airflow.decorators` path.

The DAG below fetches rows from an API and writes them into an [Iceberg](/catalog/iceberg) table,
queryable from notebooks, SQL Lab and dashboards as soon as the load finishes.

```python title="dags/open_meteo_forecast.py"
"""Daily forecast summary, written to Iceberg."""

from datetime import timedelta

import pendulum
from airflow.sdk import dag, task
from databaas_utils.airflow import scoped
from databaas_utils.airflow.target import target

NAMESPACE = scoped("open_meteo")


@dag(
    dag_id=target.dag_id("open_meteo_forecast"),
    schedule="0 7 * * *",
    start_date=pendulum.datetime(2026, 9, 1, tz="UTC"),
    catchup=False,
    max_active_runs=1,
    is_paused_upon_creation=target.paused,
    tags=target.tags("open_meteo"),
    doc_md=__doc__ + target.doc_suffix(),
    default_args={"retries": 2, "retry_delay": timedelta(minutes=2)},
)
def open_meteo_forecast():
    @task
    def extract() -> list[dict]:
        import requests

        response = requests.get("https://api.example.com/forecast", timeout=30)
        response.raise_for_status()
        return response.json()["rows"]

    @task
    def load(rows: list[dict], ds: str | None = None) -> int:
        import pyarrow as pa
        from pyiceberg.expressions import EqualTo
        from databaas_utils.airflow import get_catalog

        catalog = get_catalog()
        catalog.create_namespace_if_not_exists(NAMESPACE)

        batch = pa.Table.from_pylist([{**row, "run_date": ds} for row in rows])
        table = catalog.create_table_if_not_exists(
            f"{NAMESPACE}.daily_summary", schema=batch.schema
        )
        table.overwrite(batch, overwrite_filter=EqualTo("run_date", ds))
        return batch.num_rows

    load(extract())


open_meteo_forecast()
```

- `dag_id` is the name shown in the Airflow UI. `schedule` decides when it runs on its own; `None`
  means trigger-only.
- `target` and `scoped()` make the same file safe to deploy both as production and as a development
  copy for a branch. See [deployment identity](#deployment-identity-target) below.
- `load` overwrites the slice of rows its run owns instead of appending, so a retried task replaces
  its own rows rather than doubling them.

See the Apache Airflow [TaskFlow documentation](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/taskflow.html)
for the full model.

## What databaas-utils gives you

`databaas-utils` is preinstalled wherever your pipelines run. It reaches the lakehouse as your
pipeline's own identity through an Airflow connection, `databaas_lakehouse`, which is set up for
you. Your DAG code never holds a credential or a storage key.

Import everything from `databaas_utils.airflow`, except `target` (from
`databaas_utils.airflow.target`) and `FileRecord` (from `databaas_utils`). The full API is in the
[databaas-utils documentation](https://wolkwork.github.io/databaas-utils/).

| You want | Call |
| -------- | ---- |
| The Iceberg catalog (pyiceberg) | `get_catalog()` |
| Files under a generic table | `get_store(namespace=...)` |
| A Zarr group under a generic table | `get_zarr_group("<table>", namespace=...)` |
| An Iceberg record of which files landed | `get_index(namespace=...)` |
| A data name a branch deploy cannot collide on | `scoped("<name>")` |
| This deployment's DAG id, tags and pause state | `target` |
| All of the above from one explicit connection | `LakehouseHook(conn_id=...)` |

Every helper takes `conn_id=` (default `databaas_lakehouse`) if your pipeline uses a different
connection. `get_store`, `get_zarr_group` and `get_index` also take `namespace=`, which overrides the
connection's default namespace.

<Callout type="warn">
  Do not `pip install` databaas-utils or add it to your pipeline's requirements. The preinstalled
  version is the one your DAGs get; a second copy conflicts with it.
</Callout>

### Iceberg tables: `get_catalog()`

`get_catalog()` returns a standard pyiceberg `RestCatalog`. Everything in the
[pyiceberg docs](https://py.iceberg.apache.org/api/) applies.

- Always use fully qualified identifiers, `"<namespace>.<table>"`. The connection's default
  namespace does not apply to pyiceberg calls.
- Overwrite the slice a run owns rather than `append`, and keep the filter on one value per run.
- Reading works the same way:
  `catalog.load_table("open_meteo.daily_summary").scan(row_filter=...).to_arrow()`.

### Files: `get_store()`

Raw files that are not tables yet, such as a GRIB download, a PDF or a CSV drop, go into a
[generic table](/getting-started/concepts): a named, permissioned location in the warehouse.

```python title="Store a downloaded file"
@task
def fetch_and_store(run_date: str) -> str:
    from pathlib import Path
    from databaas_utils.airflow import get_store

    local = Path("/tmp") / f"{run_date}_f012.grib"
    ...  # download from the source into `local`

    store = get_store(namespace=NAMESPACE)
    return store.put("harmonie_raw", f"{run_date}/f012.grib", local)
```

| Method | Does |
| ------ | ---- |
| `put(table, key, source)` | Uploads a local file and returns its `s3://` URI. Creates the generic table on first use. |
| `get(table, key, dest)` | Streams one object to a local path. |
| `open(table, key)` | Returns a readable file object, without staging to disk. |
| `list(table, prefix="")` | Lists object metadata under the table. |
| `delete(table, key)` | Removes one object. |
| `uri(table, key)` | Returns the `s3://` URI an object has or would have. |

Only `put` creates the table. The other methods never do, so a misspelled table name fails instead
of silently creating an empty table.

Pass the returned URI between tasks, not the bytes. A task's return value is stored by Airflow and
is fine for a few hundred rows, not for a few hundred megabytes.

### Zarr: `get_zarr_group()`

A Zarr store also lives under a generic table. You get a normal `zarr.Group` (Zarr 3).

```python title="Write a Zarr array"
@task
def write_grid(run_date: str) -> None:
    from databaas_utils.airflow import get_zarr_group

    root = get_zarr_group("harmonie_zarr", namespace=NAMESPACE)
    run = root.require_group(run_date)
    t2m = run.create_array(
        "t2m", shape=(49, 300, 300), chunks=(1, 300, 300), dtype="float32", overwrite=True
    )
    t2m[:] = ...  # your numpy data
```

Pass `read_only=True` in a task that only reads; a read-only open never creates the table. Pass
`path="<subgroup>"` to open a group below the root directly, and `create=False` to make a missing
table an error for writers too.

### Record what landed: `get_index()`

A generic table is a location, not a file list: neither the catalog nor any UI can say which files
landed in it. The index is an ordinary Iceberg table with one row per stored object, so you can query
it from SQL Lab, notebooks and dashboards.

```python title="Record a stored file"
@task
def record(uri: str, key: str, size: int, run_date: str) -> None:
    from databaas_utils import FileRecord
    from databaas_utils.airflow import get_index

    get_index(namespace=NAMESPACE).record(
        FileRecord(uri=uri, key=key, size_bytes=size, product="harmonie", run_date=run_date)
    )
```

- The index table is `<namespace>.files` by default, created on first use. Pass `table=` for
  another name.
- Only `uri`, `key` and `size_bytes` are required. `product`, `run_date`, `cycle` and
  `forecast_hour` are optional, and `attributes` takes a `dict[str, str]` for anything else.
- `record()` takes any number of records and commits them as one append.
- `latest()` returns the newest `run_date` and `cycle` that actually landed, and `scan()` returns
  the whole index as an Arrow table. `latest()` returns `None` when there is no index yet or no row
  carries a run, and raises when the index exists but cannot be read. It scans the whole table, so
  call it once per run, not once per file.

### Deployment identity: `target`

The same DAG file runs as production once merged, and as a development copy for every open pull
request, in the same Airflow. `target` tells the two apart.

| Attribute | Production | Branch deploy |
| --------- | ---------- | ------------- |
| `target.dag_id("my_dag")` | `my_dag` | `dev_<key>_my_dag` |
| `target.paused` | `False` | `True`: development DAGs are created paused |
| `target.tags("a", "b")` | `["a", "b"]` | `["a", "b", "dev:<key>"]` |
| `target.is_production` | `True` | `False` |

`target.doc_suffix()` returns where this copy was deployed from, ready to append to `doc_md`.

<Callout type="warn">
  Import it as `from databaas_utils.airflow.target import target`. `from databaas_utils.airflow import target`
  gives you the module of the same name, and the first `target.dag_id(...)` fails with
  `AttributeError: module ... has no attribute 'dag_id'`.
</Callout>

### Keep a branch off production data: `scoped()`

Prefixed DAG ids stop two DAGs colliding. They do not stop a branch's DAG writing into a production
namespace. Pass every name that identifies data through `scoped()`: namespaces, tables in a shared
namespace, asset URIs, cursors and state keys.

```python
from databaas_utils.airflow import scoped

NAMESPACE = scoped("open_meteo")  # "open_meteo" in production, "open_meteo_dev_<key>" on a branch
```

- Call it from the DAG file, at the top or in a task defined there. A helper in your own package
  that must call it needs the DAG's path passed through: `scoped("open_meteo", caller_file=__file__)`.
- Outside a deployment, such as a test on your laptop, it returns the name unchanged.
- Use `databaas_utils.airflow.scoped`, not the top-level `databaas_utils.scoped`. The top-level one
  can leak a branch's suffix into production names.

### One connection, explicitly: `LakehouseHook`

`LakehouseHook` bundles the same entry points behind one connection, and `test_connection()` checks
that the connection can reach the catalog and warehouse.

```python
from databaas_utils.airflow import LakehouseHook

hook = LakehouseHook(conn_id="databaas_lakehouse")
catalog = hook.get_catalog()
store = hook.get_store(namespace=NAMESPACE)
index = hook.get_index(namespace=NAMESPACE)
```

## Rules

- **Import lakehouse helpers inside the task body**, not at the top of the DAG file. Airflow imports
  every DAG file just to parse it, and `get_catalog`, `get_store` and the rest pull in heavy
  libraries. `scoped` and `target` are the exception: they are light and belong at module level.
- **Fail loudly.** A missing or misspelled connection raises `ConnectionNotConfigured` rather than
  writing somewhere else. Do not catch it to keep a task green.
- **Never put storage keys in code.** Nothing in this flow needs an access key. If you find yourself
  reaching for one, the connection is set up wrong; ask your Databaas team.
- **Do not install the upstream `obstore` package.** The preinstalled `obstore-databaas` shares its
  module name, and adding `obstore` to a task's requirements replaces it and breaks storage access
  with `RemoteSigningUnavailable`.
- **Keep tasks idempotent.** A retry or a cleared task runs again; it must not double the rows.
- **Import timetables from `airflow.timetables`**, not `airflow.sdk`. `from airflow.sdk import
  CronTriggerTimetable` fails on the Airflow version Databaas runs.

## Libraries you can use

These are preinstalled and importable from any task. Import them inside the task body unless the DAG
needs them to parse.
[Python packages](/pipelines/python-packages) lists every installed package with its exact version.

| Library | Use it for |
| ------- | ---------- |
| `databaas-utils` | Catalog, files, Zarr, index, deployment identity. |
| `pyiceberg`, `pyarrow` | Reading and writing Iceberg tables. |
| `duckdb` | SQL over Arrow tables, parquet files and Iceberg tables. |
| `zarr` | Arrays; open them through `get_zarr_group()`. |
| `dlt` | Extract and load from APIs. See [dlt and dbt](/pipelines/dlt-and-dbt). |
| `dbt-core`, `dbt-trino` | SQL transformations. See [dlt and dbt](/pipelines/dlt-and-dbt). |
| `requests` | HTTP calls. |

`pandas` is not included. Use DuckDB or pyarrow instead, or install it for one task as below.

## Add a package that is not preinstalled

For a dependency specific to one pipeline, swap `@task` for `@task.virtualenv`. Airflow builds a
virtual environment for the task, installs the requirements into it, and runs the function there.

```python title="A task with its own dependency"
from datetime import timedelta

from airflow.sdk import task


@task.virtualenv(
    requirements=["example-package==1.4.0"],
    system_site_packages=True,
    venv_cache_path="/tmp/venvs",
    execution_timeout=timedelta(minutes=90),
)
def download(plan: dict) -> list[str]:
    import example_package

    ...
```

- **Pin every version.** An unpinned requirement rebuilds the environment and can break the pipeline
  on a day nothing changed.
- **Keep `system_site_packages=True`**, so the preinstalled libraries stay visible and you list only
  what is missing.
- **`venv_cache_path`** reuses the environment across runs instead of installing on every run.
- **Do not list anything that is already preinstalled**, and never `databaas-utils` or `obstore`.
- **The body closes over nothing.** It runs in a fresh interpreter: every import goes inside the
  function, module-level constants must be inlined or passed as arguments, and only the arguments and
  the return value cross the boundary. To import your own package, pass its directory in and add it
  to `sys.path` inside the function.
- **Private index**: `index_urls` and `pip_install_options` point pip at a private package index.

A package imported at the top of a DAG file must be available when Airflow parses the file, so a
virtualenv task cannot supply it. Move the import into the task body. If a dependency must be
available everywhere, ask your Databaas team.

For system libraries, command-line tools or a large set of packages, run the task in a
[custom image](/pipelines/custom-images) instead.

## Which identity a pipeline runs as

A pipeline does not run as you. It reads and writes the lakehouse as its own application identity,
through the `databaas_lakehouse` connection, and your Databaas team creates that identity for you. In
the Data Catalog's permissions picker it shows up as an **Application**. Whoever may grant on the
data grants it access like any other principal — see [data permissions](/catalog/data-permissions):

- **Select** to read, and **Modify** and **Create** to write, on the pipeline's own namespace.
- A create grant on the warehouse if the pipeline creates namespaces. `scoped()` makes every branch
  deploy create its own namespace, so a pipeline developed on branches needs this.

Grant on the pipeline's own namespace rather than the whole warehouse, so a bad overwrite cannot
reach another team's tables. A `403` on a table or namespace, or a `SignError` on a file, means the
identity has no grant there.