Getting started with Python

Superlayer offers a Python Admin SDK with an API that mirrors our JavaScript Admin SDK. If you read the JS docs, you can mentally substitute Python syntax and write working code.

#Install

uv add superlayer
# or
pip install superlayer

Python 3.10+ is required.

To scaffold a new project:

npx @superlayer/create-app --python my-app
cd my-app
uv sync

This creates a starter script with a .env, main.py, superlayer.schema.ts, and superlayer.perms.ts. Schema and permission files are included so you can manage your app with the Superlayer CLI.

#Usage

Our Python SDK mirrors the JS admin SDK's surface and behavior, with a few Pythonic adjustments. A basic example looks like this:

from superlayer import Superlayer, id
 
db = Superlayer(
    # You can pass these explicitly; by default they fall back to
    # SUPERLAYER_APP_ID and SUPERLAYER_APP_ADMIN_TOKEN environment variables
    app_id="__APP_ID__",
    admin_token="__ADMIN_TOKEN__",
)
 
# Write data
goal_id = id()
db.transact(db.tx.goals[goal_id].update({"title": "Get fit"}))
 
# Read data
result = db.query({"goals": {}})
for goal in result["goals"]:
    print(goal["title"])

Exposing your app_id is fine, but the admin_token bypasses permission checks. Keep it server-side and regenerate it from the dashboard if it leaks.

#Async usage

Use AsyncSuperlayer and await each call:

import asyncio
from superlayer import AsyncSuperlayer, id
 
db = AsyncSuperlayer()
 
async def main():
    goal_id = id()
    await db.transact(db.tx.goals[goal_id].update({"title": "Get fit"}))
 
    result = await db.query({"goals": {}})
    for goal in result["goals"]:
        print(goal["title"])
 
asyncio.run(main())

subscribe_query and streams are only available on AsyncSuperlayer. Everything else works on both the sync and async client.

#Schema and permissions

The Python starter includes superlayer.schema.ts and superlayer.perms.ts so you can manage your app from your Python project. Push changes with the Superlayer CLI as you would normally:

npx @superlayer/cli push

#FastAPI with Pydantic

At the moment the SDK doesn't generate Python types from your schema file. Query results are returned as dictionaries. If you want typed objects, define them in your application and validate the returned data there.

Here's an example of how you can use Pydantic models in a FastAPI app:

from fastapi import FastAPI
from pydantic import BaseModel
from superlayer import AsyncSuperlayer, id
 
db = AsyncSuperlayer()
app = FastAPI()
 
class TodoIn(BaseModel):
    text: str
    done: bool = False
 
class Todo(TodoIn):
    id: str
 
@app.post("/todos", response_model=Todo)
async def create_todo(todo: TodoIn):
    todo_id = id()
    await db.transact(db.tx.todos[todo_id].update(todo.model_dump()))
    return Todo(id=todo_id, **todo.model_dump())

See Modeling data for the full schema reference.

#Reading data

db.query takes the same SuperQL dict shape as our JS SDK:

# Top-level fetch
result = db.query({"goals": {}, "todos": {}})
 
# Nested children
result = db.query({"goals": {"todos": {}}})
 
# Operators
db.query({"goals": {"$": {"where": {"title": {"$ne": "Get fit"}}}}})
 
# Pagination + ordering
db.query({"goals": {"$": {"limit": 10, "order": {"createdAt": "desc"}}}})

#Rule params

Use rule_params keyword argument to pass values for permission rules.

db.query({"goals": {}}, rule_params={"region": "us"})

#Writing data

Mutations use the same SuperML proxy syntax.

from superlayer import Superlayer, id, lookup
 
db = Superlayer()
 
goal_id = id()
todo_id = id()
 
db.transact([
    db.tx.goals[goal_id].update({"title": "Get fit"}),
    db.tx.todos[todo_id].update({"title": "Run"}),
    db.tx.goals[goal_id].link({"todos": todo_id}),
    db.tx.users[lookup("email", "alyssa@interfacedb.com")].update(
        {"name": "Alyssa"}
    ),
])

#Special namespaces

System namespaces like $files and $users aren't valid Python attribute names. Use subscript access for those:

db.transact(db.tx["$files"][file_id].delete())

db.tx.goals and db.tx["goals"] are interchangeable for regular namespaces.

See Writing data for the full mutation reference.

#Subscribing to queries

AsyncSuperlayer.subscribe_query opens an SSE stream and yields payloads as the query result changes. This is similar to our JS SDK:

import asyncio
from superlayer import AsyncSuperlayer
 
db = AsyncSuperlayer()
 
async def main():
    async with db.subscribe_query({"todos": {}}) as sub:
        async for payload in sub:
            if payload["type"] == "error":
                print("error:", payload["error"])
                break
            print("data:", payload["data"])
 
asyncio.run(main())

Subscriptions keep a live connection open. Keep the async with block alive as long as you need updates.

#Streams

Streams are also supported in the Python SDK, but only on the AsyncSuperlayer client. Writing and reading streams are similar to the JS SDK, but with Pythonic async iterators and context managers.

#Writing

import asyncio
from superlayer import AsyncSuperlayer
 
db = AsyncSuperlayer()
 
async def main():
    async with db.streams.write(client_id="agent-completion-123") as writer:
        stream_id = await writer.stream_id
 
        async for chunk in claude_response:
            await writer.write(chunk)
 
asyncio.run(main())

#Reading

async with db.streams.read(stream_id="...") as reader:
    async for chunk in reader:
        print(chunk)

You can identify the stream by stream_id= or client_id=. Pass byte_offset= to resume from a specific position after a disconnect.

#Querying stream metadata

The $streams namespace is queryable like any other:

result = db.query({
    "$streams": {
        "$": {"where": {"clientId": "agent-completion-123"}},
    },
})

See Streams for the full reference.

#Logging

Pass verbose=True to AsyncSuperlayer to log the realtime connection lifecycle to the standard library superlayer logger. Pass a logger (any object with debug, info, and error, such as a logging.Logger) to send it elsewhere.

import logging
from superlayer import AsyncSuperlayer
 
logging.basicConfig(level=logging.DEBUG)
 
db = AsyncSuperlayer(verbose=True)
 
# Or send logs to your own logger
db = AsyncSuperlayer(verbose=True, logger=logging.getLogger("myapp"))

Only AsyncSuperlayer has realtime, so the sync Superlayer has no verbose option.

#Auth

Mirrors the JS db.auth namespace.

#Magic codes

db.auth.send_magic_code("alyssa@interfacedb.com")
 
# Or generate the code yourself for a custom email provider
code = db.auth.generate_magic_code("alyssa@interfacedb.com")
print(code)
 
# Verify it
user, created = db.auth.check_magic_code(
    email="alyssa@interfacedb.com",
    code="123456",
)

#Tokens

# Mint a token for a user (creates them if they don't exist)
token = db.auth.create_token(email="alyssa@interfacedb.com")
# or
token = db.auth.create_token(id="...")
 
# Verify a refresh token (e.g. one passed in from a client)
user = db.auth.verify_token(token)

#Users

user = db.auth.get_user(email="alyssa@interfacedb.com")
user = db.auth.get_user(id="...")
user = db.auth.get_user(refresh_token="...")
 
db.auth.delete_user(email="alyssa@interfacedb.com")
db.auth.sign_out(email="alyssa@interfacedb.com")

#Impersonation

You can also use the impersonation API to run queries and transactions as if you were a specific user or guest.

# As a specific user
scoped = db.as_user(email="alyssa@interfacedb.com")
goals_as_alyssa = scoped.query({"goals": {}})
 
# As a guest
guest = db.as_user(guest=True)
guest.query({"publicData": {}})
 
# Or with a refresh token
scoped = db.as_user(token="user-refresh-token")

as_user returns a new immutable client. The original db is unchanged.

#Without an admin token

You can construct a client without an admin token if you only impersonate. Direct queries on the base db will fail, but db.as_user(token=...) and db.as_user(guest=True) will work.

db = Superlayer(app_id="__APP_ID__")  # no admin_token
 
user_db = db.as_user(token="...")
user_db.query({"todos": {}})

#Storage

db.storage.upload_file accepts bytes, pathlib.Path, or a binary file-like object.

from pathlib import Path
 
# From bytes
db.storage.upload_file(
    "photos/hello.txt",
    b"hello world",
    content_type="text/plain",
)
 
# From a Path (opened and closed by the SDK)
db.storage.upload_file(
    "photos/demo.png",
    Path("./demo.png"),
    content_type="image/png",
)
 
# From an open file (uploads from the current position; not closed)
with open("demo.png", "rb") as f:
    db.storage.upload_file("photos/demo.png", f, content_type="image/png")

Content length is computed automatically. Non-seekable streams need an explicit file_size (bytes remaining):

db.storage.upload_file(
    "exports/data.jsonl",
    stream,
    content_type="application/jsonl",
    file_size=10_000_000,
)

To delete a file, use a transaction:

db.transact(db.tx["$files"][file_id].delete())

See Storage for the broader storage model.

#Rooms

Read presence data for a room:

sessions = db.rooms.get_presence("chat", "room-123")
# {peer_id: {"data": ..., "peer-id": ..., "user": ...}}

get_presence is the only rooms method on the admin SDK. See Presence, Cursors, and Activity.

#Debug helpers

debug_query and debug_transact run an operation with a permission check and return why the check passed or failed for every matching row. They require an as_user context.

result = db.as_user(guest=True).debug_query({"goals": {}})
# {"result": ..., "check_results": [{"id": ..., "check": True}, ...]}
 
db.as_user(guest=True).debug_transact(
    db.tx.goals[id()].update({"title": "x"}),
)

#Webhooks

The Python SDK ships the same two-layer webhook surface as the JS SDK, bundled as superlayer.webhooks.

db.webhooks.manager.* for CRUD on subscriptions
db.webhooks.validate / fetch_payloads / process_payload for receiving events

#Managing subscriptions

webhook = db.webhooks.manager.create(
    url="https://api.example.com/superlayer",
    namespaces=["goals", "todos"],
    actions=["create", "update", "delete"],
)
 
# Inspect
db.webhooks.manager.list()
event = db.webhooks.manager.get_event(webhook["id"], isn=42)
payload = db.webhooks.manager.get_payload(webhook["id"], isn=42)
events = db.webhooks.manager.list_events(webhook["id"])
next_page = db.webhooks.manager.list_events(
    webhook["id"], after=events["cursor"]
)
 
# Modify
db.webhooks.manager.update(
    webhook["id"],
    url="https://api.example.com/superlayer-v2",
    namespaces=["goals"],
)
db.webhooks.manager.enable(webhook["id"])
db.webhooks.manager.disable(webhook["id"], reason="Pausing for migration")
db.webhooks.manager.delete(webhook["id"])
db.webhooks.manager.resend_event(webhook["id"], isn=42)

Returned dicts keep camelCase wire-format keys (webhookId, idempotencyKey, etc.) so they match the JS shape.

#Receiving webhooks

Three primitives compose into a framework integration:

# 1. Verify the Ed25519 signature and parse the signed body
webhook_body = db.webhooks.validate(
    signature_header=request.headers["Superlayer-Signature"],
    body=raw_bytes,
    max_age_seconds=300,
)
# Raises SuperlayerError on bad signature or stale timestamp
 
# 2. Exchange the payload reference for the full payload of records
payload = db.webhooks.fetch_payloads(webhook_body)
 
# 3. Dispatch records to handlers (most-specific-wins)
db.webhooks.process_payload(handlers, payload)

Handlers are a dict of dicts keyed by namespace and action, with $default catch-alls at either level:

def on_goal_create(record):
    # record["after"] is the new entity; record["before"] is None on "create"
    save_to_search_index(record["after"])
 
handlers = {
    "goals": {
        "create": on_goal_create,
        "$default": lambda record: None,
    },
    "$default": lambda record: None,
}
 
db.webhooks.process_payload(handlers, payload)

#FastAPI example

from fastapi import FastAPI, Request, HTTPException
from superlayer import AsyncSuperlayer, SuperlayerError
 
app = FastAPI()
db = AsyncSuperlayer()
 
async def on_goal_create(record):
    print("new goal:", record["after"])
 
handlers = {"goals": {"create": on_goal_create}}
 
@app.post("/webhooks/superlayer")
async def webhook_in(request: Request):
    body = await request.body()
    sig = request.headers.get("Superlayer-Signature", "")
    try:
        webhook_body = await db.webhooks.validate(signature_header=sig, body=body)
    except SuperlayerError:
        raise HTTPException(401)
 
    payload = await db.webhooks.fetch_payloads(webhook_body)
    await db.webhooks.process_payload(handlers, payload)
    return {"ok": True}

validate, fetch_payloads, and process_payload are awaitable under AsyncSuperlayer. validate falls back to Superlayer's JWKS endpoint when a signing key is not already cached or bundled. Handlers must match: async def for AsyncSuperlayer, plain def for Superlayer.

See Webhooks for the broader webhook model.

#Errors

All errors raised by the SDK are subclasses of SuperlayerError. The API returns a non-2xx response as SuperlayerAPIError, which carries status and body:

from superlayer import Superlayer, SuperlayerAPIError, id
 
db = Superlayer()
 
try:
    db.transact(db.tx.goals[id()].update({"title": "Get fit"}))
except SuperlayerAPIError as e:
    if e.status == 429:
        print("Rate limited, backing off")
    else:
        raise

Pick your app