pgmesh

An async orchestration layer for applications that talk to many PostgreSQL databases.

Register your databases once, address them by index or label, and let pgmesh own the connection pools, the parallelism, the timeouts and the failure isolation.

from pgmesh import PGCluster

async with PGCluster({
    1: "postgresql://user:pass@db1/app",
    2: "postgresql://user:pass@db2/app",
    "analytics": "postgresql://user:pass@db3/analytics",
}, max_concurrency=10, query_timeout=5) as db:

    users = await db.connection(1).execute("SELECT * FROM users LIMIT 10")

    results = await db.parallel([
        (1, "SELECT count(*) FROM users"),
        (2, "SELECT count(*) FROM orders"),
        ("analytics", "SELECT count(*) FROM events"),
    ])

The problem

An application with more than one PostgreSQL database ends up hand-rolling the same infrastructure every time: a connection string per database, a pool per database, some routing logic, an asyncio.gather for the fan-out, a timeout that half-works, and a try/except that lets one dead tenant take down an endpoint that never needed it.

Application
    |
    v
  pgmesh
    |
    +---- DB 1 / tenant_a       Pool(min=1, max=10)
    +---- DB 2 / tenant_b       Pool(min=1, max=10)
    +---- DB 3 / analytics      Pool(min=1, max=10)
    +---- DB 4 / tenant_c       Pool(min=1, max=10)

pgmesh is that layer, and nothing more. It is not a proxy, a driver, a sharding engine, or a query rewriter. Routing is explicit, statements go straight to asyncpg, and there is no server to run.

Install

pip install pgmesh

Python 3.10+. One runtime dependency: asyncpg.

Quickstart

Register

Both integer indexes and string labels work, and they share one namespace — 1 and "1" address the same database.

from pgmesh import PGCluster

db = PGCluster({
    1: "postgresql://user:pass@db1/app",
    "analytics": "postgresql://user:pass@db3/analytics",
})

Connection strings come from your configuration or environment. pgmesh never reads them itself and never manages your secrets.

Query one database

rows  = await db.connection(1).execute("SELECT * FROM users WHERE status = $1", "active")
row   = await db.connection(1).fetchrow("SELECT * FROM users WHERE id = $1", 42)
count = await db.connection("analytics").fetchval("SELECT count(*) FROM events")
tag   = await db.connection(1).command("UPDATE users SET seen = now()")   # 'UPDATE 3'

Parameters are bound server-side as $1, $2, … — they are never interpolated into the statement text.

Query many databases at once

results = await db.parallel([
    (1, "SELECT count(*) FROM users"),
    (2, "SELECT count(*) FROM orders WHERE status = $1", ["open"]),
    ("analytics", "SELECT count(*) FROM events"),
])

You get back a dict keyed by database:

results[1].value          # rows from database 1
results.successes         # {database: rows} for the ones that worked
results.failures          # {database: error} for the ones that did not
results.all_ok            # bool

Same statement everywhere:

await db.parallel_map("SELECT count(*) FROM users")
await db.parallel_map("SELECT count(*) FROM users", databases=[1, 2])

Lifecycle

async with PGCluster(databases) as db:
    ...
# every pool is closed on the way out

Pools are created on first use. Call await db.startup() to open them all up front, so a bad connection string fails at boot rather than inside the first request. await db.close() is idempotent and safe in a finally.

What it guarantees

One pool per database, reused

Every registered database gets its own asyncpg pool. A new TCP connection is never opened per query.

Query → acquire from pool → execute → release back to pool

Bounded concurrency

max_concurrency caps how many operations are in flight in a single parallel() call. Submit 100 operations against a limit of 20 and 20 run while 80 queue; slots free up as work finishes. pgmesh never spawns an unbounded number of tasks.

db = PGCluster(databases, max_concurrency=20)
await db.parallel(operations, max_concurrency=5)   # or per call

Timeouts that release the connection

db = PGCluster(databases, query_timeout=5)             # cluster default
await db.connection(1).execute(sql, timeout=0.5)       # per query

A statement that overruns raises QueryTimeoutError, and its connection goes straight back to the pool — a timeout can never strand a connection or drain a pool.

Failure isolation

One database being slow, down, or misconfigured does not cancel operations against the others.

results = await db.parallel_map("SELECT count(*) FROM users")

for database, result in results.items():
    if result.ok:
        print(database, result.value)
    else:
        print(database, "failed:", result.error)     # QueryTimeoutError, ...

Every entry is a Success(database, value) or a Failure(database, error), so you can always tell which database failed and why. Nothing is raised unless you ask:

results.raise_for_failures()                       # after the fact
await db.parallel(ops, raise_on_error=True)        # or up front — still runs everything

Errors you can actually catch

Driver exceptions are translated into one documented hierarchy, with the original attached as __cause__ and .original:

PGMeshError
├── DatabaseNotFoundError        unknown id or label
├── DatabaseConfigurationError   bad identifier, DSN or option
├── DatabaseConnectionError      pool or connection failure
├── QueryExecutionError          the server rejected the statement
├── QueryTimeoutError            exceeded its timeout budget
└── ClusterClosedError           used after close()

DatabaseNotFoundError is also a KeyError and QueryTimeoutError is also a TimeoutError, so existing error handling keeps working.

No secrets in your logs

Passwords never appear in log lines, reprs, or exception messages — anywhere a DSN might surface, it is masked first.

>>> from pgmesh import mask_dsn
>>> mask_dsn("postgresql://user:hunter2@db:5432/app")
'postgresql://user:***@db:5432/app'

Beyond the basics

Transactions

async with db.connection(1).transaction() as conn:
    await conn.execute("INSERT INTO orders(total) VALUES ($1)", 99)
    await conn.execute("UPDATE stock SET n = n - 1 WHERE sku = $1", "abc")
# commits on a clean exit, rolls back on any exception

Transactions are per-database. pgmesh does not do distributed transactions, and does not pretend to.

Raw connections

For anything pgmesh does not wrap — COPY, cursors, LISTEN/NOTIFY:

async with db.connection(1).acquire() as conn:
    await conn.copy_to_table("users", source=path)

The connection is always released, including when the body raises.

Health

await db.health()                    # {1: True, 2: False, "analytics": True}
await db.connection(1).ping()        # True / False, never raises

Inspection

db.databases          # [1, 2, "analytics"]
1 in db               # True
db[1]                 # same as db.connection(1)
db.describe()         # per-database config, passwords masked

Tuning

PGCluster(
    databases,
    max_concurrency=20,      # ops in flight per parallel() call
    query_timeout=5,         # seconds; None for no client-side limit
    connect_timeout=10,      # seconds to open a connection
    pool_min_size=1,
    pool_max_size=10,        # per database
    connect_kwargs={"ssl": "require"},   # passed through to asyncpg
)

Sizing note: the cluster-wide ceiling is len(databases) × pool_max_size. Check it against your server's max_connections before registering many databases.

With FastAPI

Build the cluster once at startup and share it across requests — never one per request, which would mean one pool per request.

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from pgmesh import PGCluster

@asynccontextmanager
async def lifespan(app: FastAPI):
    cluster = PGCluster(load_databases(), query_timeout=5)
    app.state.cluster = cluster
    await cluster.startup()
    try:
        yield
    finally:
        await cluster.close()

app = FastAPI(lifespan=lifespan)

@app.get("/users/{database}")
async def users(database: int, request: Request):
    rows = await request.app.state.cluster.connection(database).execute(
        "SELECT * FROM users LIMIT 10"
    )
    return [dict(r) for r in rows]

A complete service — dependency wiring, fan-out endpoint, and the pgmesh error hierarchy mapped onto HTTP status codes — is in examples/fastapi_demo/main.py.

cp .env.example .env
./run.sh          # http://127.0.0.1:8000/docs

Development

git clone https://github.com/Mayuradlak123/pgmesh
cd pgmesh
./setup.sh                       # or: uv sync

uv run pytest                    # unit tests — no PostgreSQL needed
uv run ruff check .
uv run mypy

The unit suite runs against a fake driver, so it is fast and hermetic. Integration tests need a real server:

docker compose up -d
PGMESH_TEST_DSN=postgresql://postgres:postgres@localhost:5432/postgres \
  uv run pytest -m integration

Releasing

Publishing runs on GitHub Actions via PyPI Trusted Publishing — no API token lives in the repo.

1. Bump version in pyproject.toml and __version__ in src/pgmesh/__init__.py.
2. Merge to main and let CI pass.
3. Publish a GitHub Release tagged vX.Y.Z.

The workflow refuses to publish if the tag and the two versions disagree, then builds the wheel and sdist, runs twine check, and uploads.

Scope

In: registration, id/label routing, one pool per database, query execution, bounded parallel execution, timeouts, failure isolation, a clean error model.

Out, deliberately: proxying, SQL parsing, distributed transactions, cross-database joins, automatic sharding, query rewriting, replication management. Retries, circuit breakers, structured logging and metrics are on the roadmap, not in v0.1.

License

MIT — see LICENSE.