1
┌─────────────────────────────────────────┐
2
│           客户端（浏览器/Postman）         │
3
└─────────────────────────────────────────┘
4
                    ↓ HTTP 请求
5
┌─────────────────────────────────────────┐
6
│              FastAPI 应用层              │
7
│  ┌─────────┐ ┌─────────┐ ┌─────────┐   │
8
│  │ 路由    │ │ 依赖注入 │ │ 响应处理 │   │
9
│  │ @app.get│ │ Depends │ │ return  │   │
10
│  └────┬────┘ └────┬────┘ └─────────┘   │
11
└───────┼───────────┼─────────────────────┘
12
        │           │
13
        │    ┌──────┘
14
        │    ↓
15
        │ ┌─────────────────┐
16
        │ │   get_db()      │  ← 依赖提供者
17
        │ │   yield db      │     创建/管理数据库会话
18
        │ └────────┬────────┘
19
        │          ↓
20
        │ ┌─────────────────┐
21
        └→│   Session       │  ← 数据库会话
22
          │   (SQLAlchemy)  │     执行具体 SQL 操作
23
          └────────┬────────┘
24
                   ↓
25
          ┌─────────────────┐
26
          │   SQLite 数据库  │  ← 数据持久化存储
27
          │   .my_database.db│
28
          └─────────────────┘

1
# ==================== 数据库配置部分 ====================
2
from sqlalchemy import create_engine, Column, Integer, String, Boolean
3
from sqlalchemy.ext.declarative import declarative_base
4
from sqlalchemy.orm import sessionmaker, Session
5

6
# 创建引擎、会话工厂、基类
7

8
# ==================== ORM 模型定义 ====================
9
class DBTodo(Base):
10
    # 定义数据表结构
11

12
# ==================== FastAPI 应用 ====================
13
from fastapi import FastAPI, Depends
14
from pydantic import BaseModel
15

16
app = FastAPI()
17

18
# Pydantic 模型（API 数据校验）
19
class TodoItem(BaseModel):
20
    ...
21

22
# 依赖注入函数
23
def get_db():
24
    ...
25

26
# ==================== API 路由 ====================
27
@app.post("/todos")     # 增
28
@app.get("/todos")      # 查
29
@app.delete(...)        # 删
30

31
# ==================== 启动 ====================
32
if __name__ == "__main__":
33
    uvicorn.run(app, ...)

1
from sqlalchemy import create_engine
2

3
DATABASE_URL = "sqlite:///.my_database.db"
4

5
engine = create_engine(
6
    DATABASE_URL,
7
    connect_args={"check_same_thread": False}
8
)

1
from sqlalchemy.orm import sessionmaker
2

3
SessionLocal = sessionmaker(
4
    autocommit=False,   # 不自动提交事务
5
    autoflush=False,    # 不自动刷新会话
6
    bind=engine         # 绑定到引擎
7
)

1
会话工厂（SessionLocal）
2
    ↓ 每次调用 SessionLocal()
3
创建新的会话实例（Session）
4
    ↓ 用于
5
执行数据库操作（增删改查）
6
    ↓ 完成后
7
提交（commit）或回滚（rollback）
8
    ↓ 最后
9
关闭会话（close）

1
from sqlalchemy.ext.declarative import declarative_base
2

3
Base = declarative_base()

1
Base.metadata.create_all(bind=engine)

1
from sqlalchemy import Column, Integer, String, Boolean
2

3
class DBTodo(Base):
4
    __tablename__ = "todos"
5

6
    id = Column(Integer, primary_key=True, index=True)
7
    title = Column(String, index=True)
8
    is_done = Column(Boolean, default=False)

1
CREATE TABLE todos (
2
    id INTEGER PRIMARY KEY AUTOINCREMENT,
3
    title VARCHAR,
4
    is_done BOOLEAN DEFAULT 0
5
);
6

7
CREATE INDEX ix_todos_id ON todos (id);
8
CREATE INDEX ix_todos_title ON todos (title);

1
# 创建实例（内存中）
2
todo = DBTodo(title="学习 SQLAlchemy", is_done=False)
3

4
# 添加到会话
5
db.add(todo)
6

7
# 提交到数据库
8
db.commit()
9

10
# 刷新获取生成的主键
11
db.refresh(todo)
12
print(todo.id)  # 1（数据库自动生成的 ID）

1
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey
2
from sqlalchemy.orm import relationship
3

4
class Document(Base):
5
    """完整文档（藏书仓库里的整本书）"""
6
    __tablename__ = "documents"
7

8
    id = Column(Integer, primary_key=True)
9
    title = Column(String(200), nullable=False)
10
    source = Column(String(500))
11
    content = Column(Text, nullable=False)
12
    created_at = Column(DateTime, default=datetime.now)
13

14
    chunks = relationship("DocumentChunk", back_populates="document",
15
                          cascade="all, delete-orphan")
16

17
class DocumentChunk(Base):
18
    """文档切片（索引卡片）"""
19
    __tablename__ = "document_chunks"
20

21
    id = Column(Integer, primary_key=True)
22
    document_id = Column(Integer, ForeignKey("documents.id"), nullable=False)
23
    chunk_index = Column(Integer, nullable=False)
24
    content = Column(Text, nullable=False)
25
    embedding_id = Column(String(100), unique=True)
26

27
    document = relationship("Document", back_populates="chunks")

1
Document (父表)                     DocumentChunk (子表)
2
┌─────────────────┐               ┌──────────────────┐
3
│ id (PK)         │◄──────FK──────│ document_id (FK) │
4
│ title           │               │ chunk_index      │
5
│ source          │               │ content          │
6
│ content         │               │ embedding_id     │
7
│ created_at      │               └──────────────────┘
8
│                 │                        ▲
9
│ chunks ─────────┼─── relationship ───────┘
10
└─────────────────┘        │
11
        ▲                  │
12
        └── relationship ──┘
13
          (双向导航)

1
# 从 Document 访问其所有 chunks
2
doc = session.query(Document).first()
3
for chunk in doc.chunks:           # doc.chunks 自动查询所有关联的 DocumentChunk
4
    print(chunk.content)
5

6
# 从 DocumentChunk 访问其所属的 Document
7
chunk = session.query(DocumentChunk).first()
8
print(chunk.document.title)        # chunk.document 自动查询所属的 Document

1
chunks = relationship("DocumentChunk", back_populates="document",
2
                      cascade="all, delete-orphan")

1
# 删除 Document 时，它的所有 DocumentChunk 也会被自动删除
2
session.delete(doc)
3
session.commit()          # doc 和它的所有 chunks 一起被删除

1
doc = Document(title="Python教程", content="...")
2
chunk1 = DocumentChunk(chunk_index=0, content="第一部分...")
3
chunk2 = DocumentChunk(chunk_index=1, content="第二部分...")
4

5
# 方式1：通过外键手动赋值
6
chunk1.document_id = doc.id
7

8
# 方式2：通过 relationship（更直观、推荐）
9
doc.chunks.append(chunk1)
10
doc.chunks.append(chunk2)
11

12
session.add(doc)
13
session.commit()
14
# SQLAlchemy 自动处理外键的赋值

1
# SQLite（开发/测试）
2
DATABASE_URL = "sqlite:///./app.db"
3

4
# PostgreSQL（生产环境）
5
DATABASE_URL = "postgresql://user:password@localhost/dbname"
6

7
# MySQL
8
DATABASE_URL = "mysql+pymysql://user:password@localhost/dbname"

1
id = Column(Integer, primary_key=True)

1
content = Column(Text)

1
document_id = Column(Integer, ForeignKey("documents.id"))
2
chunks = relationship("DocumentChunk", back_populates="document")

1
@app.post("/todos")
2
def create_todo(todo_item: TodoItem):
3
    # ❌ 每个函数都要重复这些代码
4
    db = SessionLocal()     # 创建连接
5
    try:
6
        db_item = DBTodo(...)
7
        db.add(db_item)
8
        db.commit()
9
        return {...}
10
    finally:
11
        db.close()          # 关闭连接
12

13
@app.get("/todos")
14
def read_todos():
15
    # ❌ 又重复一遍
16
    db = SessionLocal()
17
    try:
18
        todos = db.query(DBTodo).all()
19
        return {...}
20
    finally:
21
        db.close()

1
def get_db_return():
2
    db = SessionLocal()     # 创建连接
3
    return db               # 返回连接，函数结束
4
    db.close()              # ❌ 永远不会执行！
5

6
# 使用
7
@app.post("/todos")
8
def create_todo(db: Session = Depends(get_db_return)):
9
    db.add(...)             # 使用连接
10
    db.commit()
11
    # 函数结束，连接永远不会关闭！
12
    # 多次请求后，数据库连接池耗尽，系统崩溃！

1
def get_db():
2
    db = SessionLocal()     # 创建连接
3
    try:
4
        yield db            # 返回连接，但函数暂停（不结束）
5
    finally:
6
        db.close()          # 使用完毕后，从这里继续，关闭连接

1
客户端发送 POST /todos 请求
2
        ↓
3
┌─────────────────────────────────────────┐
4
│  FastAPI 接收到请求                      │
5
│  发现 create_todo 有 Depends(get_db)    │
6
└─────────────────────────────────────────┘
7
        ↓
8
┌─────────────────────────────────────────┐
9
│  调用 get_db() 函数                      │
10
│                                         │
11
│  db = SessionLocal()                    │
12
│       ↓                                 │
13
│  创建数据库连接                          │
14
│       ↓                                 │
15
│  yield db                               │
16
│       ↓                                 │
17
│  【暂停】返回 db 给路由函数              │
18
│  函数状态被保存，等待后续继续            │
19
└─────────────────────────────────────────┘
20
        ↓
21
┌─────────────────────────────────────────┐
22
│  执行 create_todo 函数体                 │
23
│                                         │
24
│  db_item = DBTodo(...)                  │
25
│  db.add(db_item)                        │
26
│  db.commit()                            │
27
│  db.refresh(db_item)                    │
28
│  return {"message": "存到硬盘了！"}       │
29
└─────────────────────────────────────────┘
30
        ↓
31
┌─────────────────────────────────────────┐
32
│  FastAPI 检测到路由函数执行完毕          │
33
│  回到 get_db() 的 yield 暂停点           │
34
│       ↓                                 │
35
│  执行 finally 块                         │
36
│       ↓                                 │
37
│  db.close()                             │
38
│       ↓                                 │
39
│  数据库连接关闭                          │
40
└─────────────────────────────────────────┘
41
        ↓
42
┌─────────────────────────────────────────┐
43
│  返回 HTTP 响应给客户端                  │
44
│  {"message": "存到硬盘了！"}              │
45
└─────────────────────────────────────────┘

1
def get_db():
2
    db = SessionLocal()
3
    try:
4
        yield db
5
        # 如果路由函数正常完成，会回到这里
6
    except Exception:
7
        # 如果路由函数抛出异常，也会捕获
8
        db.rollback()       # 回滚事务
9
        raise               # 重新抛出异常
10
    finally:
11
        db.close()          # ✅ 无论如何都会关闭连接

1
@app.post("/todos")
2
def create_todo(todo_item: TodoItem, db: Session = Depends(get_db)):
3
    # 1. 类型转换：Pydantic → SQLAlchemy
4
    db_item = DBTodo(
5
        title=todo_item.title,
6
        is_done=todo_item.is_done
7
    )
8

9
    # 2. 添加到会话（暂存区）
10
    db.add(db_item)
11

12
    # 3. 提交事务（真正写入数据库）
13
    db.commit()
14

15
    # 4. 刷新对象（获取数据库生成的 ID）
16
    db.refresh(db_item)
17

18
    return {"message": "存到硬盘了！", "data": db_item}

1
Pydantic 对象 (TodoItem)
2
        ↓
3
创建 SQLAlchemy 对象 (DBTodo)
4
        ↓
5
db.add(db_item)     → 添加到会话（内存中）
6
        ↓
7
db.commit()         → 执行 INSERT SQL
8
        ↓
9
db.refresh(db_item) → 获取数据库生成的 id
10
        ↓
11
返回 JSON 响应

1
@app.get("/todos")
2
def read_todos(db: Session = Depends(get_db)):
3
    # 查询所有记录
4
    todos = db.query(DBTodo).all()
5
    return {"data": todos}

1
# 查询所有
2
db.query(DBTodo).all()
3

4
# 查询第一个
5
db.query(DBTodo).first()
6

7
# 按条件查询
8
db.query(DBTodo).filter(DBTodo.is_done == True).all()
9

10
# 按 ID 查询
11
db.query(DBTodo).filter(DBTodo.id == todo_id).first()
12

13
# 链式调用
14
db.query(DBTodo)\
15
  .filter(DBTodo.is_done == False)\
16
  .order_by(DBTodo.id.desc())\
17
  .limit(10)\
18
  .all()

1
@app.delete("/todos/{todo_id}")
2
def delete_todo(todo_id: int, db: Session = Depends(get_db)):
3
    # 1. 查询要删除的记录
4
    todo = db.query(DBTodo).filter(DBTodo.id == todo_id).first()
5

6
    if todo:
7
        # 2. 标记删除
8
        db.delete(todo)
9
        # 3. 提交事务
10
        db.commit()
11
        return {"message": "删除成功"}
12

13
    return {"message": "删除失败"}

1
DELETE /todos/1
2
        ↓
3
查询 id=1 的记录
4
        ↓
5
找到？
6
  ├─ 是 → db.delete(todo) → db.commit() → 返回成功
7
  └─ 否 → 返回失败

1
# 1. 创建引擎
2
engine = create_engine(DATABASE_URL, ...)
3

4
# 2. 创建会话工厂
5
SessionLocal = sessionmaker(...)
6

7
# 3. 定义 ORM 模型
8
class DBTodo(Base): ...
9

10
# 4. 创建数据表
11
Base.metadata.create_all(bind=engine)
12
# 执行 SQL: CREATE TABLE IF NOT EXISTS todos (...)
13

14
# 5. 创建 FastAPI 应用
15
app = FastAPI()
16

17
# 6. 注册路由
18
@app.post("/todos")
19
@app.get("/todos")
20
@app.delete("/todos/{todo_id}")
21

22
# 7. 启动服务器
23
uvicorn.run(app, host="0.0.0.0", port=8000)

1
┌─────────────────────────────────────────────────────────────┐
2
│  阶段 1：接收请求                                             │
3
├─────────────────────────────────────────────────────────────┤
4
│  客户端: POST /todos                                          │
5
│  Body: {"title": "学习 FastAPI", "is_done": false}            │
6
└─────────────────────────────────────────────────────────────┘
7
                              ↓
8
┌─────────────────────────────────────────────────────────────┐
9
│  阶段 2：解析请求体                                           │
10
├─────────────────────────────────────────────────────────────┤
11
│  FastAPI 根据 TodoItem 模型校验数据                           │
12
│  - title: "学习 FastAPI" (str ✅)                            │
13
│  - is_done: false (bool ✅)                                  │
14
│  创建 Pydantic 对象: todo_item = TodoItem(title=..., ...)     │
15
└─────────────────────────────────────────────────────────────┘
16
                              ↓
17
┌─────────────────────────────────────────────────────────────┐
18
│  阶段 3：依赖注入                                             │
19
├─────────────────────────────────────────────────────────────┤
20
│  检测到 db: Session = Depends(get_db)                       │
21
│  调用 get_db()                                               │
22
│    ├─ db = SessionLocal() → 创建 Session 对象               │
23
│    ├─ yield db → 返回 Session，暂停 get_db()                │
24
│    └─ 等待路由函数执行完毕                                   │
25
└─────────────────────────────────────────────────────────────┘
26
                              ↓
27
┌─────────────────────────────────────────────────────────────┐
28
│  阶段 4：执行业务逻辑                                         │
29
├─────────────────────────────────────────────────────────────┤
30
│  1. db_item = DBTodo(title="学习 FastAPI", is_done=False)   │
31
│     → 创建 SQLAlchemy 对象（仅内存中）                        │
32
│                                                              │
33
│  2. db.add(db_item)                                          │
34
│     → 添加到会话的暂存区（pending）                           │
35
│                                                              │
36
│  3. db.commit()                                              │
37
│     → 执行 INSERT INTO todos (title, is_done) VALUES (...)  │
38
│     → 数据写入 SQLite 数据库文件                              │
39
│                                                              │
40
│  4. db.refresh(db_item)                                      │
41
│     → 从数据库获取生成的 id（如：1）                          │
42
│     → db_item.id = 1                                         │
43
│                                                              │
44
│  5. return {"message": "存到硬盘了！", "data": db_item}        │
45
└─────────────────────────────────────────────────────────────┘
46
                              ↓
47
┌─────────────────────────────────────────────────────────────┐
48
│  阶段 5：资源清理                                             │
49
├─────────────────────────────────────────────────────────────┤
50
│  路由函数执行完毕，回到 get_db() 的 yield 处                  │
51
│  执行 finally 块: db.close()                                 │
52
│  → 关闭数据库连接                                             │
53
└─────────────────────────────────────────────────────────────┘
54
                              ↓
55
┌─────────────────────────────────────────────────────────────┐
56
│  阶段 6：返回响应                                             │
57
├─────────────────────────────────────────────────────────────┤
58
│  FastAPI 将 return 的字典转换为 JSON                         │
59
│  返回给客户端:                                                │
60
│  {"message": "存到硬盘了！", "data": {"id": 1, "title": ...}}  │
61
└─────────────────────────────────────────────────────────────┘

1
# 引擎 - 数据库连接池管理
2
engine = create_engine(DATABASE_URL)
3

4
# 会话工厂 - 创建会话的工厂
5
SessionLocal = sessionmaker(bind=engine)
6

7
# 基类 - ORM 模型的基类
8
Base = declarative_base()
9

10
# 模型定义
11
class Model(Base):
12
    __tablename__ = "表名"
13
    id = Column(Integer, primary_key=True)
14
    name = Column(String)
15

16
# 创建表
17
Base.metadata.create_all(bind=engine)

1
def get_db():
2
    db = SessionLocal()
3
    try:
4
        yield db
5
    finally:
6
        db.close()
7

8
@app.get("/items")
9
def read_items(db: Session = Depends(get_db)):
10
    return db.query(Model).all()

1
# 增
2
db.add(instance)
3
db.commit()
4
db.refresh(instance)
5

6
# 查
7
db.query(Model).all()           # 所有
8
db.query(Model).first()         # 第一个
9
db.query(Model).filter(...)     # 条件过滤
10
db.query(Model).get(id)         # 按主键查
11

12
# 改
13
instance.field = new_value
14
db.commit()
15

16
# 删
17
db.delete(instance)
18
db.commit()