Authoring pipelines

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. If you only operate pipelines, monitoring and running 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 table, queryable from notebooks, SQL Lab and dashboards as soon as the load finishes.

"""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 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 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.

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.

Iceberg tables: get_catalog()

get_catalog() returns a standard pyiceberg RestCatalog. Everything in the pyiceberg docs 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: a named, permissioned location in the warehouse.

@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)

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).

@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.

@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.

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

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.

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.

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 lists every installed package with its exact version.

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.

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 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:

• 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.