Docs
Sign in
On this page

Python

stitchdb: the Python client for StitchDB (sync and async) — database, auth, storage, realtime, Edge Functions, errors and every app service: push, purchases, flags, jobs, metering, SQL and more.

#Install

Python 3.9 or newer.

Terminal
pip install stitchdb

The package is stitchdb on PyPI. Pin a version in requirements.txt, e.g. stitchdb==2.1.0. Also with uv add stitchdb or poetry add stitchdb.

#Create a client

Python
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

Async apps (FastAPI, Starlette, asyncio scripts) use the async client; every method is the same, with await. Realtime is only in the async client.

Python
from stitchdb import acreate_client

stitchdb = await acreate_client(url, key)
rows = (await stitchdb.table("todos").select("*").execute()).data
Python
from stitchdb import create_client, ClientOptions

stitchdb = create_client(url, key, options=ClientOptions(
    schema="public",            # default schema for table()
    auto_refresh_token=True,
    persist_session=True,
    headers={"x-my-header": "1"},
    postgrest_client_timeout=10,
))

#Database

Python
# 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

# One row (raises if there isn't exactly one) or None
todo = stitchdb.table("todos").select("*").eq("id", 1).single().execute().data
maybe = stitchdb.table("todos").select("*").eq("id", 1).maybe_single().execute()

# Insert, update, upsert, delete (the changed rows are returned)
stitchdb.table("todos").insert({"task": "Ship it"}).execute()
stitchdb.table("todos").insert([{"task": "a"}, {"task": "b"}]).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()

# Call a Postgres function
total = stitchdb.rpc("count_open_todos", {"owner": user_id}).execute().data

# Count without fetching rows
n = stitchdb.table("todos").select("*", count="exact", head=True).execute().count
Filters
Compareeq, neq, gt, gte, lt, lte, is_
Textlike, ilike, text_search
Lists and JSONin_, contains, contained_by, overlaps
Combineor_("status.eq.open,priority.gt.3"), not_.eq(...), match({...}), filter(column, operator, value)
Modifiersorder, limit, range, single, maybe_single, csv

Use another schema with stitchdb.schema("api").table("orders"). Filter syntax is the same as the REST API.

#Auth

Python
# Sign up and sign in
stitchdb.auth.sign_up({"email": "[email protected]", "password": "a-strong-password"})
res = stitchdb.auth.sign_in_with_password({"email": "[email protected]", "password": "a-strong-password"})
print(res.user.id, res.session.access_token)

# Magic link / one-time code by e-mail
stitchdb.auth.sign_in_with_otp({"email": "[email protected]", "options": {"email_redirect_to": "https://example.com/welcome"}})
stitchdb.auth.verify_otp({"email": "[email protected]", "token": "123456", "type": "email"})

# OAuth: send the user to res.url, then exchange the code from the redirect
res = stitchdb.auth.sign_in_with_oauth({"provider": "github", "options": {"redirect_to": "https://example.com/callback"}})
stitchdb.auth.exchange_code_for_session({"auth_code": code})

# Session and user
session = stitchdb.auth.get_session()
user = stitchdb.auth.get_user().user
stitchdb.auth.update_user({"data": {"display_name": "Ada"}})
stitchdb.auth.reset_password_for_email("[email protected]", {"redirect_to": "https://example.com/reset"})
stitchdb.auth.sign_out()

# React to sign-in, sign-out and token refresh
stitchdb.auth.on_auth_state_change(lambda event, session: print(event))

On a server that acts for many users, create a client per request and set the user's token with stitchdb.auth.set_session(access_token, refresh_token), or keep one service client for admin work:

Python
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()
admin.auth.admin.delete_user(user_id)

#Storage

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"})

data = bucket.download("user-1/avatar.png")           # bytes
url = bucket.get_public_url("user-1/avatar.png")       # public buckets
signed = bucket.create_signed_url("user-1/avatar.png", 60)["signedURL"]  # private, 60 s
files = bucket.list("user-1")
bucket.move("user-1/avatar.png", "user-1/old.png")
bucket.remove(["user-1/old.png"])

# Buckets (service_role key)
admin.storage.create_bucket("avatars", options={"public": True})
admin.storage.list_buckets()
admin.storage.delete_bucket("avatars")

#Realtime

Realtime needs the async client. Callbacks run on the event loop.

Python
stitchdb = await acreate_client(url, key)

# Database changes (only rows the user may read, see Row Level Security)
changes = stitchdb.channel("todos-changes")
changes.on_postgres_changes("INSERT", lambda payload: print(payload), schema="public", table="todos")
await changes.subscribe()

# Broadcast
room = stitchdb.channel("room-1", {"config": {"broadcast": {"self": True}}})
room.on_broadcast("cursor", lambda message: print(message))
await room.subscribe()
await room.send_broadcast("cursor", {"x": 10, "y": 20})

# Presence
room.on_presence_sync(lambda: print(room.presence_state()))
await room.track({"user": "ada", "online_at": "2026-10-24T10:00:00Z"})

await stitchdb.remove_all_channels()

Postgres changes need the table in the realtime publication; see Postgres changes.

#Edge Functions

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

Without responseType the raw response body (bytes) is returned. The signed-in user's token is sent with the call.

#Errors

Python
from stitchdb import PostgrestAPIError, AuthApiError, StorageException, FunctionsHttpError

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"

try:
    stitchdb.auth.sign_in_with_password({"email": email, "password": "wrong"})
except AuthApiError as e:
    print(e.status, e.code, e.message)            # 400 invalid_credentials

Database, Auth, Storage and Functions errors are raised as exceptions (PostgrestAPIError, AuthApiError and the other AuthError classes, StorageException, FunctionsHttpError / FunctionsRelayError). See Errors for the status codes.

#App services

Since 2.1.0 the same client reaches every StitchDB app service, in both the sync and the async client (with await). Turn each service on in Studio first. Errors raise a subclass of StitchDBAppsError with status, code and message. In these examples admin is a client created with the service_role key.

#Push notifications

Python
stitchdb.push.register({"platform": "android", "token": fcm_token, "topics": ["news"]})
stitchdb.push.subscribe("sale")
inbox = stitchdb.push.inbox(unread=True)      # {"notifications": [...], "unread": 3}
stitchdb.push.mark_read("all")

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

Methods: register, unregister, subscribe, unsubscribe, inbox, mark_read, delete_notifications, config, send and status. The async client also has on_notification(callback) for new inbox items. See Push notifications.

#In-app purchases

Python
stitchdb.purchases.sync("apple", signed_transaction=jws)
stitchdb.purchases.sync("google", purchase_token=token, product_id="pro_monthly")
if stitchdb.purchases.has_entitlement("pro"):
    unlock_pro()

Methods: sync, me, has_entitlement, config, and with the service key refresh(user_id). See In-app purchases.

#Feature flags

Python
stitchdb.flags.load(attributes={"app_version": "2.4.1"})
if stitchdb.flags.is_on("new-checkout"):
    ...
limit = stitchdb.flags.get("upload-limit-mb", 10)
off = stitchdb.flags.on("change", lambda keys: print(keys))

# Server side: flags for any user
admin.flags.evaluate({"id": user_id}, attributes={"plan": "pro"})

Methods: load, refresh, set_attributes, get, is_on, detail, all, on("change"), stop, evaluate. Flags refresh every minute and on sign-in and sign-out. See Feature flags.

#Analytics and crash reporting

Python
stitchdb.analytics.start(release="2.4.1")
stitchdb.analytics.track("purchase", {"plan": "pro"})
stitchdb.analytics.flush()

stitchdb.crashes.start(release="2.4.1", environment="production")   # uncaught exceptions are reported
try:
    risky()
except Exception as e:
    stitchdb.crashes.capture_exception(e)

Analytics: start, track, flush, stop, opt_out, opt_in, and the reports overview, trend, funnel, retention, user. Crashes: start (hooks sys.excepthook and threading.excepthook), capture_exception, capture_message, add_breadcrumb, flush, issues, update_issue, releases, upload_sourcemap. See Analytics and Crash reporting.

Python
link = stitchdb.links.resolve("https://go.example.com/l/summer")   # path and data of the link
made = admin.links.create(path="/invite/7", data={"ref": "ada"}, title="Join me")
stats = admin.links.get("summer", days=30)

Methods: resolve, deferred, create, list, get (stats), update, delete. See Deep links.

Python
res = stitchdb.search("articles", "postgres backups", limit=10, mode="hybrid")
for row in res["data"]:
    print(row["title"])

Returns data, scores, mode and, when the search fell back, note. Modes: hybrid, keyword, semantic. See Search.

#Offline sync

Python
offline = stitchdb.offline(["todos"], path="todos.json")
offline.sync()
todos = offline.table("todos")
todos.upsert({"id": 1, "task": "Ship it"})   # queued, sent on the next sync
offline.on("conflict", lambda c: print(c))

A local copy with queued writes. Events: status, conflict, change. See Offline sync.

#Organizations

Python
org = stitchdb.orgs.create("Acme")
stitchdb.orgs.invite(org["id"], "[email protected]", role="admin")
stitchdb.orgs.set_active(org["id"])   # the session's org_id / org_role claims

Methods: create, update, delete, list, members, invite, invites, revoke_invite, accept_invite, set_role, remove_member, leave, set_active, transfer_ownership. See Organizations.

#Realtime message history

Python
stitchdb.history.publish("room-1", "message", {"text": "hi"})   # stored, then broadcast
page = stitchdb.history.list("room-1", limit=50)                 # oldest first

# Async client: deliver what a channel missed to its on_broadcast listeners
n = await stitchdb.history.replay(room, since=last_seen)

Replayed messages have meta.replayed set to True. See Message history.

#Background jobs and job monitoring

Python
run = admin.jobs.start("send-invoice", {"order": 42}, idempotency_key="order-42")
admin.jobs.get(run["id"])
admin.jobs.logs(run["id"])

# At the end of a cron script
stitchdb.heartbeat.ping(os.environ["HEARTBEAT_URL"])           # or "fail", "start", an exit code

Jobs: start, get, list, logs, cancel, retry. Heartbeat: ping, monitors, monitor. See Background jobs and Job monitoring.

#Usage metering

Python
admin.metering.record("api_calls", 1, customer="acme", idempotency_key=request_id)
admin.metering.usage(customer="acme", group="month")
csv = admin.metering.export(format="csv")

Methods: record (one event or a list), usage, customers, export, set_stripe_customer, stripe_customers, alerts. See Usage metering.

#Collaborative documents

Python
docs = stitchdb.collab.documents()
doc = stitchdb.collab.get("plan-2026")          # stored updates
stitchdb.collab.push_update("plan-2026", update_bytes)
stitchdb.collab.create_snapshot("plan-2026", label="Before review")

Over HTTP, for servers and scripts: documents, get, push_update, members, add_member, remove_member, snapshots, create_snapshot, snapshot. Updates are Yjs updates. See Collaborative documents.

#GraphQL

Python
res = stitchdb.graphql("query($n: Int) { todosCollection(first: $n) { edges { node { id } } } }", {"n": 10})
print(res["data"], res.get("errors"))

Pass raise_on_errors=True to raise GraphQLError when the answer has errors. See GraphQL.

#Status page

Python
status = stitchdb.status_page.status()
stitchdb.status_page.subscribe("[email protected]")

See Status page.

#Serverless SQL

Python
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]),
])

The async version is create_async_sql. Statements are strings, (text, params) tuples or {"query", "params"} dicts. See Serverless driver.

#Webhooks and agent keys

Python
from stitchdb import verify_webhook, AgentClient

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

agent = AgentClient(os.environ["STITCHDB_AGENT_KEY"])   # sdb_agt_ keys for AI agents
agent.sql("select count(*) from orders", session="task-1")

verify_webhook checks the StitchDB-Signature header and the timestamp. AgentClient (and AsyncAgentClient) also has approval, session, revert_session and temporary database logins. See Webhooks and Agent keys.

#More sign-in methods

Python
stitchdb.auth.sign_in_with_web3("ethereum", message, signature)
options = stitchdb.auth.passkey.start_authentication()
stitchdb.auth.passkey.verify_authentication(options["challenge_id"], credential)
codes = stitchdb.auth.mfa.recovery_codes.generate()

Also link_identity_with_id_token, auth.oauth (OAuth server consent: authorization details, approve, deny, grants), auth.admin.passkey and auth.admin.custom_providers. See Web3 and Passkeys.

Was this helpful?