用 FastAPI + SQLite 快速搭建内部工具后端

内部工具对高并发的要求不高，但对开发速度和可维护性要求很高。 FastAPI + SQLite 这个组合几乎零配置，不需要安装数据库服务，部署只需要复制一个文件夹。本文记录我在实际项目中用到的几个关键设计。

为什么选 SQLite

单文件，备份就是 copy cps_data.db backup.db
不需要数据库服务进程，Windows Server 上少一个依赖
内部工具并发量极低，SQLite 的写锁完全够用
迁移简单：启动时检查并 ALTER TABLE 加字段

项目结构

project/
├── main.py          # FastAPI 入口，init_db() 在启动时执行迁移
├── database.py      # SQLite 连接池 + 迁移逻辑
├── routers/
│   ├── auth.py      # 登录 / 登出 / 鉴权依赖
│   └── data.py      # 业务 CRUD
└── templates/
    ├── index.html   # 前端页面（原生 HTML/JS）
    └── public.html  # 无需登录的公开页

数据库连接

用 threading.local() 让每个线程持有独立连接，避免 SQLite "same thread" 问题的同时保持简单：

import sqlite3, threading

_local = threading.local()

def get_conn() -> sqlite3.Connection:
    if not hasattr(_local, 'conn'):
        _local.conn = sqlite3.connect(DB_PATH, check_same_thread=False)
        _local.conn.row_factory = sqlite3.Row
    return _local.conn

启动时自动迁移

用版本号控制迁移，每次启动自动执行未跑过的步骤，不需要 Alembic 这类重型工具：

MIGRATIONS = [
    # v1.0.1
    "ALTER TABLE daily_data ADD COLUMN is_locked INTEGER DEFAULT 0",
    # v1.0.2
    "ALTER TABLE users ADD COLUMN display_name TEXT",
]

def init_db():
    conn = get_conn()
    conn.execute(
        "CREATE TABLE IF NOT EXISTS _migrations (version INTEGER PRIMARY KEY)"
    )
    done = {r[0] for r in conn.execute("SELECT version FROM _migrations")}
    for i, sql in enumerate(MIGRATIONS, start=1):
        if i not in done:
            try:
                conn.execute(sql)
                conn.execute("INSERT INTO _migrations VALUES (?)", (i,))
                conn.commit()
            except Exception as e:
                print(f"Migration {i} failed: {e}")

会话认证

不引入 JWT 库，用 PBKDF2 + secrets.token_urlsafe() 实现简单可靠的会话管理：

import hashlib, secrets

def hash_password(pwd: str) -> str:
    salt = secrets.token_hex(16)
    h = hashlib.pbkdf2_hmac('sha256', pwd.encode(), salt.encode(), 100_000)
    return f"{salt}:{h.hex()}"

def verify_password(pwd: str, stored: str) -> bool:
    salt, hashed = stored.split(':', 1)
    h = hashlib.pbkdf2_hmac('sha256', pwd.encode(), salt.encode(), 100_000)
    return secrets.compare_digest(h.hex(), hashed)

def create_session(user_id: int) -> str:
    token = secrets.token_urlsafe(32)
    expires = datetime.now() + timedelta(days=7)
    get_conn().execute(
        "INSERT INTO sessions(user_id, token, expires_at) VALUES(?,?,?)",
        (user_id, token, expires.strftime('%Y-%m-%d %H:%M:%S'))
    )
    get_conn().commit()
    return token

FastAPI 鉴权依赖

from fastapi import Cookie, HTTPException

def require_login(session_token: str = Cookie(None)):
    if not session_token:
        raise HTTPException(401)
    row = get_conn().execute(
        "SELECT u.* FROM sessions s JOIN users u ON u.id=s.user_id "
        "WHERE s.token=? AND s.expires_at > datetime('now','localtime')",
        (session_token,)
    ).fetchone()
    if not row:
        raise HTTPException(401)
    return dict(row)

# 路由中使用
@router.get("/api/data")
def get_data(user = Depends(require_login)):
    ...

小结

这套方案在实际运营了半年以上，稳定无故障。唯一需要注意的是 SQLite 写操作不支持真正并发，如果你的工具有批量写入的需求，考虑加 timeout 参数或升级到 PostgreSQL。