// Scaffold Zig-native database routes using pg.zig + TurboPG. Use when adding database-backed CRUD endpoints, custom SQL queries (pgvector, JSONB, full-text search, JOINs, CTEs), or standalone TurboPG usage.
Run performance benchmarks for TurboAPI. Use when testing performance, checking for regressions, or comparing against FastAPI.
Build the TurboAPI Zig native backend. Use when the turbonet extension needs to be recompiled, after changing Zig source files, or when seeing "Native core not available".
Use TurboPG for standalone Postgres queries. Use when writing database code outside of TurboAPI routes, running migrations, seeding data, or building scripts that talk to Postgres.
Scaffold a new TurboAPI route with handler, model, and tests. Use when adding a new API endpoint, creating a new route, or scaffolding CRUD for a resource.
Run TurboAPI tests. Use when running tests, checking for regressions, or verifying changes.
| name | turbo-db-route |
| description | Scaffold Zig-native database routes using pg.zig + TurboPG. Use when adding database-backed CRUD endpoints, custom SQL queries (pgvector, JSONB, full-text search, JOINs, CTEs), or standalone TurboPG usage. |
| argument-hint | <table-name> [crud|query|standalone] |
Create database routes that execute entirely in Zig — no Python, no GIL. Supports CRUD auto-generation, custom SQL, and standalone TurboPG usage.
$ARGUMENTS[0] as the table name$ARGUMENTS[1] — crud (default), query for custom SQL, or standalone for TurboPGconfigure_db is called before any db routesfrom turboapi import TurboAPI
from dhi import BaseModel, Field
app = TurboAPI()
app.configure_db("postgres://user:pass@localhost/mydb", pool_size=16)
class User(BaseModel):
name: str = Field(min_length=1, max_length=100)
email: str
age: int = Field(gt=0)
@app.db_get("/users/{user_id}", table="users", pk="id")
def get_user(): pass
@app.db_list("/users", table="users")
def list_users(): pass
@app.db_post("/users", table="users", model=User)
def create_user(): pass
@app.db_delete("/users/{user_id}", table="users", pk="id")
def delete_user(): pass
# Full-text search
@app.db_query("GET", "/search", sql="""
SELECT id, title, ts_rank(tsv, plainto_tsquery('english', $1)) AS rank
FROM articles WHERE tsv @@ plainto_tsquery('english', $1)
ORDER BY rank DESC LIMIT $2
""", params=["q", "limit"])
def search(): pass
# pgvector nearest neighbors
@app.db_query("GET", "/similar/{item_id}", sql="""
SELECT id, name, 1 - (embedding <=> (SELECT embedding FROM items WHERE id = $1)) AS sim
FROM items ORDER BY embedding <=> (SELECT embedding FROM items WHERE id = $1) LIMIT $2
""", params=["item_id", "limit"])
def similar(): pass
# JSONB filter
@app.db_query("GET", "/admins", sql="""
SELECT id, name, metadata->>'plan' AS plan
FROM users WHERE metadata @> '{"role": "admin", "active": true}'
""")
def admins(): pass
# Multi-table JOIN + aggregation
@app.db_query("GET", "/users/{user_id}/stats", sql="""
SELECT u.name, count(DISTINCT p.id) AS posts, sum(o.total) AS revenue
FROM users u LEFT JOIN posts p ON u.id = p.user_id
LEFT JOIN orders o ON u.id = o.user_id WHERE u.id = $1
GROUP BY u.name
""", params=["user_id"], single=True)
def user_stats(): pass
# Array column query
@app.db_query("GET", "/posts/tagged", sql="""
SELECT id, title, tags FROM posts WHERE $1 = ANY(tags) LIMIT 10
""", params=["tag"])
def tagged(): pass
from turbopg import Database
db = Database("postgres://user:pass@localhost/mydb")
users = db.query("SELECT * FROM users WHERE age > $1 LIMIT $2", [18, 10])
user = db.query_one("SELECT * FROM users WHERE id = $1", [42])
affected = db.execute("INSERT INTO users (name, email) VALUES ($1, $2)", ["Alice", "a@b.com"])
with Database("postgres://...") as db:
result = db.query("SELECT count(*) as n FROM users")
host=/var/run/postgresql for ~50% less latencyThe forked pg.zig includes writeJsonRow() which handles every Postgres type:
["a", "b"][1, 2, 3][0.1, 0.2, 0.3]