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. Lakebase address parsing comes
directly from the generated dbx-tools-core-rs Rust bindings.
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/appkitthrough the native Rust parser; - 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_connectevent rather than storing an expiring password in the engine URL, using the SDK's provisioned-instance API or the Autoscaling/postgres/credentialsendpoint; - 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, andpublishedAt.
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.runin a cell raisesRuntimeError: 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 fornest_asyncio— the bus holds a dedicatedLISTENconnection bound to whichever loop started it, so one loop per bus lifetime is the invariant to preserve. - Install with
%pip install, notpip install --target. A--targetinstall leaves the runtime's preloadedtyping_extensionsahead of the new one onsys.path, and importingdbx_tools.postgresthen fails withImportError: cannot import name 'TypeAliasType'.%piprestarts 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.broadcastis 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.
1 maintainer
Release files for dbx-tools-postgres 0.6.214
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| dbx_tools_postgres-0.6.214.tar.gz | 20.1 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| dbx_tools_postgres-0.6.214-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 43.3 kB
Release files / dbx_tools_postgres-0.6.214.tar.gz
| Download URL | dbx_tools_postgres-0.6.214.tar.gz |
|---|---|
| Size | 20.1 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
b3bba60f6d2ee0ab75e0405c403e5cbe43d0ec55f9352efb64af96a8c1bf16af
|
|
BLAKE2b-256 checksum How to use checksums |
5a02b64bb578ef1159cb334f2f8f63cc056624ba7e25a60cca724812523cba92
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Sep 24, 2026.
Transparency logRelease files / dbx_tools_postgres-0.6.214-py3-none-any.whl
| Download URL | dbx_tools_postgres-0.6.214-py3-none-any.whl |
|---|---|
| Size | 23.1 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
3ff32f51c546bec7ecbb8a25b976e8efe4036dcf29cd2209bc4dd7ad8fcd4fa8
|
|
BLAKE2b-256 checksum How to use checksums |
0aa52dedbd277138a42796eb5368db81e3ccefe7db29242d67f3237f302f7901
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Sep 24, 2026.
Transparency log
