Python + SQLite：将查询逻辑封装成 FastAPI 接口-前端学习

一、核心需求

当一段查询逻辑已经可以在 Python 中跑通，下一步常见需求就是把它封装成一个 FastAPI 接口，让前端通过 HTTP 请求直接调用。

这个改造里有两个关键点不能漏掉：

使用 sqlite3.Row
在接口目录中补上 __init__.py

一个解决返回数据结构问题，一个解决 uvicorn 启动入口问题。

二、为什么要加 sqlite3.Row

sqlite3 默认返回的是 tuple 列表，例如：

[(1, "student_1", 16, 90, 88, 95)]

这种结构不适合直接作为接口返回给前端。前端更容易处理的是带字段名的对象列表。

数据库连接时可以增加这一行：

conn.row_factory = sqlite3.Row

完整写法如下：

def connect_db() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn

这行配置的作用是让每一行查询结果支持按列名访问，后面就可以把结果转成 dict。

cursor = conn.execute(SELECT_SQL)
return [dict(row) for row in cursor.fetchall()]

返回给接口后的数据结构就会变成：

[
  {
    "id": 437,
    "name": "student_436",
    "age": 16,
    "math": 97,
    "english": 96,
    "history": 87
  }
]

这种结构更适合前端直接渲染表格或者列表。

三、排序字段为什么要做白名单校验

接口参数 orderBy 来自前端输入，不能直接拼到 SQL 里。这里适合使用字段白名单：

column_map = {
    "total": "total",
    "math": "math",
    "english": "english",
    "history": "history",
}

if orderBy not in column_map:
    raise ValueError(f"非法的排序字段: {orderBy}")

这个写法有两个作用：

防止传错字段时程序直接崩溃
防止把任意字符串拼进 ORDER BY，降低 SQL 注入风险

四、FastAPI 接口怎么写

一个最小可用的接口结构如下：

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()


@app.get("/api/top-10")
def get_top_10_api(orderBy: str = "total"):
    try:
        conn = connect_db()
        rows = get_top_10(conn, orderBy=orderBy)
        return JSONResponse(content=rows)
    except ValueError as err:
        raise HTTPException(status_code=400, detail=str(err))
    finally:
        conn.close()

这个接口做了几件事：

接收 orderBy 查询参数
调用查询函数读取数据库
把结果以 JSON 返回给前端
捕获非法参数并返回 400
在 finally 中关闭数据库连接

前端调用方式：

fetch("http://127.0.0.1:8010/api/top-10?orderBy=math")
  .then((res) => res.json())
  .then((data) => console.log(data));

五、为什么还要写 init.py

很多人第一次拆分 FastAPI 目录时，都会遇到这个报错：

Error loading ASGI app. Attribute "app" not found in module "api"

原因是运行下面这条命令时：

uvicorn api:app --reload

uvicorn 会去找 api 模块里的 app 变量。

如果接口目录下面只有路由文件，而没有 __init__.py 暴露 app，那么 api:app 就找不到入口。

这里的解决方式是在 api/__init__.py 中补上：

from .request import app

__all__ = ["app"]

这样之后，目录本身就可以作为模块入口，下面这条命令就能工作：

uvicorn api:app --reload

如果不写 __init__.py，就只能这样启动：

uvicorn api.request:app --reload

六、运行方式

推荐使用 Poetry 启动：

uvicorn api:app --reload --port 8010

接口地址示例：

http://127.0.0.1:8010/api/top-10
http://127.0.0.1:8010/api/top-10?orderBy=total
http://127.0.0.1:8010/api/top-10?orderBy=math

七、这种封装解决了什么

查询逻辑不再只是命令行脚本
前端可以通过 HTTP 请求直接拿到数据
sqlite3.Row 让查询结果更适合转成 JSON
__init__.py 让 uvicorn api:app 的入口更清晰
非法排序字段可以稳定返回 400，而不是直接崩溃

Tips

如果接口返回的是 tuple 列表，优先检查是不是忘了写：

conn.row_factory = sqlite3.Row

如果 uvicorn 提示找不到 app，优先检查是不是忘了写：

from .request import app

如果启动时报 8000 端口冲突，可以直接换端口：

poetry run uvicorn api:app --reload --port 8010

目录CONTENT

Python + SQLite：将查询逻辑封装成 FastAPI 接口