🐸 青蛙编程网
首页 / Python / FastAPI / 6. 数据库入门

数据库入门

这是什么:给 FastAPI 接上数据库——用 Python 标准库 sqlite3 操作 SQLite 单文件数据库,把前几章「存内存、重启就没」的数据真正住进硬盘。

为什么用:内存仓库一重启就清空,数据必须落盘才叫产品。数据库教程讲过 SQLite 零安装、单文件、Python 原生支持,是入门后端存储的最短路径。

如何用:启动时建表,路由里读写——本章做出「创建待办 + 列出待办」的落盘闭环,重启服务数据还在。

准备:连接与建表

新建 database.py,把数据库操作集中管理:

import sqlite3

DB_PATH = "todos.db"

def get_conn():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row     # 查询结果按下标也能按列名取
    return conn

def init_db():
    with get_conn() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS todos (
                id      INTEGER PRIMARY KEY AUTOINCREMENT,
                title   TEXT    NOT NULL,
                done    INTEGER NOT NULL DEFAULT 0
            )
        """)
  • sqlite3.connect("todos.db"):没有这个文件会自动创建,数据库就是一个普通文件
  • row_factory = sqlite3.Row:让每行结果既能 row["title"] 按列名取,比裸元组好读
  • CREATE TABLE IF NOT EXISTS:幂等建表,跑多少次都不重复建

启动时建表: lifespan

在 main.py 里让应用「启动时建表、关闭时收尾」——用 lifespan:

from contextlib import asynccontextmanager
from fastapi import FastAPI
import database

@asynccontextmanager
async def lifespan(app: FastAPI):
    database.init_db()             # 启动时执行
    yield                          # ← 服务运行中
    print("服务关闭,可做清理")      # 关闭时执行

app = FastAPI(lifespan=lifespan)

yield 之前是启动逻辑,之后是关闭逻辑。建表放这里,保证服务一起来表就存在,路由里只管读写。

路由读写:真正的落盘 CRUD

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

class TodoIn(BaseModel):
    title: str = Field(min_length=1, max_length=50)

app = FastAPI(lifespan=lifespan)   # 沿用上面的 lifespan

@app.post("/todos", status_code=201)
def create_todo(todo: TodoIn):
    with database.get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO todos (title, done) VALUES (?, ?)",
            (todo.title, 0),
        )
        conn.commit()
        return {"id": cur.lastrowid, "title": todo.title}

@app.get("/todos/{todo_id}")
def read_todo(todo_id: int):
    with database.get_conn() as conn:
        row = conn.execute(
            "SELECT id, title, done FROM todos WHERE id = ?", (todo_id,)
        ).fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="待办不存在")
    return {"id": row["id"], "title": row["title"], "done": bool(row["done"])}

「你会看到」:POST 创建返回自增 id;GET 能查到它;重启服务再 GET,数据还在——落盘完成。

参数化查询:安全红线

注意 SQL 里的 ? 占位符:值放在元组里传给 execute,而不是用 f-string 拼进 SQL。这叫参数化查询,是防 SQL 注入的硬规矩——如果拼接字符串,用户在标题里写一句 ' DROP TABLE todos; -- 就能让整个表人间蒸发。永远 ? 传值,永不拼 SQL。

进阶方向:真实项目多用 SQLModel / SQLAlchemy(ORM,用 Python 类操作表,Flask 教程用的是 SQLAlchemy 系)或异步驱动。本章用标准库 sqlite3 是为了看清「路由 → SQL → 数据」的完整链路,原理懂了,ORM 上手只是换个写法。

记忆口诀:启动 lifespan 建好表,路由 execute 来读写;问号传值防注入,commit 一下才落袋。

❓ 常见问题

Q1:todos.db 文件在哪?

在启动 uvicorn 的那个目录下(工作目录),不是 main.py 所在目录——两者通常相同但不是一回事。想固定位置,把 DB_PATH 写成绝对路径。

Q2:报 sqlite3.OperationalError: no such table: todos?

原因:表还没建(没跑 init_db)就执行了查询。修复三步:① 确认路由用了 lifespan 且里面调了 database.init_db();② 确认启动的是改过的 main.py;③ 或手动跑一次 python -c "import database; database.init_db()" 兜底。

Q3:插入成功但查不到?

漏了 conn.commit()。sqlite3 的写操作要显式提交才落盘(with conn 块退出时会自动提交,但拆开写就容易忘)。增删改后必 commit,背下来。

📝 本章小结

  • sqlite3 标准库三件套:connect 连接、execute 执行、commit 落盘
  • lifespan 把「启动建表 / 关闭清理」挂到应用生命周期上
  • ? 参数化查询是防注入红线,任何拼接 SQL 的写法都不合格
  • 入门用标准库看清原理,进阶换 SQLModel / SQLAlchemy 管理复杂度

动手改

  1. 给 todos 表加 created TEXT 列存创建时间(time.strftime("%Y-%m-%d %H:%M"))
  2. 写 GET /todos 列出全部:SELECT ... 配 conn.execute(...).fetchall(),把行转成列表返回
  3. 写 DELETE /todos/{id}:删完返回 200,id 不存在抛 404(第 5 章的活)

上一章:响应与错误处理 · 下一章:综合练习 🐸

📥 下载本章代码

ZIP 内含本章全部代码文件与说明,解压即用,与教程内容同步生成。