Settings and credentials

databaas_utils.settings

How to reach a Lakekeeper warehouse, and how to prove who you are.

Credentials are a sum type rather than a bag of nullable fields: the prior art carried six optional attributes and validated their combinations in __post_init__, which made several invalid states representable. Here an OAuthClientCredentials or a StaticToken (or any TokenProvider) is supplied, and there is nothing left to validate.

LakehouseSettings is frozen so it can safely key a cache. Namespace is the one field that varies per call site, so it is changed with with_namespace rather than by mutation.

LakehouseSettings `dataclass`

LakehouseSettings(
    base_url: str,
    warehouse: str,
    credentials: Credentials,
    namespace: str = "default",
    region: str = DEFAULT_REGION,
    endpoint: str | None = None,
    verify: bool | str = True,
)

Everything needed to reach one warehouse as one identity.

Frozen, so it is safe as a cache key. Most code never builds one directly: use :meth:from_env, :meth:from_profile, :meth:from_jupyterhub, or (in a DAG) databaas_utils.airflow.get_settings.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	The Lakekeeper server root, e.g. `https://lakekeeper.example.com`. Not the `/catalog` endpoint. A missing scheme is taken to be `https://`.	required
`warehouse`	`str`	Warehouse name.	required
`credentials`	`Credentials`	Any :class:`TokenProvider`.	required
`namespace`	`str`	Default namespace for tables and generic tables.	`'default'`
`region`	`str`	Storage region, used until Lakekeeper's table config overrides it.	`DEFAULT_REGION`
`endpoint`	`str \| None`	S3 endpoint override, for MinIO and similar.	`None`
`verify`	`bool \| str`	TLS verification: `True`, `False` or a CA bundle path.	`True`

Raises:

Type	Description
`ConfigurationError`	If `base_url` or `warehouse` is empty, or `base_url` ends in `/catalog`.

Example

settings = LakehouseSettings( ... base_url="https://lakekeeper.example.com", ... warehouse="databaas-lakehouse", ... credentials=StaticToken("eyJ..."), ... namespace="open_meteo", ... )

catalog_url `property`

catalog_url: str

The Iceberg REST endpoint: base_url plus /catalog.

auth_header

auth_header() -> str

The current Authorization header value, from credentials.

with_namespace

with_namespace(namespace: str) -> LakehouseSettings

A copy pointing at namespace; self if it already does.

from_mapping `classmethod`

from_mapping(
    values: Mapping[str, Any],
    *,
    namespace: str | None = None,
) -> LakehouseSettings

Build from a flat mapping. Shared by the env and Airflow adapters.

Credentials are chosen by which keys are present, in this order: service_account_key, then client_id (+ client_secret), then token.

Parameters:

Name	Type	Description	Default
`values`	`Mapping[str, Any]`	Keys `base_url` (or `uri`), one credential set, and optionally `warehouse` (`databaas-lakehouse` by default), `token_url`, `project_id`, `scope`, `region`, `endpoint`, `verify`, `namespace`.	required
`namespace`	`str \| None`	Overrides `values["namespace"]`.	`None`

Raises:

Type	Description
`ConfigurationError`	If the mapping has no credentials, or a credential set without its token endpoint.

from_env `classmethod`

from_env(
    prefix: str = "LAKEKEEPER_",
    *,
    env: Mapping[str, str] | None = None,
    namespace: str | None = None,
) -> LakehouseSettings

Build from the environment a container or notebook already carries.

Reads {prefix}URI, WAREHOUSE, CLIENT_ID, CLIENT_SECRET, OAUTH2_SERVER_URI, PROJECT_ID, OAUTH2_SCOPE, TOKEN, REGION and ENDPOINT.

Parameters:

Name	Type	Description	Default
`prefix`	`str`	Variable-name prefix.	`'LAKEKEEPER_'`
`env`	`Mapping[str, str] \| None`	Mapping to read instead of `os.environ`.	`None`
`namespace`	`str \| None`	Default namespace.	`None`

Raises:

Type	Description
`ConfigurationError`	If the variables do not form a complete config.

from_profile `classmethod`

from_profile(
    name: str | None = None,
    *,
    namespace: str | None = None,
    interactive: bool = True,
    store: Any = None,
) -> LakehouseSettings

Build from a saved local profile, authenticating as the logged-in user.

This is the entry point for local, interactive use — log in first with databaas login (see databaas_utils.cli). Machine contexts (Airflow, CI) should keep using from_env/from_mapping with a service credential; a browser login is not something a scheduled task can complete.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Profile name; the default profile if omitted.	`None`
`namespace`	`str \| None`	Default namespace.	`None`
`interactive`	`bool`	Whether to open a browser/device login when no valid session is cached. Set `False` for unattended code, which then raises instead of prompting.	`True`
`store`	`Any`	A `TokenStore` to keep the session in; the OS keychain (or a private file) by default.	`None`

Raises:

Type	Description
`ConfigurationError`	If no such profile exists.

Example

settings = LakehouseSettings.from_profile() catalog = rest_catalog(settings)

from_jupyterhub `classmethod`

from_jupyterhub(
    *,
    namespace: str | None = None,
    warehouse: str | None = None,
    env: Mapping[str, str] | None = None,
) -> LakehouseSettings

Build inside a JupyterHub singleuser pod, as the logged-in user.

The token comes from the hub (see databaas_utils.jupyterhub). The pod's environment names the catalog: CATALOG_URI (the Lakekeeper server root) and LAKEKEEPER_WAREHOUSE. The storage region and endpoint are not read here — Lakekeeper supplies both in each table's config, which takes precedence over the client's.

Parameters:

Name	Type	Description	Default
`namespace`	`str \| None`	Default namespace.	`None`
`warehouse`	`str \| None`	Warehouse name; `LAKEKEEPER_WAREHOUSE` or `databaas-lakehouse` if omitted.	`None`
`env`	`Mapping[str, str] \| None`	Mapping to read instead of `os.environ`.	`None`

Raises:

Type	Description
`ConfigurationError`	If this is not a JupyterHub singleuser server.

TokenProvider

Bases: Protocol

Anything that can produce an Authorization header value.

This is the seam that lets a notebook pass the logged-in user's bearer instead of a service account, without this package learning what JupyterHub is.

auth_header

auth_header() -> str

The full Authorization header value, e.g. "Bearer eyJ...".

Called on every request, so implementations should cache and refresh.

StaticToken `dataclass`

StaticToken(token: str)

A bearer token supplied verbatim. Never refreshed.

Parameters:

Name	Type	Description	Default
`token`	`str`	The raw access token (a JWT that Lakekeeper can validate).	required

Example

settings = LakehouseSettings( ... base_url="https://lakekeeper.example.com", ... warehouse="databaas-lakehouse", ... credentials=StaticToken("eyJ..."), ... )

auth_header

auth_header() -> str

Bearer <token>.

OAuthClientCredentials `dataclass`

OAuthClientCredentials(
    client_id: str,
    client_secret: str,
    token_url: str,
    scope: str | None = None,
    project_id: str | None = None,
    _cache: dict[str, Any] = dict(),
    _lock: Lock = threading.Lock(),
)

OAuth2 client-credentials, cached and refreshed on demand.

project_id is Zitadel shorthand: it expands to the audience scope that Lakekeeper expects. Supply scope directly to override.

Parameters:

Name	Type	Description	Default
`client_id`	`str`	The machine user's OAuth client id.	required
`client_secret`	`str`	The matching client secret.	required
`token_url`	`str`	The Zitadel token endpoint, e.g. `https://auth.example.com/oauth/v2/token`.	required
`scope`	`str \| None`	Full OAuth scope string. Overrides `project_id`.	`None`
`project_id`	`str \| None`	Zitadel project id; expands to the audience scope.	`None`

Raises:

Type	Description
`ConfigurationError`	If neither `scope` nor `project_id` is given.

Example

creds = OAuthClientCredentials( ... client_id="databaas-pipeline", ... client_secret=os.environ["CLIENT_SECRET"], ... token_url="https://auth.example.com/oauth/v2/token", ... project_id="123456789", ... )

effective_scope `property`

effective_scope: str

scope if set, else the Zitadel project-audience scope.

auth_header

auth_header() -> str

Bearer <token>, fetching or refreshing the token if it is stale.

Raises:

Type	Description
`AuthenticationError`	If the token endpoint is unreachable or rejects the credentials.

ServiceAccountKey `dataclass`

ServiceAccountKey(
    key_id: str,
    private_key: str,
    user_id: str,
    token_url: str,
    scope: str | None = None,
    project_id: str | None = None,
    audience: str | None = None,
    _cache: dict[str, Any] = dict(),
    _lock: Lock = threading.Lock(),
)

A Zitadel service-account JSON key: Private Key JWT (RFC 7523) auth.

Some Zitadel consoles only offer a downloadable JSON key for a machine user, not a plain client secret — the key's fields are exactly type/keyId/key/userId, matching :meth:from_json. This signs a JWT assertion with the private key and exchanges it for a real access token via the urn:ietf:params:oauth:grant-type:jwt-bearer grant. The resulting token is a normal Zitadel-issued JWT — structurally identical to what client_credentials with a secret produces, so anything that accepts one accepts the other. Only how the client proves itself to Zitadel differs; a Lakekeeper (or anything else) validating the access token cannot tell the two apart.

A raw Zitadel Personal Access Token is not usable here or anywhere else against Lakekeeper: it is an opaque credential meant for calling Zitadel's own APIs, not a JWT a resource server's OIDC validator can parse. Signing an assertion and exchanging it for a real access token — what this class does — is the fix.

Needs the jwt extra: pip install "databaas-utils[jwt]".

Parameters:

Name	Type	Description	Default
`key_id`	`str`	The key's `keyId`.	required
`private_key`	`str`	The PEM private key (the JSON key's `key` field).	required
`user_id`	`str`	The machine user's id (the JSON key's `userId` field).	required
`token_url`	`str`	The Zitadel token endpoint.	required
`scope`	`str \| None`	Full OAuth scope string. Overrides `project_id`.	`None`
`project_id`	`str \| None`	Zitadel project id; expands to the audience scope.	`None`
`audience`	`str \| None`	The assertion's `aud`. Defaults to `token_url` without the `/oauth/v2/token` suffix.	`None`

Raises:

Type	Description
`ConfigurationError`	If neither `scope` nor `project_id` is given.

Example

creds = ServiceAccountKey.from_json( ... "/secrets/zitadel-key.json", ... token_url="https://auth.example.com/oauth/v2/token", ... project_id="123456789", ... )

effective_scope `property`

effective_scope: str

scope if set, else the Zitadel project-audience scope.

effective_audience `property`

effective_audience: str

The assertion's aud: audience, else the issuer from token_url.

from_json `classmethod`

from_json(
    source: Mapping[str, Any] | str | PathLike[str],
    **overrides: Any,
) -> ServiceAccountKey

Build from a Zitadel service-account JSON key file's own fields.

source is the file's parsed contents, its raw JSON text, or a path to the file — accepting all three covers "I already loaded it" and "I have the path" without asking the caller to pick one.

Parameters:

Name	Type	Description	Default
`source`	`Mapping[str, Any] \| str \| PathLike[str]`	Parsed key, raw JSON text, or path to the JSON file.	required
`**overrides`	`Any`	The remaining constructor fields (`token_url`, `project_id`, `scope`, `audience`).	`{}`

Raises:

Type	Description
`ConfigurationError`	If the key lacks `keyId`/`key`/`userId`.

auth_header

auth_header() -> str

Bearer <token>, signing a new assertion if the token is stale.

Raises:

Type	Description
`AuthenticationError`	If the token exchange fails.
`ConfigurationError`	If the private key cannot sign.

databaas_utils.userauth

Interactive login as the user's own Zitadel identity.

Two flows, both ending in a refresh token: Authorization Code + PKCE through a browser and a loopback redirect (interactive desktop use), and the OAuth2 Device Authorization Grant (SSH / headless, where nothing can reach a loopback redirect). After the first login, auth_header() renews silently from the refresh token; an interactive login only happens again once that refresh token itself has expired or been revoked.

Mirrors the double-checked, expiry-aware caching in settings.py's OAuthClientCredentials/ServiceAccountKey, but with two extra layers: the cache is cross-process (the token store), and refreshing must be serialised across processes too, since Zitadel rotates the refresh token on every use.

UserLogin `dataclass`

UserLogin(
    issuer: str,
    client_id: str,
    project_id: str,
    store_key: str,
    interactive: bool = True,
    flow: str = "auto",
    redirect_port: int = 0,
    store: TokenStore = default_store(),
)

A TokenProvider backed by a user's own interactive Zitadel login.

Built from a saved profile via UserLogin.from_profile — see LakehouseSettings.from_profile, the normal entry point. auth_header() serves a cached token, refreshes silently when a stored refresh token is still good, and — only when interactive is true — falls back to an interactive login when there is none.

scope `property`

scope: str

The OAuth scope requested at login.

from_profile `classmethod`

from_profile(
    profile: Profile, **overrides: Any
) -> UserLogin

Build from a saved :class:~databaas_utils.Profile.

Parameters:

Name	Type	Description	Default
`profile`	`Profile`	The profile to log in with.	required
`**overrides`	`Any`	Other fields, e.g. `interactive=False`, `flow="device"` or `store=...`.	`{}`

auth_header

auth_header() -> str

Bearer <token>, refreshing or logging in first if needed.

Raises:

Type	Description
`AuthenticationError`	If there is no usable session and `interactive` is false, or the login fails.

invalidate

invalidate() -> None

Drop the in-memory access token only. Used by 401-retry logic.

login

login() -> None

Run an interactive login now, regardless of any cached token.

logout

logout() -> None

Revoke the stored refresh token (best effort) and forget it.

claims

claims() -> dict[str, Any]

Decode the current access token's payload, unverified.

The server has already verified it; this is only for local display (databaas whoami), never for authorisation decisions.

userinfo

userinfo() -> dict[str, Any]

The signed-in user's profile (email, name) from the issuer.

Zitadel puts the email in the ID token and the userinfo response, not in the access token that :meth:claims decodes. Only for display; returns {} when the issuer has no userinfo endpoint or refuses.

login

login(
    profile: str | None = None, **overrides: Any
) -> UserLogin

Log in interactively and return a live UserLogin bound to the profile.

Convenience for notebooks: opens a browser (or prints a device code), persists the session, and returns something LakehouseSettings can be built from directly — or just call LakehouseSettings.from_profile() afterwards, which reuses the cache this writes.

Parameters:

Name	Type	Description	Default
`profile`	`str \| None`	Profile name; the default profile if omitted.	`None`
`**overrides`	`Any`	Passed to :meth:`UserLogin.from_profile`.	`{}`

Raises:

Type	Description
`ConfigurationError`	If no such profile exists.

databaas_utils.jupyterhub.JupyterHubToken `dataclass`

JupyterHubToken(
    api_url: str,
    api_token: str,
    timeout: float = REQUEST_TIMEOUT,
)

A TokenProvider that asks the hub for the user's access token.

Most code should use databaas_utils.user instead of this directly.

Parameters:

Name	Type	Description	Default
`api_url`	`str`	The hub API URL (`JUPYTERHUB_API_URL`).	required
`api_token`	`str`	The server's API token (`JUPYTERHUB_API_TOKEN`).	required
`timeout`	`float`	Per-request timeout in seconds.	`REQUEST_TIMEOUT`

endpoint `property`

endpoint: str

The hub's oauth-token URL.

from_env `classmethod`

from_env(
    env: Mapping[str, str] | None = None,
) -> JupyterHubToken

Build from JUPYTERHUB_API_URL and JUPYTERHUB_API_TOKEN.

Raises:

Type	Description
`ConfigurationError`	If either variable is unset.

auth_header

auth_header() -> str

Bearer <token>, asking the hub only when the cached token is stale.

Raises:

Type	Description
`AuthenticationError`	If the hub cannot supply a token. A 401 means the user's login expired; log out of Jupyter and back in.

invalidate

invalidate() -> None

Drop the cached token so the next call asks the hub again.

databaas_utils.profiles

Named local profiles: which Zitadel issuer and Lakekeeper warehouse to use.

One environment — one Zitadel instance, one Lakekeeper — is one profile. A machine can hold several (dev, a customer's prod, ...), selected with --profile/DATABAAS_PROFILE, with one marked default. discover() fills a profile in from the portal's public discovery document rather than requiring the values to be typed in by hand.

Profile `dataclass`

Profile(
    name: str,
    issuer: str,
    client_id: str,
    project_id: str,
    catalog_url: str,
    warehouse: str = DEFAULT_WAREHOUSE,
    region: str = DEFAULT_REGION,
    redirect_port: int = 0,
)

Everything needed to log in and reach one warehouse as one identity.

Profiles live in ~/.config/databaas/config.toml and are written by databaas login. Most code loads one with :func:load_profile or, more often, never touches it: LakehouseSettings.from_profile() does.

Parameters:

Name	Type	Description	Default
`name`	`str`	Profile name.	required
`issuer`	`str`	The Zitadel issuer URL.	required
`client_id`	`str`	The Native app's client id.	required
`project_id`	`str`	The Zitadel project id, requested as an extra token audience. Empty relies on the app's own project being in the audience, which Zitadel adds by default.	required
`catalog_url`	`str`	The Lakekeeper server root.	required
`warehouse`	`str`	Warehouse name, `databaas-lakehouse` by default.	`DEFAULT_WAREHOUSE`
`region`	`str`	Storage region.	`DEFAULT_REGION`
`redirect_port`	`int`	Fixed loopback port for the browser login; `0` picks a free one.	`0`

Raises:

Type	Description
`ConfigurationError`	If a required field is empty.

__post_init__

__post_init__() -> None

Reject a profile with an empty required field.

load_profile

load_profile(name: str | None = None) -> Profile

Resolve: an explicit name, then DATABAAS_PROFILE, else the default.

Raises:

Type	Description
`ConfigurationError`	If no profile is configured or the name is unknown.

discover

discover(
    portal_url: str, *, name: str | None = None
) -> Profile

Fetch a profile's settings from the portal's public discovery document.

portal_url is the environment's own address (e.g. https://infoplaza.databaas.eu), not the Zitadel issuer or the catalog. The document needs only client_id: the issuer and catalog default to the auth. and catalog. addresses next to the portal, and the warehouse to databaas-lakehouse.

Parameters:

Name	Type	Description	Default
`portal_url`	`str`	The environment's portal address.	required
`name`	`str \| None`	Profile name; the portal's first hostname label by default.	`None`

Raises:

Type	Description
`HTTPError`	If the portal serves no discovery document.

endpoints_from_base_url

endpoints_from_base_url(base_url: str) -> tuple[str, str]

The Zitadel issuer and Lakekeeper root of the environment at base_url.

A Databaas environment serves Zitadel on auth.<host> and Lakekeeper on catalog.<host>, next to the portal on <host>.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	The environment's portal address, e.g. `https://next.dev.databaas.app`. The scheme may be left out.	required

Returns:

Type	Description
`tuple[str, str]`	`(issuer, catalog_url)`.

Raises:

Type	Description
`ConfigurationError`	If `base_url` is already the `auth.` or `catalog.` address.

Example

endpoints_from_base_url("https://next.dev.databaas.app") ('https://auth.next.dev.databaas.app', 'https://catalog.next.dev.databaas.app')

Settings and credentials

databaas_utils.settings

LakehouseSettings dataclass

catalog_url property

auth_header

with_namespace

from_mapping classmethod

from_env classmethod

from_profile classmethod

from_jupyterhub classmethod

TokenProvider

auth_header

StaticToken dataclass

auth_header

OAuthClientCredentials dataclass

effective_scope property

auth_header

ServiceAccountKey dataclass

effective_scope property

effective_audience property

from_json classmethod

auth_header

databaas_utils.userauth

UserLogin dataclass

scope property

from_profile classmethod

auth_header

invalidate

login

logout

claims

userinfo

login

databaas_utils.jupyterhub.JupyterHubToken dataclass

endpoint property

from_env classmethod

auth_header

invalidate

databaas_utils.profiles

Profile dataclass

__post_init__

load_profile

discover

endpoints_from_base_url

LakehouseSettings `dataclass`

catalog_url `property`

from_mapping `classmethod`

from_env `classmethod`

from_profile `classmethod`

from_jupyterhub `classmethod`

StaticToken `dataclass`

OAuthClientCredentials `dataclass`

effective_scope `property`

ServiceAccountKey `dataclass`

effective_scope `property`

effective_audience `property`

from_json `classmethod`

UserLogin `dataclass`

scope `property`

from_profile `classmethod`

databaas_utils.jupyterhub.JupyterHubToken `dataclass`

endpoint `property`

from_env `classmethod`

Profile `dataclass`