Python SDK for the PgBeam API: a globally distributed PostgreSQL proxy that enforces agent access policy in the wire protocol, with connection pooling and query caching.

Manage projects, databases, agent credentials, and policy profiles from Python, with a blocking client and an asyncio client that expose exactly the same surface.

pip install pgbeam

Python 3.10 or newer. The only runtime dependency is httpx.

from pgbeam import PgBeamClient

with PgBeamClient(token="your-api-key") as client:
    # Tag-based access, one attribute per API area
    projects = client.projects.list_projects(org_id="org_123")

    for project in projects["projects"]:
        print(project["id"], project["name"])

    project = client.projects.get_project(project_id="prj_123")

Omit token and the client reads PGBEAM_API_KEY from the environment (PGBEAM_TOKEN and PGBEAM_API_TOKEN are accepted as aliases, checked in that order, as the CLI does; the Terraform and Pulumi providers read PGBEAM_API_KEY only). PGBEAM_API_URL overrides the base URL.

The async client is the same surface with await in front of it. Same method names, same arguments, same return types.

import asyncio
from pgbeam import AsyncPgBeamClient


async def main() -> None:
    async with AsyncPgBeamClient(token="your-api-key") as client:
        projects = await client.projects.list_projects(org_id="org_123")
        print(len(projects["projects"]))


asyncio.run(main())

Every request and response body is a TypedDict in pgbeam.models, and the package ships a py.typed marker, so mypy and pyright check your calls against the API contract.

from pgbeam import PgBeamClient
from pgbeam.models import CreateProjectRequest, Project

body: CreateProjectRequest = {
    "org_id": "org_123",
    "name": "analytics",
    "region": "us-east-1",
}

with PgBeamClient() as client:
    created = client.projects.create_project(body=body)
    project: Project = created["project"]

Timestamps are RFC 3339 strings, exactly as the API sends them. They are deliberately not parsed into datetime, so what you read is what came over the wire.

ApiError means the API answered and the answer was an error. NetworkError means it never answered at all: DNS failure, a refused or reset connection, or a timeout.

from pgbeam import ApiError, NetworkError, PgBeamClient

with PgBeamClient() as client:
    try:
        client.projects.get_project(project_id="prj_123")
    except ApiError as err:
        print(err.status, err)  # 404 Project not found
        print(err.body)  # the decoded error body
    except NetworkError as err:
        print(err.url, err.attempts, err.timed_out)

Every call retries 408, 429, 502, 503 and 504 up to five times with jittered exponential backoff, honouring Retry-After when the server sends one. A POST or PATCH that is retried carries one Idempotency-Key across all its attempts, so a retry of a request the server already accepted is not a second write.

Three bounds apply, in the order they bite: a 30 second per-attempt timeout, a five-retry ceiling, and a 120 second budget for the whole call measured from the first attempt. A retry that would land past the budget is not made.

from pgbeam import PgBeamClient, RetryConfig

with PgBeamClient(
    timeout_ms=10_000,
    retry=RetryConfig(max_retries=2, total_budget_ms=30_000),
) as client:
    ...

# No retrying at all
with PgBeamClient(retry=RetryConfig(max_retries=0)) as client:
    ...

The TypeScript SDK applies the same policy with the same defaults, so a 429 or a 409 means the same thing whichever one you reach for.

Pass an httpx.Client (or httpx.AsyncClient) to share a connection pool, route through a proxy, or install a custom transport. A client you supply is yours to close.

import httpx
from pgbeam import PgBeamClient

http = httpx.Client(proxy="http://localhost:8080")
client = PgBeamClient(token="...", http_client=http)

pgbeam.OPERATIONS_BY_TAG and pgbeam.OPERATIONS_BY_PATH carry every operation's method and path, for building tooling on top of the SDK.

from pgbeam import OPERATIONS_BY_PATH

print(OPERATIONS_BY_PATH["GET /v1/projects/{project_id}"])
# OperationMeta(method='GET', path='/v1/projects/{project_id}')

Full API reference at pgbeam.com/docs/python-sdk.

Issues and pull requests are welcome here. An issue is the right place to start for a bug, a wrong doc, or a missing capability; say what you ran, what happened, what you expected, and which version you were on.

models.py, operations.py and services.py are generated from the OpenAPI specification and carry a DO NOT EDIT header. A change to any of them belongs in the specification, not in the file. Everything else is ordinary hand-written Python.

To build and test it locally:

uv sync --all-groups
uv run ruff check
uv run ruff format --check
uv run mypy
uv run pytest

Do not open a public issue for a suspected security vulnerability. Email [email protected], or report it privately from this repository's Security tab.

Apache 2.0, see LICENSE.