mini.py - Native Async Python HTTP Server

Single-file, plug-and-play, extremely object-oriented server using only stdlib + asyncio. Modern Python with pathlib and f-strings.

Run:

python mini.py
python mini.py 8000

Data defaults to ./data. Config in ./.env.json (port, version, future settings).

Core Classes

• Mini - app, router, server. app.serve(port=None) auto-picks 22000-24000 remembering last in .env.json.
• Request - method, path, query, headers, cookies, path_params, query_params, form, body, json. req.param(name, default), req.all_params.
• Response - Response.text/html/json/redirect. Return dict -> JSON, str -> text, bytes -> raw.
• Route - compiled {param} patterns, optional auth=True.
• Field / Validator - validation framework.
• Template(dir) - render(name, ctx) with {{key}} escaped.
• DB - async sqlite wrapper.
• Sessions - cookie sid sessions, in-memory cache to 50MB synced to sqlite.
• PubSub - builtin sub/unsub/pub with asyncio.Queue.
• Time - Time.now() timezone-aware UTC.
• Ids - Ids.uuid7().
• Esc - Esc.x(s) html-escape.

Routing

app = Mini()
app.static()  # serve ./static

@app.get("/")
async def idx(req: Request):
    return {"ok": True}

@app.get("/hi/{name}")
def hi(req):  # sync -> run in thread like FastAPI
    return f"hi {req.path_params['name']}"

@app.post("/submit")
async def sub(req):
    return {"got": req.all_params}

app.route("GET", "/x", handler, auth=True)
app.get("/p", handler, auth=False)
app.post("/p", handler)

URL-decoding automatic via urllib.parse. All output escaping via Esc.x in templates.

GET/POST/JSON

query_params from querystring, form from application/x-www-form-urlencoded and multipart (basic), json_data from application/json auto-decoded. param() checks path > query > form > json.

async def h(req):
    page = req.param("page", "1")
    return Response.json(req.all_params)

Validation

v = Validator({
  "email": Field(required=True, regex=r"^[^@]+@[^@]+$", max_len=200),
  "age": Field(type_=int, default=18),
  "name": Field(required=True, min_len=2, max_len=50),
})
clean = v.clean(req.all_params)

Field(required, regex, type_, default, min_len, max_len). Raises ValueError(dict) on fail. Returns 400 automatically if you call in handler? Catch yourself:

try:
    data = v.clean(req.all_params)
except ValueError as e:
    return Response.json({"errors": e.args[0]}, 400)

Templates

./templates/hello.html:

<h1>Hello {{name}}</h1>

tpl = Template("templates")
@app.get("/t")
async def t(req):
    return tpl.render("hello.html", {"name": req.param("name","world")})

Auto-escaped.

Static + Versioned URLs + Mime + Video Streaming

app.static(prefix="/static", dir="./static")
app.static(prefix="/v{version}/static", dir="./static")  # header caching, bump version

Mime via mimetypes. Range: bytes= supported for video streaming. Path-traversal blocked via Path.resolve() check. Cache-Control: max-age=31536000 on versioned.

Base64 Auth

@app.get("/priv", auth=True)
async def priv(req):
    return {"user": req.headers.get("authorization")}

Set auth=True -> requires Authorization: Basic base64(user:pass). Returns 401 otherwise. Check req.auth_user, req.auth_pass (set by Mini).

Sessions

Automatic cookie sid. Access via req.session dict (loaded), save automatically at end of request.

@app.get("/count")
async def c(req):
    req.session["n"] = req.session.get("n",0)+1
    return req.session

Cached in memory to 50MB, synced to sessions table in sqlite file.

SQLite + Auto CRUD

File ./data/app.db. Single-writer lock (asyncio.Lock), concurrent readers. All datetimes timezone-aware ISO strings.

Tables auto-created with: id INTEGER PK AUTOINCREMENT, uid TEXT(uuid7), created_at TEXT, updated_at TEXT, deleted_at TEXT

Missing columns auto-added then retry (up to 6 tries). Multi-field WHERE auto-indexed (CREATE INDEX IF NOT EXISTS), cached in self._idx.

db = app.db
await db.insert("users", name="bob", email="b@x.com")
await db.update("users", {"email":"b@x.com"}, {"name":"bobby"})
await db.upsert("users", {"email":"b@x.com"}, name="bob")
await db.delete("users", {"email":"b@x.com"})  # soft-delete
await db.find("users", {"name":"bob"})
await db.insert_many("users", [{"name":"a"},{"name":"b"}])
await db.update_many("users", [({"name":"a"}, {"name":"a2"})])
await db.upsert_many("users", [({"name":"a"}, {"name":"a2"})])

_later variants queue and return preview instantly:

await db.insert_later("users", name="fast")
await db.update_later("users", {"name":"a"}, {"name":"b"})
# also: upsert_later, delete_later, insert_many_later, upsert_many_later, update_many_later

Background drain() task flushes queue.

Version / Header Caching

app = Mini(version="1.2.3")

Use /v{version}/... routes. Bump version to bust cache.

PubSub

q = asyncio.Queue()
app.pubsub.sub("chat", q)
await app.pubsub.pub("chat", {"hi":1})
msg = await q.get()

Sync handlers in threads

If handler is not async, executed via run_in_executor like FastAPI.

.env.json

{"port": 23000, "version": "1.0.0"}

Auto-created. serve() prefers saved port, else scans 22000-24000.

Full Example

from mini import Mini, Field, Validator

app = Mini()
app.static()
v = Validator({"name": Field(required=True, min_len=2)})

@app.get("/")
async def idx(req):
    return {"ok": True, "v": app.version}

@app.post("/users")
async def create(req):
    try:
        d = v.clean(req.all_params)
    except ValueError as e:
        from mini import Response
        return Response.json({"errors": e.args[0]}, 400)
    row = await app.db.insert("users", **d)
    return row

@app.get("/users/{uid}")
async def one(req):
    rows = await app.db.find("users", {"uid": req.path_params["uid"]})
    return rows[0] if rows else ({"error":"not found"}, 404)

app.serve()