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.

Project links

Data verified by PyPI on Sep 3, 2026
Data provided by the project maintainers, verified at the time the release was uploaded to PyPI.

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Newer release available (0.6.201)

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for reggie-db from gravatar.com reggie-db

GitHub Statistics

Data verified by PyPI on Sep 3, 2026
The GitHub source repository was provided by the project maintainers and verified by PyPI at the time of upload. Stars, forks, and open issues/PRs are derived from that repository and have not been independently verified.

Requires

Python <3.14, >=3.10

Report project as malware

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.140.tar.gz (13.9 kB view details)

Uploaded Source

Built Distribution

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

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

Uploaded Python 3

File details

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

File metadata

  • Download URL: dbx_tools_postgres-0.6.140.tar.gz
  • Upload date:
  • 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.140.tar.gz
Algorithm Hash digest
SHA256 bb1df223b01ecc54b41eb3f5e6246d0b32df798ea0db682eb687344f86bd0609
MD5 b749dab7e4a3c525a9841640c39227e0
BLAKE2b-256 42e708246b4c3044f545f1f58a99c089b427da33d733ee73c776a557bf28ee94

See more details on using hashes here.

Provenance

The following attestation bundles were made for dbx_tools_postgres-0.6.140.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.

File details

Details for the file dbx_tools_postgres-0.6.140-py3-none-any.whl.

File metadata

File hashes

Hashes for dbx_tools_postgres-0.6.140-py3-none-any.whl
Algorithm Hash digest
SHA256 c76fdf1643a8671d7b11e51c8af4d69534d74c01add5bd1375744d98c08386ad
MD5 1db68feb9cb059e7a7449481f73ea0fb
BLAKE2b-256 9330c499a86b8f969d5585f9fc615be04a8955a358ebc6c01bd7f67720b13d59

See more details on using hashes here.

Provenance

The following attestation bundles were made for dbx_tools_postgres-0.6.140-py3-none-any.whl:

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.

Release history Release notifications | RSS feed

0.6.201

2 files

0.6.200

2 files

0.6.199

2 files

0.6.198

2 files

0.6.197

2 files

0.6.196

2 files

0.6.195

2 files

0.6.194

2 files

0.6.193

2 files

0.6.192

2 files

0.6.191

2 files

0.6.188

2 files

0.6.186

2 files

0.6.185

2 files

0.6.183

2 files

0.6.182

2 files

0.6.181

2 files

0.6.180

2 files

0.6.179

2 files

0.6.177

2 files

0.6.176

2 files

0.6.175

2 files

0.6.174

2 files

0.6.173

2 files

0.6.172

2 files

0.6.171

2 files

0.6.170

2 files

0.6.168

2 files

0.6.163

2 files

0.6.161

2 files

0.6.160

2 files

0.6.158

2 files

0.6.153

2 files

0.6.147

2 files

0.6.146

2 files

0.6.145

2 files

0.6.144

2 files

0.6.143

2 files

0.6.142

2 files

0.6.141

2 files

This release

0.6.140 This release

2 files

0.6.139

2 files

0.6.138

2 files

0.6.137

2 files

0.6.136

2 files

0.6.135

2 files

0.6.134

2 files

0.6.133

2 files

0.6.131

2 files

0.6.130

2 files

0.6.129

2 files

0.6.128

2 files

0.6.127

2 files

0.6.126

2 files

0.6.125

2 files

0.6.124

2 files

0.6.123

2 files

0.6.122

2 files

0.6.121

2 files

0.6.120

2 files

0.6.119

2 files

0.6.118

2 files

0.6.117

2 files

0.6.116

2 files

0.6.115

2 files

0.6.114

2 files

0.6.113

2 files

0.6.112

2 files

0.6.111

2 files

0.6.110

2 files

0.6.109

2 files

0.6.108

2 files

0.6.107

2 files

0.6.106

2 files

0.6.105

2 files

0.6.104

2 files

0.6.103

2 files

0.6.102

2 files

0.6.101

2 files

0.6.100

2 files

0.6.99

2 files

0.6.98

2 files

0.6.97

2 files

0.6.96

2 files

0.6.95

2 files

0.6.94

2 files

0.6.93

2 files

0.6.92

2 files

0.6.91

2 files

0.6.90

2 files

0.6.89

2 files

0.6.88

2 files

0.6.87

2 files

0.6.86

2 files

0.6.85

2 files

0.6.82

2 files

0.6.78

2 files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page