Storage

Files

databaas_utils.generic

Blob storage under a Lakekeeper generic table, without S3 credentials.

Every request is signed by Lakekeeper's own signer: the client asks the catalog to sign a method/URI/header triple and then issues the signed request itself, so no long-lived S3 key ever reaches the task.

The credentialed fallback from the prior art is deliberately absent. A fallback that quietly works without signing is how a deployment ends up on it without anyone deciding to.

BlobStore

Bases: Protocol

Opaque bytes under a named table.

Implemented by :class:GenericTableStore (remote-signed storage) and :class:LocalStore (a directory), so code written against this runs on either.

put

put(table: str, key: str, source: Path) -> str

Store the local file source as key; return its URI.

get

get(table: str, key: str, dest: Path) -> Path

Download key to dest and return dest.

uri

uri(table: str, key: str) -> str

The URI key has (or would have) under table.

GenericTableStore `dataclass`

GenericTableStore(
    settings: LakehouseSettings,
    client: CatalogClient | None = None,
    _tables: dict[str, ResolvedTable] = dict(),
    _stores: dict[tuple[str, str], Any] = dict(),
    _lock: Lock = threading.Lock(),
)

Bases: BlobStore

Blob access under generic tables, signed rather than credentialed.

Caches are instance-scoped on purpose. settings varies by namespace across call sites, so a module-level cache keyed only on table name would collide across namespaces and branch deploys.

Needs the obstore extra. Tables are created on first write.

Parameters:

Name	Type	Description	Default
`settings`	`LakehouseSettings`	Which warehouse and namespace, as which identity.	required
`client`	`CatalogClient \| None`	A `CatalogClient` to reuse.	`None`

Example

store = GenericTableStore(settings) store.put("grib", "2026/09/30/run.grib2", Path("run.grib2")) 's3://.../2026/09/30/run.grib2' for meta in store.list("grib", prefix="2026/09"): ... print(meta["path"], meta["size"])

table

table(
    table: str,
    *,
    create: bool = True,
    doc: str = "Generic table",
) -> ResolvedTable

The table's location and signing details. Cached per instance.

object_store

object_store(
    table: str, *, prefix: str = "", create: bool = True
) -> RemoteSignedS3Store

The raw obstore handle, for callers that need more than put/get.

Parameters:

Name	Type	Description	Default
`table`	`str`	Generic table name.	required
`prefix`	`str`	Restrict the handle to keys under this prefix.	`''`
`create`	`bool`	Create the table if missing.	`True`

uri

uri(table: str, key: str) -> str

The s3:// URI of key under table.

put

put(table: str, key: str, source: Path) -> str

Upload a local file under the table; returns its s3:// URI.

Parameters:

Name	Type	Description	Default
`table`	`str`	Generic table name; created if missing.	required
`key`	`str`	Object key within the table.	required
`source`	`Path`	Local file to upload.	required

get

get(table: str, key: str, dest: Path) -> Path

Stream one object under the table to dest.

Parameters:

Name	Type	Description	Default
`table`	`str`	Generic table name; it must already exist.	required
`key`	`str`	Object key within the table.	required
`dest`	`Path`	Local path to write; parent directories are created.	required

Returns:

Type	Description
`Path`	`dest`.

open

open(table: str, key: str) -> Any

A readable file object, without staging to disk first.

Parameters:

Name	Type	Description	Default
`table`	`str`	Generic table name; it must already exist.	required
`key`	`str`	Object key within the table.	required

list

list(table: str, *, prefix: str = '') -> Iterator[Any]

Object metadata under the table.

Lakekeeper records only a location for a generic table, so listing is the only way to learn what actually landed.

Parameters:

Name	Type	Description	Default
`table`	`str`	Generic table name; it must already exist.	required
`prefix`	`str`	Only list keys under this prefix.	`''`

Yields:

Type	Description
`Any`	obstore object-metadata dicts (`path`, `size`, `last_modified`, ...).

delete

delete(table: str, key: str) -> None

Delete one object under the table.

LocalStore `dataclass`

LocalStore(root: Path)

Bases: BlobStore

A filesystem stand-in for laptop runs and tests. Never a silent fallback.

Files live at <root>/<table>/<key>.

Parameters:

Name	Type	Description	Default
`root`	`Path`	Directory that plays the role of the warehouse.	required

uri

uri(table: str, key: str) -> str

The filesystem path of key under table.

put

put(table: str, key: str, source: Path) -> str

Copy source to <root>/<table>/<key>; return that path.

get

get(table: str, key: str, dest: Path) -> Path

Copy <root>/<table>/<key> to dest and return dest.

ResolvedTable `dataclass`

ResolvedTable(
    location: str,
    signer_uri: str,
    region: str,
    endpoint: str,
)

A generic table's location plus what is needed to sign requests under it.

Attributes:

Name	Type	Description
`location`	`str`	The table's `s3://` base URI.
`signer_uri`	`str`	The full URL of Lakekeeper's signing endpoint.
`region`	`str`	Storage region reported by Lakekeeper.
`endpoint`	`str`	S3 endpoint, or `""` for AWS proper.

resolve

resolve(
    settings: LakehouseSettings,
    table: str,
    *,
    create: bool = True,
    doc: str = "Generic table",
    client: CatalogClient | None = None,
) -> ResolvedTable

Look up settings.namespace.table, creating it if asked.

Parameters:

Name	Type	Description	Default
`settings`	`LakehouseSettings`	Which warehouse and namespace, as which identity.	required
`table`	`str`	Generic table name.	required
`create`	`bool`	Create the table (and namespace) if missing. With `False` a missing table raises.	`True`
`doc`	`str`	Description recorded on a newly created table.	`'Generic table'`
`client`	`CatalogClient \| None`	A `CatalogClient` to reuse.	`None`

Returns:

Type	Description
`ResolvedTable`	The table's location and signing details.

Raises:

Type	Description
`NotFoundError`	The table is missing and `create` is false.
`LakehouseError`	Lakekeeper did not advertise a signer for the table.

signed_store

signed_store(
    settings: LakehouseSettings,
    resolved: ResolvedTable,
    *,
    prefix: str = "",
    session: Session | None = None,
    client_options: Any = None,
) -> RemoteSignedS3Store

An obstore store over resolved, signing each request as it goes out.

Parameters:

Name	Type	Description	Default
`settings`	`LakehouseSettings`	Supplies the bearer used to call Lakekeeper's signer.	required
`resolved`	`ResolvedTable`	The table to open, from :func:`resolve`.	required
`prefix`	`str`	Restrict the store to keys under this prefix.	`''`
`session`	`Session \| None`	Pin all signing calls to one session (tests only).	`None`
`client_options`	`Any`	Extra obstore client options.	`None`

Raises:

Type	Description
`RemoteSigningUnavailable`	If `obstore-databaas` is not installed.

remote_signing_available

remote_signing_available() -> bool

Whether this environment can sign generic-table requests.

A capability probe, not a version check: it asks whether the class and method exist rather than which distribution or release provided them.

Zarr

databaas_utils.zarr

Zarr groups backed by a Lakekeeper generic table.

Iceberg models tables; numerical weather output is multidimensional gridded arrays, so the payload itself is written as zarr into a generic table and the Iceberg side only records what landed (see databaas_utils.index).

This is the same signed object store GenericTableStore uses — zarr just gets a zarr.storage.ObjectStore wrapper around it instead of put/get. It lives in its own module because zarr is the heaviest optional dependency here and nothing else needs it.

Install plain zarr, never zarr[remote]. That extra depends on upstream obstore, which shares a top-level module name with obstore-databaas and would silently overwrite it — taking RemoteSignedS3Store, and every signed write, with it. The object store is supplied here, so the extra buys nothing.

open_zarr_group

open_zarr_group(
    settings: LakehouseSettings,
    table: str,
    *,
    path: str = "",
    read_only: bool = False,
    create: bool = True,
    doc: str = "Zarr store",
    client: CatalogClient | None = None,
) -> Group

Open a zarr group stored under settings.namespace.table.

Reads and writes are remote-signed by Lakekeeper; no S3 credentials are involved. create is ignored when read_only is set — opening a reader should never create a table as a side effect. Needs the zarr extra.

Parameters:

Name	Type	Description	Default
`settings`	`LakehouseSettings`	Which warehouse and namespace, as which identity.	required
`table`	`str`	Generic table holding the zarr store; created if missing when `create` is true.	required
`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'`
`client`	`CatalogClient \| None`	A `CatalogClient` to reuse.	`None`

Returns:

Type	Description
`Group`	A `zarr.Group`.

Example

group = open_zarr_group(settings, "harmonie_zarr", read_only=True) list(group.keys())

local_zarr_group

local_zarr_group(
    root: str | Path,
    table: str,
    *,
    path: str = "",
    read_only: bool = False,
    create: bool = True,
) -> Group

A zarr group on the local filesystem, for laptop runs and tests.

Explicit, never a fallback: a missing catalog connection raises rather than silently redirecting a production write to an ephemeral pod disk.

Parameters:

Name	Type	Description	Default
`root`	`str \| Path`	Directory that plays the role of the warehouse.	required
`table`	`str`	Subdirectory holding the zarr store.	required
`path`	`str`	Group path within the store.	`''`
`read_only`	`bool`	Open for reading only.	`False`
`create`	`bool`	Create the directory and group if missing.	`True`

File index

databaas_utils.index

An Iceberg index of what actually landed in the generic tables.

Lakekeeper records only a location for a generic table — no file list — so neither the catalog nor the console can say which objects exist under one. This table is that record.

The schema deliberately carries no source-format knowledge: product, run_date, cycle and forecast_hour are all optional, and anything else goes in attributes. The prior art required product and forecast_hour, which is numerical-weather-prediction knowledge that a package indexing invoices should not have to satisfy.

Index

Bases: Protocol

Something that records ingested files and knows the newest run.

Implemented by :class:FileIndex and :class:NullIndex.

record

record(*records: FileRecord) -> None

Record that these files landed.

latest

latest() -> str | None

The newest run_date + cycle recorded, or None.

FileIndex

FileIndex(
    settings: LakehouseSettings,
    table: str = DEFAULT_INDEX_TABLE,
    *,
    catalog: Any = None,
    client: CatalogClient | None = None,
)

Bases: Index

Appends rows to <namespace>.<table>, creating the table on first use.

Needs the iceberg extra.

Parameters:

Name	Type	Description	Default
`settings`	`LakehouseSettings`	Which warehouse and namespace, as which identity.	required
`table`	`str`	Index table name.	`DEFAULT_INDEX_TABLE`
`catalog`	`Any`	A pyiceberg catalog to use instead of building one.	`None`
`client`	`CatalogClient \| None`	A `CatalogClient` to reuse.	`None`

Example

index = FileIndex(settings) index.record(FileRecord(uri=uri, key="run.grib2", size_bytes=1024)) index.latest() '2026093000'

identifier `property`

identifier: str

The Iceberg identifier, <namespace>.<table>.

catalog `property`

catalog: Any

The pyiceberg catalog, built on first use.

record

record(*records: FileRecord) -> None

Append records as one commit, retrying if a parallel task wins.

Raises:

Type	Description
`CommitFailedException`	After five conflicting attempts.

latest

latest() -> str | None

The newest run_date || cycle, or None if there is no index yet.

Three outcomes, kept distinct on purpose: no table means nothing was ever ingested (None); a table with no run identity means the same (None); a table that exists but cannot be read is a real failure and raises — it must not look like "nothing ingested" to a caller deciding whether to re-ingest.

scan

scan(**kwargs: Any) -> pa.Table

The index as an Arrow table. Raises if the table does not exist.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Passed to pyiceberg's `Table.scan` (`row_filter`, `selected_fields`, `limit`, ...).	`{}`

NullIndex `dataclass`

NullIndex()

Bases: Index

No catalog to index into (laptop runs) — just say what would happen.

record

record(*records: FileRecord) -> None

Log what would have been indexed.

latest

latest() -> str | None

Always None.

FileRecord `dataclass`

FileRecord(
    uri: str,
    key: str,
    size_bytes: int,
    product: str | None = None,
    run_date: str | None = None,
    cycle: str | None = None,
    forecast_hour: int | None = None,
    attributes: dict[str, str] = dict(),
    ingested_at: datetime = (lambda: datetime.now(UTC))(),
)

One stored object, as the index describes it.

Parameters:

Name	Type	Description	Default
`uri`	`str`	The object's full URI.	required
`key`	`str`	Its key within the generic table.	required
`size_bytes`	`int`	Size in bytes.	required
`product`	`str \| None`	Optional product/dataset name.	`None`
`run_date`	`str \| None`	Optional run date, `YYYYMMDD`.	`None`
`cycle`	`str \| None`	Optional cycle within the run date, e.g. `"00"`.	`None`
`forecast_hour`	`int \| None`	Optional forecast lead time in hours.	`None`
`attributes`	`dict[str, str]`	Free-form string metadata.	`dict()`
`ingested_at`	`datetime`	When it landed; defaults to now (UTC).	`(lambda: now(UTC))()`

as_row

as_row() -> dict[str, Any]

The record as a column-name to value mapping, matching :func:arrow_schema.

Deployment scoping

databaas_utils.scoping

Naming a thing per deployment, so a branch run cannot write production data.

CI prefixes dev dag_ids so two deployments coexist in one Airflow. That is not enough on its own: without scoping the data too, a branch run writes the production tables and moves the production cursor.

This module is deliberately Airflow-free and knows nothing about databaas_target. The dependency runs the other way — databaas_utils.airflow. scoping resolves the slug from the calling DAG file and passes it in. That inversion is what lets this code live in site-packages at all: the prior art called databaas_target from library code, and its caller-frame lookup then resolved against the library's own directory rather than the DAG's.

scoped

scoped(
    name: str, *, slug: str | _Unset | None = UNSET
) -> str

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

Suffixed rather than prefixed so the production name stays the sort-order prefix of its branches, and so a bare name is always the real one.

In a DAG use databaas_utils.airflow.scoped, which finds the slug from the deployment for you.

Parameters:

Name	Type	Description	Default
`name`	`str`	The production name, e.g. a table or namespace.	required
`slug`	`str \| _Unset \| None`	Override the slug; `None` forces the production name.	`UNSET`

Example

set_default_slug("add-login") scoped("daily_summary") 'daily_summary_dev_add_login'

set_default_slug

set_default_slug(value: str | None) -> None

Set the process-wide slug.

For containers and notebooks, where there is one deployment per process by construction. Not for Airflow: the dag-processor parses production and dev bundles in one process, so a process-wide slug would leak a branch's suffix into production table names — the data-layer twin of the dag-id incident that databaas_target exists to prevent.

Parameters:

Name	Type	Description	Default
`value`	`str \| None`	The slug, or `None` to mean production.	required

normalise

normalise(key: str) -> str

Lower-case key and squash anything unsafe to single underscores.

normalise("Feature/Add-Login") 'feature_add_login'

Storage

Files

databaas_utils.generic

BlobStore

put

get

uri

GenericTableStore dataclass

table

object_store

uri

put

get

open

list

delete

LocalStore dataclass

uri

put

get

ResolvedTable dataclass

resolve

signed_store

remote_signing_available

Zarr

databaas_utils.zarr

open_zarr_group

local_zarr_group

File index

databaas_utils.index

Index

record

latest

FileIndex

identifier property

catalog property

record

latest

scan

NullIndex dataclass

record

latest

FileRecord dataclass

as_row

Deployment scoping

databaas_utils.scoping

scoped

set_default_slug

normalise

GenericTableStore `dataclass`

LocalStore `dataclass`

ResolvedTable `dataclass`

identifier `property`

catalog `property`

NullIndex `dataclass`

FileRecord `dataclass`