Python Postgres Backend: SDK, SQLAlchemy and FastAPI

A Python Postgres backend can mean two very different things: Python code that talks to a database over a Postgres connection, or Python code that talks to a backend API that sits in front of the database. With StitchDB you get both from the same project. The stitchdb package on PyPI calls the REST, Auth, Storage, Realtime and Functions APIs; the connection pooler accepts psycopg, asyncpg and SQLAlchemy like any Postgres. This guide shows when to use which, with code for scripts, Django or Flask style apps and FastAPI.

#Three ways in, side by side

The deciding question is whose permissions the code should have. If the answer is "the user who made the request", use the SDK with their token so your policies keep protecting the data. If the answer is "the application", a direct connection or the service key is simpler and faster.

#Install the SDK

pip install stitchdb

Python 3.9 or newer. Pin a version in requirements.txt, for example stitchdb==2.1.0. It also works with uv add stitchdb and poetry add stitchdb.

import os
from stitchdb import create_client

stitchdb = create_client(os.environ["STITCHDB_URL"], os.environ["STITCHDB_KEY"])
# STITCHDB_URL = https://<project-ref>.p.stitchdb.com

#Queries with the SDK

The query builder mirrors the REST API filters:

# Select with filters, order and pagination
res = (stitchdb.table("todos")
    .select("id, task, is_complete")
    .eq("is_complete", False)
    .order("created_at", desc=True)
    .range(0, 9)
    .execute())
rows = res.data

# Insert, update, upsert, delete (the changed rows are returned)
stitchdb.table("todos").insert({"task": "Ship it"}).execute()
stitchdb.table("todos").update({"is_complete": True}).eq("id", 1).execute()
stitchdb.table("todos").upsert({"id": 1, "task": "Ship it today"}).execute()
stitchdb.table("todos").delete().eq("id", 1).execute()

# Count without fetching rows
n = stitchdb.table("todos").select("*", count="exact", head=True).execute().count

Errors are raised, not returned:

from stitchdb import PostgrestAPIError

try:
    stitchdb.table("todos").insert({"task": None}).execute()
except PostgrestAPIError as e:
    print(e.code, e.message, e.details, e.hint)   # e.g. 23502 null value in column "task"

Each response returns at most your plan's rows per request (1,000 on Free, 5,000 on Pro), so page with .range(). For deep pages, filter on an indexed column instead of a big offset.

#FastAPI: act as the user

For an API that serves your web or mobile frontend, keep authorization in Postgres. Create a client per request with the anon key and set the caller's session, so auth.uid() in your policies is that user:

import os
from fastapi import FastAPI, Header, HTTPException
from stitchdb import create_client

app = FastAPI()
URL = os.environ["STITCHDB_URL"]
ANON = os.environ["STITCHDB_ANON_KEY"]

def client_for(access_token: str, refresh_token: str):
    db = create_client(URL, ANON)
    db.auth.set_session(access_token, refresh_token)
    return db

@app.get("/todos")
def list_todos(authorization: str = Header(...), x_refresh_token: str = Header(...)):
    if not authorization.startswith("Bearer "):
        raise HTTPException(401, "Sign in first")
    db = client_for(authorization.removeprefix("Bearer "), x_refresh_token)
    return db.table("todos").select("id, task, is_complete").execute().data

With a policy like this on the table, the endpoint can only ever return the caller's rows, even if your Python code has a bug in its filters:

create policy "Users manage their own todos" on public.todos
  for all to authenticated
  using (user_id = auth.uid())
  with check (user_id = auth.uid());

For async apps, use the async client. Every method is the same, with await, and Realtime is only available there:

from stitchdb import acreate_client

stitchdb = await acreate_client(url, key)
rows = (await stitchdb.table("todos").select("*").execute()).data

#Admin work with the service key

Keep one service client for jobs that are not tied to a user:

admin = create_client(url, os.environ["STITCHDB_SERVICE_ROLE_KEY"])
admin.auth.admin.create_user({"email": "[email protected]", "password": "...", "email_confirm": True})
admin.auth.admin.list_users()

# Server side, with the service key
msg = admin.push.send(user_id=user_id, title="Your order shipped", data={"order": 42})

The service_role key bypasses every policy. Load it from the environment, never log it, and never send it to a browser.

#Direct Postgres with psycopg or SQLAlchemy

When you want SQL, an ORM or Alembic migrations, connect to the pooler. Copy the connection string from Settings → Database:

# Prisma, Drizzle, Knex, SQLAlchemy, ...
DATABASE_URL="postgresql://p_<ref>_owner:<password>@<db-host>:6543/p_<ref>?sslmode=require"

import os
import psycopg

with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
    rows = conn.execute(
        "select id, task from public.todos where is_complete = %s order by id desc limit 5",
        (False,),
    ).fetchall()

from sqlalchemy import create_engine, text

engine = create_engine(os.environ["DATABASE_URL"].replace("postgresql://", "postgresql+psycopg://"), pool_pre_ping=True)
with engine.begin() as conn:
    conn.execute(text("update public.todos set is_complete = true where id = :id"), {"id": 1})

Pick the port with care:

• 6543, transaction mode shares database connections between clients per transaction. Good for web apps with many short requests and for serverless. Session state (SET, LISTEN, temp tables, session advisory locks) does not carry over between transactions.
• 5432, session mode behaves like a direct connection. Use it for Alembic migrations, psql, LISTEN/NOTIFY and long-running workers.

TLS is required; a missing sslmode=require shows up as FATAL: no authentication method is found. Startup options such as options=-c search_path=... in the URL are ignored by the pooler, so run SET after connecting. On Pro and above you can restrict which IPv4 addresses may connect (Connect to Postgres).

Direct connections run as the project owner. Row Level Security does not limit them, so this is the path for trusted server code, not for code acting on behalf of an untrusted user.

#Serverless SQL over HTTPS

For AWS Lambda, cron containers or anything that dislikes holding TCP connections, a SQL key runs queries over HTTPS:

from stitchdb import create_sql

sql = create_sql(os.environ["STITCHDB_SQL_URL"])   # https://sdb_sql_...@<host>
rows = sql.query("select * from orders where id = $1", [42])
sql.transaction([
    ("update stock set n = n - 1 where sku = $1", [sku]),
    ("insert into orders (sku) values ($1)", [sku]),
])

A SQL key runs as the project owner, past Row Level Security. Keep it on servers. See the serverless driver.

#Storage, functions and webhooks from Python

bucket = stitchdb.storage.from_("avatars")

with open("avatar.png", "rb") as f:
    bucket.upload("user-1/avatar.png", f.read(), {"content-type": "image/png", "upsert": "true"})

signed = bucket.create_signed_url("user-1/avatar.png", 60)["signedURL"]  # private, 60 s

res = stitchdb.functions.invoke("hello-world", invoke_options={"body": {"name": "Ada"}, "responseType": "json"})

To receive database or Auth events in a Flask or FastAPI endpoint, add a webhook in Studio and check its signature before trusting the body:

from stitchdb import verify_webhook

event = verify_webhook(request_body, request_headers, secret=WEBHOOK_SECRET)   # raises if not genuine

Since 2.1.0 the same client also reaches the app services: push, purchases, feature flags, analytics, crash reporting, deep links, search, offline sync, background jobs and usage metering. The full list is on the Python SDK page.

#Which should you use?

• Building an API for your own frontend: SDK with the user's session. Policies do the authorization.
• Data pipelines, reports, Django or SQLAlchemy models: direct connection on 6543 (or 5432 for migrations).
• Scheduled scripts in short-lived containers: the serverless SQL key or the SDK with the service key.
• AI agents writing SQL: neither of the above. Give them an agent key with a firewall and undo; see letting AI agents write to your database safely.

#FAQ

#Is the stitchdb package compatible with supabase-py code?

StitchDB speaks the same APIs as the open-source Supabase stack, and its compatibility notes say client libraries for that stack in Python work by changing the URL and keys. The stitchdb package has the same style of methods and adds StitchDB's app services.

#Can I run Django on it?

Yes. Django connects like to any Postgres. Use the session-mode port for migrate, and keep in mind that transaction mode does not keep session state between transactions.

#How do I get realtime events in Python?

Use the async client: channel.on_postgres_changes("INSERT", callback, schema="public", table="todos"), then await channel.subscribe(). The table must be in the realtime publication.

#Where should long-running Python jobs run?

On your own worker or container. For work that fits in TypeScript, StitchDB's background jobs run durable, multi-step Edge Functions next to the database.

Get a Postgres database with an API for your Python code: create a free project.