数据库入门
这是什么:给 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 管理复杂度
动手改
- 给
todos表加created TEXT列存创建时间(time.strftime("%Y-%m-%d %H:%M")) - 写
GET /todos列出全部:SELECT ...配conn.execute(...).fetchall(),把行转成列表返回 - 写
DELETE /todos/{id}:删完返回 200,id 不存在抛 404(第 5 章的活)