1
# main.py - 屎山版本（不要这样写！）
2
from fastapi import FastAPI, Depends
3
from pydantic import BaseModel
4
from sqlalchemy import create_engine, Column, Integer, String, Boolean
5
from sqlalchemy.orm import sessionmaker, Session, declarative_base
6
import uvicorn
7

8
# ========== 数据库配置（30行）==========
9
DATABASE_URL = "sqlite:///./my_database.db"
10
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
11
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
12
Base = declarative_base()
13

14
# ========== 模型定义（20行）==========
15
class DBTodo(Base):
16
    __tablename__ = "todos"
17
    id = Column(Integer, primary_key=True, index=True)
18
    title = Column(String, index=True)
19
    is_done = Column(Boolean, default=False)
20

21
Base.metadata.create_all(bind=engine)
22

23
# ========== Pydantic模型（10行）==========
24
class TodoItem(BaseModel):
25
    title: str
26
    is_done: bool = False
27

28
# ========== 依赖注入（10行）==========
29
def get_db():
30
    db = SessionLocal()
31
    try:
32
        yield db
33
    finally:
34
        db.close()
35

36
# ========== 路由（100+行）==========
37
app = FastAPI()
38

39
@app.post("/todos/")
40
def create_todo(item: TodoItem, db: Session = Depends(get_db)):
41
    ...
42

43
@app.get("/todos/")
44
def get_todos(db: Session = Depends(get_db)):
45
    ...
46

47
# ... 还有更新、删除、以及其他功能的路由 ...
48
# 文件长度：300+ 行！
49

50
if __name__ == "__main__":
51
    uvicorn.run(app, host="127.0.0.1", port=8000)

1
🏢 公司组织架构                    💻 代码分层架构
2

3
    总经理                         main.py
4
      │                          (应用入口)
5
      │                              │
6
    部门经理                        routers.py
7
   /    \                         (业务逻辑)
8
 销售   技术                          │
9
  │      │                            │
10
 客户   程序员                      models.py
11
 经理    经理                       (数据模型)
12
  │      │                            │
13
 客户   数据库                      database.py
14
 代表    管理员                     (数据访问)

1
┌─────────────────────────────────────────┐
2
│           表现层 (Presentation)          │
3
│              main.py                    │
4
│         应用入口、路由注册                │
5
├─────────────────────────────────────────┤
6
│           业务层 (Business)              │
7
│             routers.py                  │
8
│      API接口、业务逻辑、数据校验           │
9
├─────────────────────────────────────────┤
10
│           模型层 (Model)                 │
11
│             models.py                   │
12
│      ORM模型、数据库表结构定义             │
13
├─────────────────────────────────────────┤
14
│           数据层 (Data)                  │
15
│            database.py                  │
16
│      数据库连接、会话管理                 │
17
└─────────────────────────────────────────┘

1
from sqlalchemy import create_engine
2
from sqlalchemy.orm import sessionmaker
3

4
DATABASE_URL = "sqlite:///./my_database.db"
5
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
6
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

1
场景1：换数据库
2
    只需要修改 database.py 中的 DATABASE_URL
3
    models.py 和 routers.py 完全不用动！
4

5
场景2：改连接池配置
6
    只需要在 database.py 中加参数
7
    其他文件不受影响！

1
from sqlalchemy import Column, Integer, String, Boolean
2
from sqlalchemy.orm import declarative_base
3
from database import engine  # 从数据层导入引擎
4

5
Base = declarative_base()
6

7
class DBTodo(Base):
8
    __tablename__ = "todos"
9
    id = Column(Integer, primary_key=True, index=True)
10
    title = Column(String, index=True)
11
    is_done = Column(Boolean, default=False)
12

13
# 建表动作只执行一次
14
Base.metadata.create_all(bind=engine)

1
场景：新增一个 User 表
2
    只需要在 models.py 加：
3
    class User(Base): ...
4

5
    routers.py 可以导入 User 使用
6
    database.py 完全不用改！

1
from fastapi import APIRouter, Depends
2
from pydantic import BaseModel
3
from sqlalchemy.orm import Session
4
from database import SessionLocal  # 从数据层导入
5
from models import DBTodo         # 从模型层导入
6

7
# 创建路由器
8
router = APIRouter(prefix="/todos", tags=["Todos"])
9

10
class TodoItem(BaseModel):
11
    title: str
12
    is_done: bool = False
13

14
def get_db():
15
    db = SessionLocal()
16
    try:
17
        yield db
18
    finally:
19
        db.close()
20

21
# 增删改查路由...
22
@router.post("/")
23
@router.get("/")
24
@router.put("/{todo_id}")
25
@router.delete("/{todo_id}")

1
from fastapi import FastAPI
2
import uvicorn
3
from routers import router  # 从业务层导入路由器
4

5
app = FastAPI()
6

7
# 注册路由器
8
app.include_router(router)
9

10
if __name__ == "__main__":
11
    uvicorn.run(app, host="127.0.0.1", port=8000)

1
主应用 app = FastAPI() 就像主板
2
APIRouter 就像 PCI 插槽
3

4
你可以有：
5
    显卡插槽（Todos 路由）
6
    声卡插槽（Users 路由）
7
    网卡插槽（Orders 路由）
8

9
每个插槽（Router）有自己的：
10
    - 前缀（prefix）
11
    - 标签（tags）
12
    - 依赖（dependencies）

1
┌─────────────────────────────────────────┐
2
│  第1步：创建 Router（在 routers.py）      │
3
│                                          │
4
│  router = APIRouter(prefix="/todos")     │
5
│                                          │
6
│  @router.post("/")                       │
7
│  def create(): ...                       │
8
└─────────────────────────────────────────┘
9
                    ↓
10
            定义了路由规则
11
                    ↓
12
┌─────────────────────────────────────────┐
13
│  第2步：导入 Router（在 main.py）         │
14
│                                          │
15
│  from routers import router              │
16
└─────────────────────────────────────────┘
17
                    ↓
18
            获取路由对象
19
                    ↓
20
┌─────────────────────────────────────────┐
21
│  第3步：注册 Router（在 main.py）         │
22
│                                          │
23
│  app.include_router(router)              │
24
│                                          │
25
│  路由生效！可以访问 /todos/ 了            │
26
└─────────────────────────────────────────┘

1
router = APIRouter(
2
    prefix="/todos",      # 路由前缀
3
    tags=["Todos"],       # OpenAPI 文档标签
4
    dependencies=[],      # 全局依赖（可选）
5
    responses={}          # 默认响应（可选）
6
)

1
[Users]
2
  POST /users/        创建用户
3
  GET  /users/        获取用户列表
4
  GET  /users/{user_id} 获取单个用户
5

6
[Orders]
7
  POST /orders/       创建订单
8
  GET  /orders/       获取订单列表
9

10
[Products]
11
  GET  /products/     获取商品列表

1
# ❌ 不用 APIRouter - 屎山代码
2
from fastapi import FastAPI
3
app = FastAPI()
4

5
@app.post("/todos/")
6
@app.get("/todos/")
7
@app.put("/todos/{id}")
8
@app.delete("/todos/{id}")
9
@app.post("/users/")
10
@app.get("/users/")
11
@app.post("/orders/")
12
# ... 几百行后 ...

1
# PUT - 全量替换
2
# 请求: PUT /todos/1
3
# Body: {"title": "新标题", "is_done": true}
4
# 结果: 整条记录被替换
5

6
# PATCH - 局部修改
7
# 请求: PATCH /todos/1
8
# Body: {"is_done": true}
9
# 结果: 只修改 is_done 字段，title 保持不变

1
@router.put("/{todo_id}")
2
def update_todo(todo_id: int, item: TodoItem, db: Session = Depends(get_db)):
3
    # 第1步：查询要更新的记录
4
    todo = db.query(DBTodo).filter(DBTodo.id == todo_id).first()
5

6
    # 第2步：检查是否存在
7
    if not todo:
8
        return {"error": "找不到"}
9

10
    # 第3步：修改属性（ORM 会自动追踪变化）
11
    todo.title = item.title
12
    todo.is_done = item.is_done
13

14
    # 第4步：提交事务
15
    db.commit()
16

17
    # 第5步：返回更新后的数据
18
    return todo

1
客户端: PUT /todos/1
2
Body: {"title": "学习 FastAPI", "is_done": true}
3
        ↓
4
┌─────────────────────────────────────────┐
5
│  第1步：查询记录                          │
6
│  db.query(DBTodo).filter(DBTodo.id == 1) │
7
│       .first()                           │
8
│  执行 SQL: SELECT * FROM todos WHERE id=1 │
9
└─────────────────────────────────────────┘
10
        ↓
11
┌─────────────────────────────────────────┐
12
│  第2步：检查存在性                        │
13
│  if not todo: return {"error": "找不到"}  │
14
└─────────────────────────────────────────┘
15
        ↓
16
┌─────────────────────────────────────────┐
17
│  第3步：修改属性（内存中）                 │
18
│  todo.title = "学习 FastAPI"              │
19
│  todo.is_done = True                      │
20
│                                          │
21
│  SQLAlchemy 追踪到变化，标记为"脏数据"      │
22
└─────────────────────────────────────────┘
23
        ↓
24
┌─────────────────────────────────────────┐
25
│  第4步：提交事务                          │
26
│  db.commit()                             │
27
│                                          │
28
│  执行 SQL: UPDATE todos                   │
29
│           SET title='学习 FastAPI',       │
30
│               is_done=1                   │
31
│           WHERE id=1                      │
32
└─────────────────────────────────────────┘
33
        ↓
34
┌─────────────────────────────────────────┐
35
│  第5步：返回响应                          │
36
│  return todo                             │
37
│  FastAPI 自动转换为 JSON                  │
38
└─────────────────────────────────────────┘

1
todo = db.query(DBTodo).filter(DBTodo.id == 1).first()
2
# todo 此时是 "干净" 的
3

4
todo.title = "新标题"
5
# SQLAlchemy 检测到属性变化，标记为 "脏"（dirty）
6
# 但还没有执行 SQL！
7

8
db.commit()
9
# 此时 SQLAlchemy 检查所有"脏"对象
10
# 自动生成并执行 UPDATE 语句

1
                    main.py
2
                      │
3
                      │ from routers import router
4
                      ▼
5
                 routers.py
6
                   /      \
7
                  /        \
8
 from database    /          \  from models
9
 import          /            \ import
10
 SessionLocal    /              \ DBTodo
11
                /                \
12
               ▼                  ▼
13
        database.py          models.py
14
               \                /
15
                \              /
16
                 \            /
17
                  \          /
18
                   ▼        ▼
19
               from database import engine

1
# 1. 运行 main.py
2
#    ↓
3
# 2. 遇到 from routers import router
4
#    ↓
5
# 3. 加载 routers.py
6
#    ↓
7
# 4. 遇到 from database import SessionLocal
8
#    ↓
9
# 5. 加载 database.py
10
#    ↓
11
# 6. database.py 执行完毕，创建 engine 和 SessionLocal
12
#    ↓
13
# 7. 回到 routers.py，继续执行
14
#    ↓
15
# 8. 遇到 from models import DBTodo
16
#    ↓
17
# 9. 加载 models.py
18
#    ↓
19
# 10. models.py 从 database 导入 engine
20
#     ↓
21
# 11. models.py 执行 Base.metadata.create_all(bind=engine)
22
#     ↓
23
# 12. 数据表创建完成
24
#     ↓
25
# 13. 回到 routers.py，继续定义路由
26
#     ↓
27
# 14. 回到 main.py，注册路由，启动应用

1
database.py ← models.py ← routers.py ← main.py
2
     ↑
3
     └── 不会反向导入

1
# main.py - 100行
2
所有代码在一起，适合学习

1
# main.py - 50行
2
# database.py - 10行
3
# models.py - 15行
4
# routers.py - 60行
5
按功能拆分，适合小项目

1
PyCharmMiscProject/
2
├── app/                     # 核心代码包（Python package）
3
│   ├── __init__.py
4
│   ├── main.py              # 应用入口（路由注册+中间件）
5
│   ├── database.py          # 数据库配置
6
│   ├── models.py            # 所有ORM模型（集中放一起）
7
│   ├── embedding.py         # Embedding模型工具
8
│   └── routers/             # 路由按功能拆分
9
│       ├── __init__.py
10
│       ├── ai.py            # AI流式对话
11
│       ├── auth.py          # JWT认证
12
│       ├── todos.py         # Todo CRUD
13
│       ├── chat_memory.py   # 聊天记忆
14
│       ├── rag.py           # 手搓RAG
15
│       ├── langchain_rag.py # LangChain版RAG
16
│       ├── websocket.py     # WebSocket
17
│       └── prompt.py        # 提示词工程
18
│
19
├── archive/                 # 归档：早期学习代码
20
│   └── month1-python-basics/
21
│
22
├── playground/              # 实验区：demo、试错代码
23
│
24
├── tests/                   # 单元测试
25
│   ├── conftest.py
26
│   └── test_*.py
27
│
28
├── main.py                  # 根目录启动入口
29
├── alembic/                 # 数据库迁移
30
├── pyproject.toml
31
└── README.md

1
# 所有核心代码从 app 包导入
2
from app.database import get_db
3
from app.models import DBTodo, User
4
from app.routers import todos, auth, rag

1
from fastapi import APIRouter
2

3
router = APIRouter(
4
    prefix="/前缀",
5
    tags=["标签"]
6
)
7

8
@router.get("/")
9
def get_items(): ...
10

11
# main.py
12
app.include_router(router)

1
@router.put("/{id}")
2
def update_item(id: int, item: ItemSchema, db: Session = Depends(get_db)):
3
    # 1. 查询
4
    db_item = db.query(Model).filter(Model.id == id).first()
5
    if not db_item:
6
        raise HTTPException(status_code=404, detail="Not found")
7

8
    # 2. 修改
9
    db_item.field = item.field
10

11
    # 3. 提交
12
    db.commit()
13
    return db_item

1
# 标准库
2
import os
3
from typing import Optional
4

5
# 第三方库
6
from fastapi import APIRouter
7
from sqlalchemy.orm import Session
8

9
# 本地模块
10
from database import SessionLocal
11
from models import DBTodo