JSON APIs

uv run stk new <module> writes a complete, working set of these endpoints in stk/<module>/views.py. Start from that file; this page explains its conventions.

List endpoints

Lists paginate manually and serialize with orjson. The response shape is {items, total, perPage}, which is what the frontend’s server-paginated table expects.

import orjson as json
from quart import Response, g, request
from sqlalchemy import func, or_, select

PER_PAGE = 25

@bp.get("/api/posts")
async def list_posts():
    page = request.args.get("page", 1, type=int)
    per_page = request.args.get("per_page", PER_PAGE, type=int)

    query = select(Post)
    if search := request.args.get("search"):
        query = query.where(or_(Post.title.ilike(f"%{search}%"), Post.content.ilike(f"%{search}%")))

    total = await g.db_session.scalar(select(func.count()).select_from(query.subquery()))
    result = await g.db_session.execute(query.offset((page - 1) * per_page).limit(per_page))
    items = [p.to_dict() for p in result.scalars().all()]
    return Response(
        json.dumps({"items": items, "total": total, "perPage": per_page}),
        content_type="application/json",
    )

Count from query.subquery() so filters apply to the total too.

Mutations

The frontend sends {item: {...}}. Create is POST /api/post/, update is POST /api/post/<id> (POST, not PUT), delete is DELETE /api/post/<id>. Admin mutations are logged with Activity.register.

@bp.post("/api/post/")
async def create_post():
    data = (await request.json).get("item", {})
    if not data.get("title"):
        return {"message": "Title is required"}, 400
    post = await Post().from_dict(data)
    g.db_session.add(post)
    try:
        await g.db_session.flush()  # assigns post.id for the audit entry
        await Activity.register(current_user.id, "Post Create", post.to_dict())
        await g.db_session.commit()
        return {"message": "Post created"}
    except Exception:
        await g.db_session.rollback()
        log.exception("Error creating post")
        return {"message": "Error creating post"}, 412

Responses are {"message": ...}: 400 for invalid input, 404 for a missing row, 412 for a failed write. An unhandled exception reaches the global error handler in app.py, which rolls back the session and returns JSON for API requests.

Protecting endpoints

Put auth on the blueprint so no route can forget it:

@bp.before_request
@auth_required("session")
@roles_required("admin")
async def before_request():
    pass

uv run stk inspect routes --json reports which routes are guarded, and stk verify fails if a route reported as guarded is reachable anonymously.

CSRF and webhooks

Every mutation requires a CSRF token, which Axios sends from the shared layout token. A third-party webhook without a session opts out with @csrf_exempt from stk.security, placed under the route decorator, and must authenticate the caller itself, for example by checking an HMAC signature. See Authentication.