Initial template: Vue 3 + FastAPI + SQLite full-stack with K8s deployment

Extracted from oil project — business logic removed, auth/db/deploy infrastructure
generalized with APP_NAME placeholders.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

backend/auth.py

@@ -0,0 +1,37 @@
+from fastapi import Request, Depends, HTTPException
+
+from backend.database import get_db
+
+ANON_USER = {"id": None, "role": "viewer", "username": "anonymous", "display_name": "匿名"}
+
+
+def get_current_user(request: Request):
+    """Extract user from Bearer token. Returns anonymous if no/invalid token."""
+    token = request.headers.get("Authorization", "").removeprefix("Bearer ").strip()
+    if not token:
+        return ANON_USER
+    conn = get_db()
+    user = conn.execute(
+        "SELECT id, username, role, display_name, password FROM users WHERE token = ?",
+        (token,),
+    ).fetchone()
+    conn.close()
+    if not user:
+        return ANON_USER
+    return dict(user)
+
+
+def require_role(*roles):
+    """Dependency that checks the user has one of the given roles."""
+    def checker(user=Depends(get_current_user)):
+        if user["role"] not in roles:
+            raise HTTPException(403, "权限不足")
+        return user
+    return checker
+
+
+def require_login(user=Depends(get_current_user)):
+    """Dependency that requires any authenticated user."""
+    if user["id"] is None:
+        raise HTTPException(401, "请先登录")
+    return user

backend/database.py

@@ -0,0 +1,62 @@
+import sqlite3
+import os
+import secrets
+
+DB_PATH = os.environ.get("DB_PATH", "/data/app.db")
+
+
+def get_db():
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA journal_mode=WAL")
+    conn.execute("PRAGMA foreign_keys=ON")
+    return conn
+
+
+def init_db():
+    """Initialize database schema. Add your tables here."""
+    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
+    conn = get_db()
+    c = conn.cursor()
+    c.executescript("""
+        CREATE TABLE IF NOT EXISTS users (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            username TEXT UNIQUE NOT NULL,
+            password TEXT,
+            token TEXT UNIQUE NOT NULL,
+            role TEXT NOT NULL DEFAULT 'viewer',
+            display_name TEXT,
+            created_at TEXT DEFAULT (datetime('now'))
+        );
+        CREATE TABLE IF NOT EXISTS audit_log (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            user_id INTEGER,
+            action TEXT NOT NULL,
+            target_type TEXT,
+            target_id TEXT,
+            target_name TEXT,
+            detail TEXT,
+            created_at TEXT DEFAULT (datetime('now'))
+        );
+    """)
+
+    # Seed admin user if no users exist
+    count = c.execute("SELECT COUNT(*) FROM users").fetchone()[0]
+    if count == 0:
+        admin_token = os.environ.get("ADMIN_TOKEN", secrets.token_hex(24))
+        c.execute(
+            "INSERT INTO users (username, token, role, display_name) VALUES (?, ?, ?, ?)",
+            ("admin", admin_token, "admin", "Admin"),
+        )
+        print(f"[INIT] Admin user created. Token: {admin_token}")
+
+    conn.commit()
+    conn.close()
+
+
+def log_audit(conn, user_id, action, target_type=None, target_id=None, target_name=None, detail=None):
+    conn.execute(
+        "INSERT INTO audit_log (user_id, action, target_type, target_id, target_name, detail) "
+        "VALUES (?, ?, ?, ?, ?, ?)",
+        (user_id, action, target_type, str(target_id) if target_id else None, target_name, detail),
+    )

backend/main.py

@@ -0,0 +1,115 @@
+from fastapi import FastAPI, HTTPException, Depends
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel
+from typing import Optional
+import hashlib
+import secrets
+import os
+import threading
+import time as _time
+
+from backend.database import get_db, init_db, log_audit
+from backend.auth import get_current_user, require_role, require_login
+
+app = FastAPI(title="App API")
+
+
+# ── Periodic WAL checkpoint ───────────────────────────
+def _wal_checkpoint_loop():
+    while True:
+        _time.sleep(300)
+        try:
+            conn = get_db()
+            conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+            conn.close()
+        except:
+            pass
+
+threading.Thread(target=_wal_checkpoint_loop, daemon=True).start()
+
+
+# ── Models ────────────────────────────────────────────
+class LoginRequest(BaseModel):
+    username: str
+    password: str
+
+class RegisterRequest(BaseModel):
+    username: str
+    password: str
+    display_name: str = ""
+
+
+# ── Health & Version ──────────────────────────────────
+APP_VERSION = "0.1.0"
+
+@app.get("/api/health")
+def health():
+    return {"status": "ok"}
+
+@app.get("/api/version")
+def version():
+    return {"version": APP_VERSION}
+
+
+# ── Auth endpoints ────────────────────────────────────
+def _hash_password(pw: str) -> str:
+    return hashlib.sha256(pw.encode()).hexdigest()
+
+@app.get("/api/me")
+def get_me(user=Depends(get_current_user)):
+    return {
+        "id": user.get("id"),
+        "username": user["username"],
+        "role": user["role"],
+        "display_name": user.get("display_name", ""),
+    }
+
+@app.post("/api/login")
+def login(body: LoginRequest):
+    conn = get_db()
+    user = conn.execute(
+        "SELECT id, username, token, password, role, display_name FROM users WHERE username = ?",
+        (body.username,),
+    ).fetchone()
+    conn.close()
+    if not user:
+        raise HTTPException(401, "用户名或密码错误")
+    user = dict(user)
+    if user.get("password") and user["password"] != _hash_password(body.password):
+        raise HTTPException(401, "用户名或密码错误")
+    return {"token": user["token"]}
+
+@app.post("/api/register", status_code=201)
+def register(body: RegisterRequest):
+    conn = get_db()
+    existing = conn.execute("SELECT id FROM users WHERE username = ?", (body.username,)).fetchone()
+    if existing:
+        conn.close()
+        raise HTTPException(409, "用户名已存在")
+    token = secrets.token_hex(24)
+    pw_hash = _hash_password(body.password)
+    conn.execute(
+        "INSERT INTO users (username, password, token, role, display_name) VALUES (?, ?, ?, ?, ?)",
+        (body.username, pw_hash, token, "viewer", body.display_name or body.username),
+    )
+    conn.commit()
+    conn.close()
+    return {"token": token}
+
+
+# ── User management (admin) ──────────────────────────
+@app.get("/api/users")
+def list_users(user=Depends(require_role("admin"))):
+    conn = get_db()
+    rows = conn.execute("SELECT id, username, role, display_name, token, created_at FROM users ORDER BY id").fetchall()
+    conn.close()
+    return [dict(r) for r in rows]
+
+
+# ── Static files (frontend) ──────────────────────────
+@app.on_event("startup")
+def on_startup():
+    init_db()
+    frontend_dir = os.environ.get("FRONTEND_DIR", "frontend/dist")
+    if os.path.isdir(frontend_dir):
+        app.mount("/", StaticFiles(directory=frontend_dir, html=True), name="frontend")

backend/requirements.txt

@@ -0,0 +1,3 @@
+fastapi==0.115.6
+uvicorn[standard]==0.34.0
+aiosqlite==0.20.0