数据库与 SQL

前两篇的接口用内存列表存数据——进程一重启数据就没了。真实项目要用数据库。这篇从最基础的开始：连数据库、写 SQL、在 FastAPI 里执行查询。

一、SQLite——开箱即用的数据库

MySQL 要装服务端、配密码、建用户，学习阶段用 SQLite 更省事——它是一个文件，Python 标准库自带驱动，不用装任何东西。

import sqlite3

# 连接一个本地数据库文件（不存在会自动创建）
conn = sqlite3.connect("articles.db")

# 创建一个游标，用它执行 SQL
cursor = conn.cursor()

cursor.execute("""
    CREATE TABLE IF NOT EXISTS articles (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL,
        content TEXT NOT NULL,
        status TEXT DEFAULT 'draft',
        created_at TEXT DEFAULT (datetime('now'))
    )
""")

conn.commit()    # 建表、插入、更新、删除后要 commit
conn.close()

sqlite3.connect("articles.db") 打开（或创建）当前目录下的 articles.db 文件，所有数据都存在这一个文件里。commit() 提交事务——DDL（建表）和 DML（增删改）之后必须 commit 才真正生效。

二、SQL 基础——增删改查

1 插入

cursor.execute(
    "INSERT INTO articles (title, content, status) VALUES (?, ?, ?)",
    ("Python 入门", "Python 基础语法...", "published"),
)
conn.commit()

print(cursor.lastrowid)  # 刚插入这条的自增 ID

参数用 ? 占位，值放在元组里传——绝对不要用 f-string 或字符串拼接拼 SQL。? 占位符是防 SQL 注入的：用户传入的值只会被当成数据，不会被当成 SQL 代码执行。

字符串拼接 SQL 的后果：标题里如果包含 'DROP TABLE articles; --，拼接后整条 SQL 变成恶意删除。用 ? 传参，这种内容只是被当成普通字符串存进数据库。

2 查询

cursor.execute("SELECT id, title, status FROM articles")
rows = cursor.fetchall()    # 取全部结果
print(rows)
# [(1, 'Python 入门', 'published'), (2, 'Go 并发', 'draft')]

cursor.execute("SELECT id, title, status FROM articles WHERE status = ?", ("published",))
rows = cursor.fetchall()

查单条：

cursor.execute("SELECT id, title, content FROM articles WHERE id = ?", (1,))
row = cursor.fetchone()  # 取一条，没有返回 None

3 更新

cursor.execute(
    "UPDATE articles SET status = ? WHERE id = ?",
    ("published", 2),
)
conn.commit()
print(cursor.rowcount)  # 受影响的行数

4 删除

cursor.execute("DELETE FROM articles WHERE id = ?", (1,))
conn.commit()

三、在 FastAPI 里用数据库

每次请求都要连接数据库，用完关闭。FastAPI 用依赖注入管理数据库连接：

from fastapi import FastAPI, Depends
from pydantic import BaseModel
import sqlite3

app = FastAPI()


def get_db():
    """每次请求创建一个连接，请求结束后自动关闭。"""
    conn = sqlite3.connect("articles.db")
    conn.row_factory = sqlite3.Row  # 查询结果按列名取值，不靠下标
    try:
        yield conn
    finally:
        conn.close()


class ArticleCreate(BaseModel):
    title: str
    content: str
    status: str = "draft"


@app.post("/articles")
def create_article(article: ArticleCreate, db: sqlite3.Connection = Depends(get_db)):
    cursor = db.execute(
        "INSERT INTO articles (title, content, status) VALUES (?, ?, ?)",
        (article.title, article.content, article.status),
    )
    db.commit()
    return {"id": cursor.lastrowid, **article.model_dump()}


@app.get("/articles")
def list_articles(db: sqlite3.Connection = Depends(get_db)):
    rows = db.execute("SELECT id, title, status FROM articles ORDER BY id DESC").fetchall()
    return [dict(row) for row in rows]


@app.get("/articles/{article_id}")
def get_article(article_id: int, db: sqlite3.Connection = Depends(get_db)):
    row = db.execute(
        "SELECT id, title, content, status FROM articles WHERE id = ?",
        (article_id,),
    ).fetchone()

    if row is None:
        return {"error": "not found"}

    return dict(row)

Depends(get_db) 是依赖注入——FastAPI 在处理请求前调 get_db() 拿到连接，传给函数参数；请求结束后 finally 里的 conn.close() 自动执行。不用手动开关连接，FastAPI 全管了。

conn.row_factory = sqlite3.Row 让查询结果像字典一样按列名取值——row["title"] 比 row[1] 可读得多。dict(row) 直接转成普通字典。

四、为什么后面要换 ORM

上面的代码能跑，但 SQL 全是裸字符串——改个字段名要找所有 SQL 逐一改，查询条件一复杂 SQL 就又长又难读，而且 SQLite 和 MySQL 的 SQL 方言有差异（自增写法、时间函数、分页语法），换数据库要改一堆 SQL。

ORM（对象关系映射）解决这些问题：用 Python 类描述表结构，用 Python 方法做查询，不用写裸 SQL。下一篇讲 SQLAlchemy——Python 最主流的 ORM。这篇理解了 SQL 的增删改查，再学 ORM 会顺很多，因为 ORM 做的事就是在背后生成这些 SQL。