Airflow DAGs

databaas_utils.airflow mirrors the user helpers, but authenticates through the pipeline's Airflow connection (a Zitadel service account) instead of a person. Setting up that connection is covered in the Databaas documentation: Creating an Airflow service account.

from databaas_utils.airflow import get_catalog, get_index, get_store, scoped, target_for

t = target_for(__file__)

with DAG(t.dag_id("daily_summary"), tags=t.tags("weather"), is_paused_upon_creation=t.paused):
    ...

def load():
    catalog = get_catalog(namespace=scoped("open_meteo"))
    store = get_store()
    index = get_index()

Helpers

Each takes conn_id= (default databaas_lakehouse) and namespace=.

Function	Returns
`get_catalog()`	A pyiceberg catalog
`get_store()`	A blob store for generic tables
`get_index()`	The Iceberg file index
`get_zarr_group()`	A `zarr.Group`
`get_settings()`	`LakehouseSettings`

You can also use the hook, LakehouseHook, which has the same get_* methods.

A missing or misspelt connection fails the task. It never falls back to local storage. Pass local_root="/tmp/x" to get_store() or get_zarr_group() if you want the filesystem explicitly, for example when running a DAG on your laptop.

Dev deployments and branch-safe data

CI deploys a branch of a pipeline next to production with databaas dag deploy, with dev_<key>_ prefixed DAG ids. Without care, that branch would still write to the production tables. Two helpers make names deployment-aware:

target_for(__file__) : Reads the _databaas_target.json that CI wrote into the bundle. Gives you dag_id(name), tags(...), paused, is_production and doc_suffix().

scoped(name) : Returns name in production and name_dev_<key> on a branch.

table = scoped("daily_summary")   # "daily_summary" or "daily_summary_dev_add_login"

Call scoped() directly from the DAG file

It finds the deployment from the file that called it. Called from a helper module, it raises. Pass the DAG's path explicitly instead: scoped("daily_summary", caller_file=__file__).

from databaas_target import target still works for older DAGs, but is deprecated and removed in 2.0.

Testing a DAG locally

Set the connection in the environment, with the service account's JSON key. No Airflow database is needed for this route:

export AIRFLOW_CONN_DATABAAS_LAKEHOUSE="$(python -c '
import json, sys
print(json.dumps({
    "host": "https://catalog.<your-databaas-domain>",
    "extra": {
        "service_account_key": json.load(open(sys.argv[1])),
        "token_url": "https://auth.<your-databaas-domain>/oauth/v2/token",
        "scope": "openid urn:zitadel:iam:org:project:id:<project id>:aud",
    },
}))' key.json)"

The warehouse defaults to databaas-lakehouse; add "schema" for another one. Set scope rather than only project_id: a service account's token carries the project as audience only when the scope asks for it with :aud, and Trino rejects it otherwise.