Kui Framework

@app.router.http.get("/")
async def handler() -> Annotated[Any, JSONResponse[200, {}, UserModel]]:
    ...

# Status-only: JSONResponse[200]
# Status + headers: JSONResponse[200, {"X-Custom": {"schema": {"type": "string"}}}]
# Status + headers + body schema: JSONResponse[200, {}, UserModel]

@app.router.http.get("/events")
async def events():
    async def generate():
        for i in range(10):
            yield {"id": i, "data": "hello"}
            await asyncio.sleep(1)
    return generate()

from dataclasses import dataclass

@dataclass
class Error:
    code: int
    message: str

app = Kui(
    response_converters={
        Error: lambda e, status=400, headers=None: JSONResponse(
            {"code": e.code, "message": e.message}, status, headers
        ),
    }
)

from kui.asgi import request, websocket

@app.router.http.get("/")
async def handler():
    request.method         # "GET"
    request.url            # Full URL object
    request.headers        # Headers mapping
    request.cookies        # Cookies dict
    request.path_params    # Extracted path params
    request.query_params   # Query params
    request.state          # Per-request State dict
    request.app            # The Kui instance
    await request.json     # Parse JSON body (cached property, no parentheses)
    await request.form     # Parse form body (cached property, no parentheses)
    await request.data()   # Auto-detect body format (method call)

request.state.user = current_user     # write
name = request.state.user.name        # read
del request.state.user                # delete

await websocket.accept()
await websocket.receive_text() / .receive_bytes() / .receive_json()
await websocket.send_text(s) / .send_bytes(b) / .send_json(obj)
await websocket.close(code=1000)

from kui.asgi import OpenAPI, Routes

openapi = OpenAPI(
    info={"title": "My API", "version": "1.0.0"},
    template_name="swagger",  # "swagger" | "redoc" | "rapidoc"
    reload=True,
)
app.router <<= "/docs" // openapi.routes

from typing import Any
from typing_extensions import Annotated
from baize.datastructures import ContentType
from kui.asgi import HttpRequest, FactoryClass, Kui

class MsgPackRequest(HttpRequest):
    async def data(self) -> Annotated[Any, ContentType("application/x-msgpack")]:
        ...

app = Kui(factory_class=FactoryClass(http=MsgPackRequest))

from kui.asgi import HTTPException

# Raise in handlers
raise HTTPException(404)
raise HTTPException(403, content="Forbidden", headers={"X-Reason": "auth"})

# Register handlers
@app.exception_handler(404)
async def not_found(exc):
    return {"error": "not found"}, 404

@app.exception_handler(ValueError)
async def value_error(exc):
    return {"error": str(exc)}, 400

from kui.asgi import Depends, bearer_auth, basic_auth, api_key_auth_dependency

# Bearer token: Authorization: Bearer <token>
@app.router.http.get("/me")
async def me(token: Annotated[str, Depends(bearer_auth)]):
    return decode_jwt(token)

# Basic auth: Authorization: Basic <base64>
@app.router.http.get("/admin")
async def admin(creds: Annotated[tuple[str, str], Depends(basic_auth)]):
    username, password = creds
    ...

# API key from header/query/cookie
api_key = api_key_auth_dependency("X-API-Key", position="header")

@app.router.http.get("/data")
async def data(key: Annotated[str, Depends(api_key)]):
    ...

import re
from kui.asgi import allow_cors, Kui

# App-level
app = Kui(cors_config={
    "allow_origins": [re.compile(r"https://example\.com")],
    "allow_methods": ["GET", "POST"],
    "allow_headers": [],
    "expose_headers": [],
    "allow_credentials": False,
    "max_age": 600,
})

# Per-route
cors = allow_cors(allow_origins=[re.compile(".*")])
app.router <<= HttpRoute("/api", handler) @ cors

async def lifespan(app: Kui):
    app.state.pool = await create_pool()
    yield
    await app.state.pool.close()

app = Kui(lifespan=lifespan)

from kui.asgi import request

@app.router.http.post("/notify")
async def notify():
    request.background_tasks.append(send_email, to="user@example.com")
    return {"status": "queued"}

async def send_email(to: str):
    ...

from kui.asgi import Jinja2Templates, TemplateResponse, request

templates = Jinja2Templates("templates")
app = Kui(templates=templates)

@app.router.http.get("/")
async def homepage():
    return TemplateResponse("index.html", {"request": request, "name": "World"})

from kui.asgi import Kui, FactoryClass, HttpRequest, WebSocket

class CustomRequest(HttpRequest): ...
class CustomWS(WebSocket): ...

app = Kui(factory_class=FactoryClass(http=CustomRequest, websocket=CustomWS))

from datetime import datetime
app = Kui(json_encoder={datetime: lambda dt: dt.isoformat()})

import asyncio
from pathlib import Path as FilePath
from typing_extensions import Annotated

from kui.asgi import (
    Kui, OpenAPI, HttpRoute, SocketRoute,
    Path, Query, Body, Depends,
    HTTPException, HttpView,
    allow_cors, required_method, websocket,
)
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float

items_db: list[Item] = []

async def get_items_db():
    return items_db

app = Kui(
    routes=[
        HttpRoute("/", lambda: "Welcome to Kui!"),
    ],
    http_middlewares=[allow_cors()],
)

@app.router.http.get("/items")
async def list_items(
    skip: Annotated[int, Query(0)],
    limit: Annotated[int, Query(10)],
    db: Annotated[list, Depends(get_items_db)],
):
    return db[skip:skip + limit]

@app.router.http.post("/items")
async def create_item(
    item: Annotated[Item, Body(exclusive=True)],
    db: Annotated[list, Depends(get_items_db)],
):
    db.append(item)
    return item, 201

app.router <<= "/docs" // OpenAPI(template_name="swagger").routes