An async Python backend toolkit. One client, one config file.
ORM · Auth · Payments · Storage · KV · Queues · AI · CLI · Studio
Warning: Derp is in alpha. The API is unstable and may change without notice before 1.0.
uv add derp-pyRequires Python 3.12+.
Define a table:
from derp.orm import Table, Field, Fn, UUID, Varchar, Integer, Boolean, TimestampTZ
class Product(Table, table="products"):
id: UUID = Field(primary=True, default=Fn.gen_random_uuid())
name: Varchar[255] = Field()
price_cents: Integer = Field()
is_active: Boolean = Field(default=True)
created_at: TimestampTZ = Field(default=Fn.now())Generate and apply a migration:
derp generate --name initial
derp migrateQuery data:
from derp import DerpClient, DerpConfig
from app.models import Product
config = DerpConfig.load("derp.toml")
derp = DerpClient(config)
await derp.connect()
# Select
products = await (
derp.db.select(Product)
.where(Product.is_active)
.order_by(Product.created_at, asc=False)
.limit(10)
.execute()
)
# Insert
product = await (
derp.db.insert(Product)
.values(name="Headphones", price_cents=4999)
.returning(Product)
.execute()
)
# Update
await (
derp.db.update(Product)
.set(price_cents=3999)
.where(Product.id == product.id)
.execute()
)from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from fastapi import FastAPI, Request, Depends
from derp import DerpClient, DerpConfig
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
config = DerpConfig.load("derp.toml")
derp = DerpClient(config)
await derp.connect()
app.state.derp = derp
yield
await derp.disconnect()
app = FastAPI(lifespan=lifespan)
def get_derp(request: Request) -> DerpClient:
return request.app.state.derp
@app.get("/products")
async def list_products(derp: DerpClient = Depends(get_derp)):
return await derp.db.select(Product).where(Product.is_active).execute()Everything lives in derp.toml. Only [database] is required — add modules as you need them:
[database]
db_url = "$DATABASE_URL"
schema_path = "app/models.py"
[auth.native]
enable_signup = true
[auth.native.jwt]
secret = "$JWT_SECRET"
[storage]
endpoint_url = "$S3_ENDPOINT"
public_urls = { assets = "https://assets.example.com" }
access_key_id = "$S3_KEY"
secret_access_key = "$S3_SECRET"
[kv.valkey]
addresses = [["localhost", 6379]]
[payments]
api_key = "$STRIPE_SECRET_KEY"
[queue.celery]
broker_url = "$CELERY_BROKER_URL"
[ai]
api_key = "$OPENAI_API_KEY"
# base_url = "https://api.openrouter.ai/v1" # for other providersEnvironment variables starting with $ are resolved at load time.
Email/password, magic links, Google/GitHub OAuth, JWTs, organizations. Native, Supabase, WorkOS, or Google Cloud Identity Platform (GCIP) backend — exactly one, selected by config.
user, tokens = await derp.auth.sign_up(email="alice@example.com", password="s3cure!")
session = await derp.auth.authenticate(request) # from Bearer token
org = await derp.auth.create_org(name="Acme", slug="acme", creator_id=user.id)Google Cloud Identity Platform (GCIP) is a hosted IdP backend (a peer of
Supabase/WorkOS). It verifies Google-issued RS256 ID tokens locally, runs
user-facing flows through the API-key Identity Toolkit endpoints, and user
administration through a service-account-signed token. GCIP sends its own
magic-link and password-reset emails, so — unlike the native backend — it does
not require an [email] block. It has no organization concept, so the org
methods raise NotImplementedError.
[auth.gcip]
project_id = "$GCIP_PROJECT_ID"
api_key = "$GCIP_API_KEY" # GCIP Web API key
service_account_json = "$GCIP_SERVICE_ACCOUNT_JSON" # full SA key JSON
# redirect_uri = "https://yourapp.com/callback"customer = await derp.payments.create_customer(email="buyer@example.com")
session = await derp.payments.create_checkout_session(
mode="payment",
line_items=[{"price_id": "price_xxx", "quantity": 1}],
success_url="https://example.com/success",
cancel_url="https://example.com/cancel",
)
event = await derp.payments.verify_webhook_event(payload=body, signature=sig)await derp.storage.upload_file(bucket="assets", key="avatar.jpg", data=img, content_type="image/jpeg")
data = await derp.storage.fetch_file(bucket="assets", key="avatar.jpg")
# Signed URLs for direct client access
url = await derp.storage.signed_download_url(bucket="assets", key="avatar.jpg")
url = await derp.storage.signed_upload_url(bucket="assets", key="uploads/new.jpg", content_type="image/jpeg")
# Batch delete and server-side copy
await derp.storage.delete_files(bucket="assets", keys=["tmp/a.txt", "tmp/b.txt"])
await derp.storage.copy_file(src_bucket="uploads", src_key="tmp.jpg", dst_bucket="assets", dst_key="final.jpg")await derp.kv.set(b"user:123", b'{"name":"Alice"}', ttl=3600)
data = await derp.kv.get(b"user:123")
# Idempotent endpoints
body, status, is_replay = await derp.kv.idempotent_execute(
key=idem_key, compute=lambda: create_order(data), status_code=201,
)
# Webhook dedup
if await derp.kv.already_processed(event_id=event["id"]):
return {"status": "duplicate"}
# Rate limiting
result = await derp.kv.rate_limit(f"api:{user.id}", limit=100, window=3600)
if not result.allowed:
raise HTTPException(429, headers={"Retry-After": str(result.retry_after)})# Chat
response = await derp.ai.chat(model="gpt-4o-mini", messages=[{"role": "user", "content": "Hello"}])
print(response.content)
# Streaming with Vercel AI SDK format
async for chunk in derp.ai.stream_chat(model="gpt-4o-mini", messages=messages):
for event in chunk.vercel_ai_json(message_id="msg-1"):
yield event.dump() # "data: {...}\n\n"
# Image generation via fal (submit + poll + get in one call)
result = await derp.ai.fal_call("fal-ai/flux", inputs={"prompt": "a cat"})task_id = await derp.queue.enqueue("send_email", payload={"user_id": str(user.id)})
status = await derp.queue.get_status(task_id)Schedules in config:
[[queue.schedules]]
name = "cleanup"
task = "cleanup_sessions"
cron = "0 */6 * * *"derp init Create derp.toml
derp generate Generate migration from schema diff
derp migrate Apply pending migrations
derp push Push schema directly (dev only)
derp pull Introspect database into snapshot
derp status Show migration status
derp check Verify schema matches snapshot (CI)
derp drop Remove migration files
derp studio Launch database browser UI
derp version Show version
Full docs at derp.readthedocs.io.
uv sync
uv run pytest
uv run ruff check src/
uv run ruff format src/MIT