Metadata-Version: 2.5
Name: pgmesh
Version: 0.2.0
Summary: Async orchestration layer for applications talking to many PostgreSQL databases — one pool per database, parallel queries, concurrency limits, timeouts and failure isolation.
Project-URL: Homepage, https://github.com/Mayuradlak123/pgmesh
Project-URL: Bug Tracker, https://github.com/Mayuradlak123/pgmesh/issues
Project-URL: Changelog, https://github.com/Mayuradlak123/pgmesh/releases
Author-email: Mayur Adlak <mayuradlak030@gmail.com>
License: MIT License
        
        Copyright (c) 2026 Mayur Adlak
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
License-File: LICENSE
Keywords: asyncio,asyncpg,connection-pool,database,multi-tenant,postgres,postgresql
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Database
Classifier: Topic :: Database :: Front-Ends
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: asyncpg>=0.29
Provides-Extra: demo
Requires-Dist: fastapi>=0.110; extra == 'demo'
Requires-Dist: python-dotenv>=1.0; extra == 'demo'
Requires-Dist: uvicorn[standard]>=0.27; extra == 'demo'
Description-Content-Type: text/markdown

# pgmesh

[![PyPI](https://img.shields.io/pypi/v/pgmesh.svg)](https://pypi.org/project/pgmesh/)
[![Python](https://img.shields.io/pypi/pyversions/pgmesh.svg)](https://pypi.org/project/pgmesh/)
[![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE)

**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.

```python
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`](https://github.com/MagicStack/asyncpg), and there is no server to run.

---

## Install

```bash
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.

```python
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

```python
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.

| Method | Returns |
| --- | --- |
| `execute(sql, *args)` / `fetch(...)` | every row, as a list |
| `fetchrow(sql, *args)` | the first row, or `None` |
| `fetchval(sql, *args)` | the first column of the first row |
| `command(sql, *args)` | PostgreSQL's status tag, e.g. `'UPDATE 3'` |
| `executemany(sql, args_seq)` | `None` — one execution per parameter tuple |
| `explain(sql, *args)` | the query plan, one string per line |

### Query many databases at once

```python
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:

```python
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:

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

### Lifecycle

```python
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.

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

### Timeouts that release the connection

```python
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.

```python
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:

```python
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.

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

---

## Beyond the basics

### Transactions

```python
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`:

```python
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.

### Query plans

Ask for a plan directly:

```python
plan = await db.connection(1).explain("SELECT * FROM users WHERE email = $1", "a@b.com")
for line in plan:
    print(line)
```

Or turn on plan capture for every query — **off by default**:

```python
db = PGCluster(databases, explain=True)
```

With the flag on, each statement's plan is logged to the `pgmesh` logger at `INFO`
before the statement runs, on the same connection so the plan describes the same session.

Two things worth knowing:

- The flag uses **plain `EXPLAIN`**, which plans a statement without executing it. Your
  `INSERT` reaches the server exactly once. This is deliberate — `EXPLAIN ANALYZE` *does*
  execute, so using it here would double-apply every write.
- `explain(sql, analyze=True)` switches to `EXPLAIN (ANALYZE, BUFFERS, VERBOSE)` for real
  timings. That **does** execute the statement. On a write, wrap it in a transaction you
  roll back.

Plan capture costs an extra round trip per query, so it's a debugging aid rather than
something to leave on in production. A statement PostgreSQL can't explain (`VACUUM`,
`SET`) is skipped silently — a diagnostic never becomes the reason a query fails.

### Health

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

### Inspection

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

### Tuning

```python
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
    explain=False,           # log a query plan for every statement
    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.

```python
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`](examples/fastapi_demo/main.py).

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

---

## Development

```bash
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:

```bash
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](https://docs.pypi.org/trusted-publishers/)
— 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](LICENSE).
