Airflow

databaas_utils.airflow.helpers

One-line entry points for DAG code.

Every one takes conn_id= so a DAG needing different Lakekeeper permissions can point at a second service account, and namespace= so a pipeline can keep its tables somewhere other than the connection's default.

get_settings

get_settings(
    conn_id: str = DEFAULT_CONN_ID,
    *,
    namespace: str | None = None,
) -> LakehouseSettings

Settings for conn_id. Raises if it cannot be resolved.

Parameters:

Name	Type	Description	Default
`conn_id`	`str`	The Airflow connection to read.	`DEFAULT_CONN_ID`
`namespace`	`str \| None`	Overrides the connection's default namespace.	`None`

Raises:

Type	Description
`ConnectionNotConfigured`	If the connection does not exist.
`ConfigurationError`	If its fields are incomplete.

get_catalog

get_catalog(
    *,
    conn_id: str = DEFAULT_CONN_ID,
    namespace: str | None = None,
) -> Catalog

A pyiceberg REST catalog, signing through Lakekeeper.

Parameters:

Name	Type	Description	Default
`conn_id`	`str`	The Airflow connection to authenticate with.	`DEFAULT_CONN_ID`
`namespace`	`str \| None`	Overrides the connection's default namespace.	`None`

Example

catalog = get_catalog() catalog.load_table("open_meteo.daily_summary")

get_store

get_store(
    *,
    conn_id: str = DEFAULT_CONN_ID,
    namespace: str | None = None,
    local_root: str | None = None,
) -> BlobStore

Blob access under generic tables.

local_root is an explicit opt-in to the filesystem stand-in. It is never reached by falling back: a missing connection raises, because a silent local write that reports success is worse than a failed task.

Parameters:

Name	Type	Description	Default
`conn_id`	`str`	The Airflow connection to authenticate with.	`DEFAULT_CONN_ID`
`namespace`	`str \| None`	Overrides the connection's default namespace.	`None`
`local_root`	`str \| None`	Directory to use instead of the lakehouse (local runs).	`None`

Example

store = get_store() store.put("grib", "2026/09/30/run.grib2", Path("run.grib2"))

get_index

get_index(
    *,
    conn_id: str = DEFAULT_CONN_ID,
    namespace: str | None = None,
    table: str = "files",
) -> Index

The Iceberg index of what landed in the generic tables.

Parameters:

Name	Type	Description	Default
`conn_id`	`str`	The Airflow connection to authenticate with.	`DEFAULT_CONN_ID`
`namespace`	`str \| None`	Overrides the connection's default namespace.	`None`
`table`	`str`	Index table name.	`'files'`

get_zarr_group

get_zarr_group(
    table: str,
    *,
    conn_id: str = DEFAULT_CONN_ID,
    namespace: str | None = None,
    path: str = "",
    read_only: bool = False,
    create: bool = True,
    doc: str = "Zarr store",
    local_root: str | None = None,
) -> Group

A zarr group under a generic table.

local_root is an explicit opt-in to the filesystem, as with get_store.

Parameters:

Name	Type	Description	Default
`table`	`str`	Generic table holding the zarr store.	required
`conn_id`	`str`	The Airflow connection to authenticate with.	`DEFAULT_CONN_ID`
`namespace`	`str \| None`	Overrides the connection's default namespace.	`None`
`path`	`str`	Group path within the store.	`''`
`read_only`	`bool`	Open for reading only.	`False`
`create`	`bool`	Create the table and group if missing.	`True`
`doc`	`str`	Description recorded on a newly created table.	`'Zarr store'`
`local_root`	`str \| None`	Directory to use instead of the lakehouse (local runs).	`None`

Example

group = get_zarr_group("harmonie_zarr", path="2026093000")

get_lakehouse_settings

get_lakehouse_settings(
    *,
    conn_id: str = DEFAULT_CONN_ID,
    namespace: str | None = None,
) -> LakehouseSettings

Settings alone, for callers building something the helpers do not cover.

Parameters:

Name	Type	Description	Default
`conn_id`	`str`	The Airflow connection to authenticate with.	`DEFAULT_CONN_ID`
`namespace`	`str \| None`	Overrides the connection's default namespace.	`None`

databaas_utils.airflow.connection

Resolving the lakehouse connection a Databaas admin registered.

The admin flow this serves: create a Zitadel service user with a client_credentials machine client, grant it permissions on the Lakekeeper warehouse, and register it as an Airflow connection. No S3 keys appear anywhere in that list — object storage is reached by asking Lakekeeper to sign.

Three resolution routes are tried in order, matching what the image's MQTT trigger already does so there is one behaviour to learn:

AIRFLOW_CONN_<CONN_ID> in the environment — how the chart delivers connections (airflowConnections.items renders one Secret of these). This route is pure stdlib, so a container or notebook with the variable set needs no Airflow installed.
Connection.get_connection_from_secrets — the secrets-backend chain.
BaseHook.get_connection — the metadata database.

Failure is loud. The prior art caught every exception here and returned None, which the caller turned into a local filesystem sink: a typo'd conn_id or an expired secret then redirected a production write to an ephemeral pod disk and the task went green. That is the worst failure mode in the prior art and it is not reproduced.

DEFAULT_CONN_ID `module-attribute`

DEFAULT_CONN_ID = 'databaas_lakehouse'

ConnectionData `dataclass`

ConnectionData(
    conn_id: str,
    host: str | None = None,
    schema: str | None = None,
    login: str | None = None,
    password: str | None = None,
    port: int | None = None,
    extra: dict[str, Any] = dict(),
)

An Airflow connection, normalised.

Its own type rather than Airflow's Connection because the env route must work where Airflow is not installed.

from_airflow `classmethod`

from_airflow(conn: Any) -> ConnectionData

Normalise an Airflow Connection object.

resolve_connection

resolve_connection(
    conn_id: str = DEFAULT_CONN_ID,
) -> ConnectionData

The connection for conn_id, tried via env, secrets, then hook.

Raises rather than returning None: an unresolvable connection is a configuration problem, and every caller here would otherwise have to invent its own fallback.

Raises:

Type	Description
`ConnectionNotConfigured`	If no route finds the connection.

settings_from_connection

settings_from_connection(
    conn: ConnectionData, *, namespace: str | None = None
) -> LakehouseSettings

Map connection fields onto settings.

schema carries the warehouse: it is Airflow's "which database" slot, it renders as a labelled form field rather than a JSON key, and it is what every other data connection uses. extra.warehouse stays accepted for compatibility, and schema wins. With neither, it is databaas-lakehouse.

Setting up the connection is described at https://docs.databaas.eu/pipelines/service-account.

Raises:

Type	Description
`ConfigurationError`	If the host or token settings are missing.

get_settings

get_settings(
    conn_id: str = DEFAULT_CONN_ID,
    *,
    namespace: str | None = None,
) -> LakehouseSettings

Settings for conn_id. Raises if it cannot be resolved.

Parameters:

Name	Type	Description	Default
`conn_id`	`str`	The Airflow connection to read.	`DEFAULT_CONN_ID`
`namespace`	`str \| None`	Overrides the connection's default namespace.	`None`

Raises:

Type	Description
`ConnectionNotConfigured`	If the connection does not exist.
`ConfigurationError`	If its fields are incomplete.

databaas_utils.airflow.hook

The Airflow hook for the lakehouse connection.

Its real value is test_connection: one click at configuration time tells an admin whether the client secret, token URL, project scope and warehouse name are all correct together. Without it the first signal is a DAG failure hours later, with an error that does not mention the connection.

LakehouseHook

LakehouseHook(
    conn_id: str = DEFAULT_CONN_ID, **kwargs: Any
)

Bases: BaseHook

Access to a Lakekeeper warehouse, as one Zitadel service account.

Parameters:

Name	Type	Description	Default
`conn_id`	`str`	The Airflow connection to authenticate with.	`DEFAULT_CONN_ID`

Example

hook = LakehouseHook() catalog = hook.get_catalog(namespace="open_meteo")

Bind the hook to conn_id.

get_ui_field_behaviour `staticmethod`

get_ui_field_behaviour() -> dict[str, Any]

Field labels and hints for the connection form.

get_settings

get_settings(
    *, namespace: str | None = None
) -> LakehouseSettings

The connection's :class:~databaas_utils.LakehouseSettings.

get_catalog

get_catalog(*, namespace: str | None = None) -> Any

A pyiceberg REST catalog for this connection.

get_store

get_store(*, namespace: str | None = None) -> BlobStore

A remote-signed blob store for this connection.

get_index

get_index(
    *, namespace: str | None = None, table: str = "files"
) -> Index

The Iceberg file index for this connection.

test_connection

test_connection() -> tuple[bool, str]

Fetch a token and read the warehouse config.

Exercises the whole chain an admin can get wrong — client secret, token URL, project scope, host, warehouse name — and reports which link failed rather than just that something did.

databaas_utils.airflow.scoping

Deployment-scoped names, resolved from the calling DAG file.

The rule this module encodes: the caller-frame trick is only sound in a function called directly from the DAG file. sys._getframe(1) inside scoped() gives the DAG; inside a helper that scoped() calls, it gives the helper. Every public function here therefore takes a caller_file= override so a wrapper can pass __file__ through instead of silently resolving itself.

The prior art violated this: lakehouse_core.deploy.slug() read databaas_target.target.mode, so the lookup resolved against deploy.py's own directory. It worked only because deploy.py shipped inside the bundle, where the parent walk still found the bundle's config. From site-packages the same call raises at DAG parse time, in production.

current_slug

current_slug(
    *, caller_file: str | None = None
) -> str | None

The branch slug for the calling DAG file; None in production.

MUST be called directly from the DAG file, or given caller_file=__file__.

Parameters:

Name	Type	Description	Default
`caller_file`	`str \| None`	The DAG file's `__file__`, if calling from a helper.	`None`

scoped

scoped(name: str, *, caller_file: str | None = None) -> str

name in production, name_dev_<slug> on a branch deployment.

MUST be called directly from the DAG file, or given caller_file=__file__.

Parameters:

Name	Type	Description	Default
`name`	`str`	The production name, e.g. a table or namespace.	required
`caller_file`	`str \| None`	The DAG file's `__file__`, if calling from a helper.	`None`

Example

table = scoped("daily_summary") # in a DAG file 'daily_summary_dev_add_login'

databaas_utils.airflow.target

Which deployment a DAG file belongs to — production, or one branch of it.

CI writes a _databaas_target.json into every deployed bundle. This module resolves that file from the DAG file using it, not from its own location, which is what lets one shared copy serve every bundle correctly.

That matters because DataBaasS3DagBundle registers every bundle directory on one process-wide sys.path and the dag-processor parses all bundles in one process. Two bundles each shipping their own databaas_target.py compete for a single global module name: a dev bundle's copy could win the import for a prod DAG file, registering it as dev_<slug>_<name>. The real prod dag id was then never produced, Airflow marked it stale, and it vanished from the DAG list while its running DagRuns stayed visible.

Shipping here — in site-packages, via a pinned wheel — is safer than the previous /opt/databaas placement rather than riskier, because _register_sys_path_recursive appends bundle directories. site-packages precedes every appended entry, so no bundle can shadow this module. Do not change that append back to insert(0).

DeploymentTarget

DeploymentTarget(data: dict[str, Any])

One deployment's identity, as CI recorded it.

Attributes:

Name	Type	Description
`mode`		`"production"` or a development mode.
`dev_key`		The branch key for a development deployment.
`bundle`		The bundle name.
`paused`		Whether DAGs from this deployment should start paused.
`source_repo`		Repository the bundle was deployed from.
`source_ref`		Git ref it was deployed from.
`source_commit`		Commit SHA.
`source_run_url`		URL of the CI run that deployed it.
`deployed_at`		Deployment timestamp.

Build from the parsed _databaas_target.json.

is_production `property`

is_production: bool

Whether this is the production deployment.

dag_id

dag_id(name: str) -> str

name in production, dev_<dev_key>_<name> otherwise.

doc_suffix

doc_suffix() -> str

Markdown "Deployed from" footer for doc_md; empty if unknown.

target_for

target_for(caller_file: str | None) -> DeploymentTarget

Resolve the deployment target for a DAG file.

Call as target_for(__file__) from the DAG itself. Library code must pass the DAG's path through explicitly — resolving it from a library module's own location is the bug this signature exists to make impossible.

Parameters:

Name	Type	Description	Default
`caller_file`	`str \| None`	The DAG file's `__file__`.	required

Raises:

Type	Description
`RuntimeError`	If `caller_file` is empty or no `_databaas_target.json` is found in its directory or parents.

Example

from databaas_utils.airflow import target_for t = target_for(file) with DAG(t.dag_id("daily_summary"), tags=t.tags("weather")): ...

Airflow

databaas_utils.airflow.helpers

get_settings

get_catalog

get_store

get_index

get_zarr_group

get_lakehouse_settings

databaas_utils.airflow.connection

DEFAULT_CONN_ID module-attribute

ConnectionData dataclass

from_airflow classmethod

resolve_connection

settings_from_connection

get_settings

databaas_utils.airflow.hook

LakehouseHook

get_ui_field_behaviour staticmethod

get_settings

get_catalog

get_store

get_index

test_connection

databaas_utils.airflow.scoping

current_slug

scoped

databaas_utils.airflow.target

DeploymentTarget

is_production property

dag_id

tags

doc_suffix

target_for

DEFAULT_CONN_ID `module-attribute`

ConnectionData `dataclass`

from_airflow `classmethod`

get_ui_field_behaviour `staticmethod`

is_production `property`