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 8, 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 8, 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.11

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.183.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.183-py3-none-any.whl (16.4 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: dbx_tools_postgres-0.6.183.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.183.tar.gz
Algorithm Hash digest
SHA256 a59e2b683d5ef515ec0d2344520ccd81e9f0f369b031bffe773ca3f9a5e3a351
MD5 980a1b03b68da7ee9a87fc4aeb3a3486
BLAKE2b-256 c15aa69fbfc0ecc1f2e1dfaedcd6152ebf8f9e326937f1fc7ac50f6fd661b1a2

See more details on using hashes here.

Provenance

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

Publisher: 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.183-py3-none-any.whl.

File metadata

File hashes

Hashes for dbx_tools_postgres-0.6.183-py3-none-any.whl
Algorithm Hash digest
SHA256 5ace72aa10313896ecffeb4ef27c183caa5d9bc546009a7569a930d703c5d140
MD5 ee21f7cfdab63256ea0f28b4a8d77ea0
BLAKE2b-256 a7f771bab6224c442913d6b0136f6d0c010baca9acc43f3b06318bac3b5fd957

See more details on using hashes here.

Provenance

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

Publisher: 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

This release

0.6.183 This release

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

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