The Python SDK

Install and connect

pip install pgbhub

from pgbhub import PGBHubClient

client = PGBHubClient(api_key="your_api_key", project_id="my-project")

project_id alone targets https://my-project.pgbhub.app. You can set everything in the environment instead: PGBHUB_API_KEY (or PGBHUB_PUBLISHABLE_KEY), PGBHUB_PROJECT_ID and PGBHUB_ENDPOINT. Timeouts default to 30 seconds, with two retries for reads. Writes never retry.

First query

Tables are available as attributes and behave like models:

User = client.user

user = User.find_one(filter={"email": "[email protected]"})
created = User.create(data={"email": "[email protected]", "name": "Alex"})
User.update(filter={"id": "8ef5-492a-bb81"}, data={"name": "Alexander"})
User.delete(filter={"id": "8ef5-492a-bb81"})

Async for web frameworks

from fastapi import FastAPI
from pgbhub import AsyncPGBHubClient

app = FastAPI()
client = AsyncPGBHubClient(api_key="...")

@app.get("/users/{user_id}")
async def get_user(user_id: str):
    return await client.user.find_one(filter={"id": user_id})

AsyncPGBHubClient has the same methods, and each is awaitable.

Auth

auth = client.auth
result = auth.sign_up("[email protected]", "a-long-password", redirect_to="https://app.example.com/welcome")
if result.get("verification_required"):
    ...  # the user must click the emailed link

login = auth.sign_in("[email protected]", "a-long-password")
if login.get("mfa_required"):
    auth.verify_mfa(login["challenge_token"], "123456")

auth.forgot_password("[email protected]", redirect_to="https://app.example.com/reset")

One-time codes and links are only emailed. The API never returns them to the app. redirect_to must be allowed in your project's authentication settings.

Mail and storage

client.mail.send("[email protected]", "Your receipt", "Thanks for your order.",
                 html="<p>Thanks for your order.</p>", reply_to="[email protected]")

client.storage.create_bucket("receipts", visibility="private")
link = client.storage.create_signed_url("receipts", "invoices/inv-1.pdf", expires_in=900)
print(link["url"], link["expires_at"])

Both need the secret key. Use them on a server.

Actions

client.actions.save_user({"email": "[email protected]"})
client.actions.save_user({"email": "[email protected]"}, dry_run=True)

client.actions.register(
    name="signup",
    sql="INSERT INTO users (email) VALUES (%(email)s) RETURNING id",
    effects=[{"type": "queue", "queue": "onboarding", "payload": {"user": "{{results.0.id}}"}}],
)
client.actions.deliveries(status="failed")

Errors and models

from pgbhub import PGBHubError
try:
    client.auth.sign_in("[email protected]", "wrong")
except PGBHubError as e:
    print(e.status_code, e.code, e.message, e.retry_after)

Generate typed models from your schema with pgbhub pull --db my_database --out models.py.

Try it

1. Install the package and create a client from environment variables.
2. Find a row with find_one.
3. Register an action with a queue effect and call it with dry_run=True to see the preview.
4. Trigger and print a PGBHubError.

Recap

• PGBHubClient and AsyncPGBHubClient share the same calls.
• Tables are attributes of the client.
• Admin calls such as mail and storage need the secret key.
• pgbhub pull generates models.