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 Aug 9, 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 Aug 9, 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.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.99.tar.gz (13.8 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.99-py3-none-any.whl (16.3 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: dbx_tools_postgres-0.6.99.tar.gz
  • Upload date:
  • Size: 13.8 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.99.tar.gz
Algorithm Hash digest
SHA256 4c884a8a449cf7e3e6f555dba45ee58eba6b7dea57b126a22b545216d28df6b5
MD5 88f5b342005970215c63d5b9fa58e493
BLAKE2b-256 02d06eb4a454020352f51ba58cc0c7e748eb14b05a5c21d1c148fa55cce89a1f

See more details on using hashes here.

Provenance

The following attestation bundles were made for dbx_tools_postgres-0.6.99.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.99-py3-none-any.whl.

File metadata

File hashes

Hashes for dbx_tools_postgres-0.6.99-py3-none-any.whl
Algorithm Hash digest
SHA256 535070cf9be0a5944321d80383190f9c16024302949f4fe6679e3d124c7ebc8f
MD5 5590a0ce987bb94d1e7b012a07b611c8
BLAKE2b-256 bc9fdbcda3571baac3f42b8ec30035f128fb403bdf7118aeda6d9aa98ef6cb1a

See more details on using hashes here.

Provenance

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

0.6.140

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

This release

0.6.99 This release

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