Files
mini/README.md
2026-10-18 20:58:22 +00:00

213 lines
6.0 KiB
Markdown

# 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:
```bash
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
```python
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.
```python
async def h(req):
page = req.param("page", "1")
return Response.json(req.all_params)
```
## Validation
```python
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:
```python
try:
data = v.clean(req.all_params)
except ValueError as e:
return Response.json({"errors": e.args[0]}, 400)
```
## Templates
`./templates/hello.html`:
```html
<h1>Hello {{name}}</h1>
```
```python
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
```python
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
```python
@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.
```python
@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`.
```python
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:
```python
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
```python
app = Mini(version="1.2.3")
```
Use `/v{version}/...` routes. Bump version to bust cache.
## PubSub
```python
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
```json
{"port": 23000, "version": "1.0.0"}
```
Auto-created. `serve()` prefers saved port, else scans 22000-24000.
## Full Example
```python
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()
```