swadhin / justapi public
/ justapi / benchmarks / workloads_crud_pg.py
swadhinbiswas p0: production hardening sweep — 19 fixes (WASM OOM, mutex poisoning, GIL pool deadlock, envelope detection, SQLite pragmas, traversal, credentials, etc.) · 7/24/2026
f06ae7c
28 lines · 1.0 KB
Raw Download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
"""Step D CRUD benchmark against Postgres (Rust-native path, no GIL)."""
import sys
from justapi import JustAPIApp, Database

PORT = sys.argv[1] if len(sys.argv) > 1 else "8099"
# Pass via CLI arg or JUSTAPI_PG_URL env var — never commit live credentials
import os
PGURL = sys.argv[2] if len(sys.argv) > 2 else os.environ.get("JUSTAPI_PG_URL", "")

INIT = (
    "DROP TABLE IF EXISTS justapi_bench_items;"
    "CREATE TABLE justapi_bench_items ("
    "id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL, qty INTEGER NOT NULL);"
)

app = JustAPIApp()
app.set_database(
    Database(PGURL, max_connections=20, init_sql=INIT),
)
COLUMNS = ["name", "qty"]
app.post("/items", crud_table="justapi_bench_items", crud_columns=COLUMNS)
app.get("/items/{id}", crud_table="justapi_bench_items", crud_columns=COLUMNS)
app.put("/items/{id}", crud_table="justapi_bench_items", crud_columns=COLUMNS)
app.delete("/items/{id}", crud_table="justapi_bench_items", crud_columns=COLUMNS)

if __name__ == "__main__":
    app.run(f"127.0.0.1:{PORT}")