`dbx-tools-postgres`

Python Lakebase/Postgres connection setup, advisory locks, and topic fan-out for services that already hold a Databricks WorkspaceClient. This package is the Python counterpart to @dbx-tools/postgres plus @dbx-tools/appkit's address parsing.

Install from PyPI:

pip install dbx-tools-postgres

To install the current main branch directly from the repository instead:

pip install "dbx-tools-postgres @ git+https://github.com/reggie-db/dbx-tools.git@main#subdirectory=packages/py/postgres"

Key features:

accepts the same Postgres URI, Lakebase resource path, hostname, and project-id address shapes as @dbx-tools/appkit;
resolves missing autoscaling endpoint fields through WorkspaceClient.api_client;
resolves provisioned Lakebase instance DNS through WorkspaceClient.database;
injects a cached database credential on SQLAlchemy's do_connect event rather than storing an expiring password in the engine URL, using the SDK's provisioned-instance API or the Autoscaling /postgres/credentials endpoint;
refreshes built-in credential providers ahead of expiry with a process-local check-lock-check load, so concurrent pool connections share one mint;
supports sync psycopg and asyncpg SQLAlchemy engines;
derives advisory-lock ids from the same stable structured keys as the Node package and holds one checked-out connection for the full critical section;
provides blocking and try-lock context managers for session and transaction locks, with sync and async SQLAlchemy variants;
fans messages out to every process on a channel with PostgresTopicBus, using the same lifecycle and wire envelope as the Node package.

from databricks.sdk import WorkspaceClient
from dbx_tools.postgres import PostgresEngineConfig, create_async_engine

engine = create_async_engine(
    WorkspaceClient(),
    PostgresEngineConfig(instance_name="my-lakebase", database="databricks_postgres"),
    pool_pre_ping=True,
    pool_recycle=1800,
)

Pass credential_provider= to either engine factory to inject credentials from another source. A custom provider owns its own cache, expiry, and refresh serialization policy.

from dbx_tools.postgres import advisory_transaction_lock

with advisory_transaction_lock(engine, ["schema-install", "v2"]) as connection:
    connection.exec_driver_sql("CREATE TABLE IF NOT EXISTS ...")

Topic bus

PostgresTopicBus is async Postgres topic fan-out built on LISTEN/NOTIFY. Its public lifecycle and wire shape match @dbx-tools/postgres's PostgresTopicBus, so Node and Python services can share a channel:

PostgresTopicBus(engine, options);
channelName;
await start();
await broadcast(topic, TopicPublishInput(...));
await listen(topic, listener) returning an async unsubscribe function;
await close();
envelope fields id, topic, type, metadata, body, and publishedAt.

Channel derivation ports the Node stable-key and FNV rules, so equivalent channel parts resolve to the same PostgreSQL identifier in Python and Node.

from dbx_tools.postgres import PostgresTopicBus, TopicPublishInput

bus = PostgresTopicBus(engine, channel=["billing", "production"])

unsubscribe = await bus.listen("invoice.updated", handle_invoice)
await bus.broadcast(
    "invoice.updated",
    TopicPublishInput(type="invoice.updated", body={"invoice_id": "inv-7"}),
)

Delivery is live and unstored, like PostgreSQL LISTEN/NOTIFY itself. Use a table or queue when consumers need replay or acknowledgements.

Databricks notebooks and Spark

Verified end to end against a Lakebase endpoint on serverless notebook compute. packages/example/notebooks/bus-lakebase.py is the runnable version of everything below.

Two things about the Databricks Python runtime change how the bus is called, and neither is a limitation of the bus itself:

A notebook kernel already runs an event loop, so asyncio.run in a cell raises RuntimeError: asyncio.run() cannot be called from a running event loop. Drive the coroutine on a short-lived thread with its own loop rather than reaching for nest_asyncio — the bus holds a dedicated LISTEN connection bound to whichever loop started it, so one loop per bus lifetime is the invariant to preserve.
Install with %pip install, not pip install --target. A --target install leaves the runtime's preloaded typing_extensions ahead of the new one on sys.path, and importing dbx_tools.postgres then fails with ImportError: cannot import name 'TypeAliasType'. %pip restarts the Python process, which resolves it.

Publishing from a Spark UDF

Publishing from executors works. Listening from them does not, and should not be attempted: a UDF invocation is short-lived, while listen keeps a connection open until close.

Executors have no Databricks credentials, so they cannot build a WorkspaceClient. This is where connect-time credential injection pays off — the driver mints the Lakebase token once and the UDF closes over it, so the executor builds a plain SQLAlchemy engine and installs the token as its provider:

from sqlalchemy import URL
from sqlalchemy.ext.asyncio import create_async_engine
from dbx_tools.postgres import (
    PostgresTopicBus,
    TopicPublishInput,
    install_credential_injection,
)

# driver: resolve once, capture in the closure
resolved = resolve_postgres_connection(workspace_client, config)
token = workspace_client.api_client.do(
    "POST",
    "/api/2.0/postgres/credentials",
    body={"endpoint": resolved.endpoint},
)["token"]


@udf(returnType=StringType())
def publish(key: str) -> str:
    async def run() -> str:
        engine = create_async_engine(
            URL.create(
                "postgresql+asyncpg",
                username=resolved.user,
                host=resolved.host,
                port=resolved.port,
                database=resolved.database,
                query={"ssl": resolved.ssl_mode},
            )
        )
        install_credential_injection(engine.sync_engine, lambda: token)
        bus = PostgresTopicBus(engine, channel="app-events")
        try:
            message = await bus.broadcast(
                "row.processed", TopicPublishInput(type="row.processed", body={"key": key})
            )
            return message.id
        finally:
            await bus.close()
            await engine.dispose()

    return asyncio.run(run())

Constraints worth knowing before this reaches production:

A captured token EXPIRES (about an hour). Re-mint per job run; a long-running streaming query needs a provider that refreshes instead of a captured string.
Each UDF call opens and closes its own connection, so batch the publish at partition scope (mapInPandas, foreachPartition) rather than per row.
spark.sparkContext.broadcast is unavailable on serverless (Spark Connect). A plain closure over driver-side values serializes with the UDF and is enough.
Delivery stays live and unstored: if no listener is connected when the UDF publishes, the message is gone. Write to a table when executors produce results a consumer must not miss.

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

dbx_tools_postgres-0.6.160.tar.gz (13.9 kB view details)

Uploaded Sep 4, 2026 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

dbx_tools_postgres-0.6.160-py3-none-any.whl (16.4 kB view details)

Uploaded Sep 4, 2026 Python 3

File details

Details for the file dbx_tools_postgres-0.6.160.tar.gz.

File metadata

Download URL: dbx_tools_postgres-0.6.160.tar.gz
Upload date: Sep 4, 2026
Size: 13.9 kB
Tags: Source
Uploaded using Trusted Publishing? Yes
Uploaded via: twine/7.0.0 CPython/3.13.14

File hashes

Hashes for dbx_tools_postgres-0.6.160.tar.gz
Algorithm	Hash digest
SHA256	`8791ce682d7cd01d3f63b41fdfc690296e4937965943b87a332a129a8e572dc0`
MD5	`564fb16db5c50fc7db77739371dfac88`
BLAKE2b-256	`40ace6eb53ff7cc18455e5fcaebd5bc22520fbdcab57a4c2525696647b88a781`

See more details on using hashes here.

Provenance

The following attestation bundles were made for dbx_tools_postgres-0.6.160.tar.gz:

Publisher: python-release.yml on reggie-db/dbx-tools

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Statement:
- Statement type: https://in-toto.io/Statement/v1
- Predicate type: https://docs.pypi.org/attestations/publish/v1
- Subject name: dbx_tools_postgres-0.6.160.tar.gz
- Subject digest: 8791ce682d7cd01d3f63b41fdfc690296e4937965943b87a332a129a8e572dc0
- Sigstore transparency entry: 2714520131
- Sigstore integration time: Sep 4, 2026
Source repository:
- Permalink: reggie-db/dbx-tools@8810f52619fd4fb950d73b61c0331af1093cec2a
- Branch / Tag: refs/heads/main
- Owner: https://github.com/reggie-db
- Access: public
Publication detail:
- Token Issuer: https://token.actions.githubusercontent.com
- Runner Environment: github-hosted
- Publication workflow: python-release.yml@8810f52619fd4fb950d73b61c0331af1093cec2a
- Trigger Event: repository_dispatch