Day 066 — 图解:Flask 进阶核心概念
本目录包含 Flask 进阶话题的图解说明:ORM 原理、JWT 认证、RESTful API 架构。
1. SQLAlchemy 架构总览
graph TB
subgraph "应用代码 (Python)"
A["class User(db.Model):<br/>id = Column(Integer)<br/>name = Column(String)"]
B["user = User(name='Alice')<br/>db.session.add(user)<br/>db.session.commit()"]
C["User.query.filter_by(name='Alice')<br/>.order_by(User.id.desc())<br/>.all()"]
end
subgraph "SQLAlchemy ORM"
D["ORM 层<br/>对象 ↔ 关系映射<br/>会话管理<br/>变更跟踪"]
end
subgraph "SQLAlchemy Core"
E["SQL 表达式语言<br/>schema/types/sql<br/>engine/connection"]
end
subgraph "数据库驱动 (Dialect)"
F["SQLite<br/>Dialect"]
G["PostgreSQL<br/>Dialect"]
H["MySQL<br/>Dialect"]
end
subgraph "数据库"
I["📁 SQLite<br/>app.db"]
J["🐘 PostgreSQL"]
K["🐬 MySQL"]
end
A --> D
B --> D
C --> D
D --> E
E --> F
E --> G
E --> H
F --> I
G --> J
H --> K
style D fill:#4A90D9,color:#fff
style E fill:#7B68EE,color:#fff
graph TB
subgraph "应用代码 (Python)"
A["class User(db.Model):<br/>id = Column(Integer)<br/>name = Column(String)"]
B["user = User(name='Alice')<br/>db.session.add(user)<br/>db.session.commit()"]
C["User.query.filter_by(name='Alice')<br/>.order_by(User.id.desc())<br/>.all()"]
end
subgraph "SQLAlchemy ORM"
D["ORM 层<br/>对象 ↔ 关系映射<br/>会话管理<br/>变更跟踪"]
end
subgraph "SQLAlchemy Core"
E["SQL 表达式语言<br/>schema/types/sql<br/>engine/connection"]
end
subgraph "数据库驱动 (Dialect)"
F["SQLite<br/>Dialect"]
G["PostgreSQL<br/>Dialect"]
H["MySQL<br/>Dialect"]
end
subgraph "数据库"
I["📁 SQLite<br/>app.db"]
J["🐘 PostgreSQL"]
K["🐬 MySQL"]
end
A --> D
B --> D
C --> D
D --> E
E --> F
E --> G
E --> H
F --> I
G --> J
H --> K
style D fill:#4A90D9,color:#fff
style E fill:#7B68EE,color:#fff
各层职责
应用层 (你的 Python 代码)
│
▼ 定义模型、CRUD 操作、业务逻辑
│
ORM 层 (对象关系映射)
│
│ • 模型 ↔ 表的映射
│ • 关系管理 (一对多/多对多)
│ • 变更追踪 (unit of work)
│ • 会话管理 (session)
│ • 惰性加载/预加载
│
▼ 翻译成 SQL 表达式
│
Core 层 (SQL 表达式语言)
│
│ • SQL 抽象语法树
│ • 类型系统与类型转换
│ • 连接池与事务管理
│ • 结果集处理
│
▼ 发送到具体数据库
│
Dialect (数据库方言)
│
│ • SQL 语法适配 (不同数据库 SQL 略有差异)
│ • 驱动封装 (sqlite3/psycopg2/mysql-connector)
│
▼
Database (数据库)
2. SQLAlchemy 对象状态生命周期
stateDiagram-v2
[*] --> Transient: User()
Transient --> Pending: db.session.add(user)
Pending --> Flushed: db.session.flush()
Pending --> Detached: db.session.expunge(user)
Flushed --> Committed: db.session.commit()
Flushed --> Pending: 事务回滚
Committed --> Loaded: user.todos
Loaded --> Expired: db.session.expire(user)
Expired --> Loaded: user.name (触发重新查询)
Loaded --> Deleted: db.session.delete(user)
Deleted --> Detached: db.session.commit()
Loaded --> Detached: 会话关闭
Detached --> Loaded: db.session.merge(user)
stateDiagram-v2
[*] --> Transient: User()
Transient --> Pending: db.session.add(user)
Pending --> Flushed: db.session.flush()
Pending --> Detached: db.session.expunge(user)
Flushed --> Committed: db.session.commit()
Flushed --> Pending: 事务回滚
Committed --> Loaded: user.todos
Loaded --> Expired: db.session.expire(user)
Expired --> Loaded: user.name (触发重新查询)
Loaded --> Deleted: db.session.delete(user)
Deleted --> Detached: db.session.commit()
Loaded --> Detached: 会话关闭
Detached --> Loaded: db.session.merge(user)
| 状态 | 说明 | 是否在 session 中 | 数据库中是否存在 |
|---|---|---|---|
| Transient | 刚创建的 Python 对象 | ❌ | ❌ |
| Pending | 已添加到 session,未刷新 | ✅ | ❌ |
| Flushed | SQL 已发送到数据库,未提交 | ✅ | ✅(事务中) |
| Committed | 已提交到数据库 | ✅ | ✅ |
| Detached | 与 session 分离 | ❌ | ✅ |
| Deleted | 标记为删除 | ✅ | ✅(等待提交删除) |
3. JWT 结构详解
graph LR
subgraph "JWT = Header.Payload.Signature"
H["Header<br/>{<br/> "alg": "HS256",<br/> "typ": "JWT"<br/>}"]
P["Payload<br/>{<br/> "sub": 1,<br/> "iat": 1700000000,<br/> "exp": 1700086400<br/>}"]
S["Signature<br/>HMACSHA256(<br/> base64(Header) + '.' +<br/> base64(Payload),<br/> SECRET_KEY<br/>)"]
end
H -->|Base64URL| H64["eyJhbGciOiJIUzI1NiIs..."]
P -->|Base64URL| P64["eyJzdWIiOjEsImlhdCI6MTcw..."]
S --> S64["dGhpcyBpcyB0aGUgc2lnbmF0..."]
H64 --> Token["eyJhbGciOiJIUzI1NiIs...<br/>.<br/>eyJzdWIiOjEsImlhdCI6MTcw...<br/>.<br/>dGhpcyBpcyB0aGUgc2lnbmF0..."]
P64 --> Token
S64 --> Token
style Token fill:#f0f0f0,stroke:#333,stroke-width:2px
graph LR
subgraph "JWT = Header.Payload.Signature"
H["Header<br/>{<br/> "alg": "HS256",<br/> "typ": "JWT"<br/>}"]
P["Payload<br/>{<br/> "sub": 1,<br/> "iat": 1700000000,<br/> "exp": 1700086400<br/>}"]
S["Signature<br/>HMACSHA256(<br/> base64(Header) + '.' +<br/> base64(Payload),<br/> SECRET_KEY<br/>)"]
end
H -->|Base64URL| H64["eyJhbGciOiJIUzI1NiIs..."]
P -->|Base64URL| P64["eyJzdWIiOjEsImlhdCI6MTcw..."]
S --> S64["dGhpcyBpcyB0aGUgc2lnbmF0..."]
H64 --> Token["eyJhbGciOiJIUzI1NiIs...<br/>.<br/>eyJzdWIiOjEsImlhdCI6MTcw...<br/>.<br/>dGhpcyBpcyB0aGUgc2lnbmF0..."]
P64 --> Token
S64 --> Token
style Token fill:#f0f0f0,stroke:#333,stroke-width:2px
JWT 三个部分的含义
第一部分 ─ Header(头部)
{"alg": "HS256"} → 签名算法:HMAC-SHA256
{"typ": "JWT"} → Token 类型
第二部分 ─ Payload(载荷)
"sub" (Subject) → 用户 ID
"iat" (Issued At) → 签发时间
"exp" (Expiration)→ 过期时间
"type" → access / refresh
第三部分 ─ Signature(签名)
防止 Token 被篡改
用 SECRET_KEY + Header + Payload 计算
服务器端唯一能验证(因为只有服务器知道 SECRET_KEY)
4. 认证流程:注册 → 登录 → 受保护请求
sequenceDiagram
participant C as 客户端
participant API as Flask API
participant JWT as JWT 模块
participant DB as SQLAlchemy
Note over C,DB: ─── 注册流程 ───
C->>API: POST /api/auth/register<br/>{username, password}
API->>API: 参数校验
API->>DB: 检查用户名是否已存在
DB-->>API: 不存在
API->>API: generate_password_hash(password)
API->>DB: INSERT user
DB-->>API: 用户已创建
API->>JWT: create_token(user_id)
JWT-->>API: "eyJhbGciOiJIUzI1NiIs..."
API-->>C: 201 Created<br/>{token, user}
Note over C,DB: ─── 登录流程 ───
C->>API: POST /api/auth/login<br/>{username, password}
API->>DB: SELECT user WHERE username=?
DB-->>API: User 对象
API->>API: check_password_hash(hash, password)
Note over API: 验证通过 ✅
API->>JWT: create_token(user_id)
JWT-->>API: "eyJhbGciOiJIUzI1NiIs..."
API-->>C: 200 OK<br/>{token, user}
Note over C,DB: ─── 受保护请求 ───
C->>API: GET /api/todos<br/>Authorization: Bearer xxx
API->>JWT: decode_token(token)
JWT-->>API: {sub: 1, exp: ...}
API->>DB: SELECT user WHERE id=1
DB-->>API: User 对象
API->>DB: SELECT todos WHERE user_id=1
DB-->>API: [Todo, Todo, Todo]
API-->>C: 200 OK<br/>{data: [...]}
sequenceDiagram
participant C as 客户端
participant API as Flask API
participant JWT as JWT 模块
participant DB as SQLAlchemy
Note over C,DB: ─── 注册流程 ───
C->>API: POST /api/auth/register<br/>{username, password}
API->>API: 参数校验
API->>DB: 检查用户名是否已存在
DB-->>API: 不存在
API->>API: generate_password_hash(password)
API->>DB: INSERT user
DB-->>API: 用户已创建
API->>JWT: create_token(user_id)
JWT-->>API: "eyJhbGciOiJIUzI1NiIs..."
API-->>C: 201 Created<br/>{token, user}
Note over C,DB: ─── 登录流程 ───
C->>API: POST /api/auth/login<br/>{username, password}
API->>DB: SELECT user WHERE username=?
DB-->>API: User 对象
API->>API: check_password_hash(hash, password)
Note over API: 验证通过 ✅
API->>JWT: create_token(user_id)
JWT-->>API: "eyJhbGciOiJIUzI1NiIs..."
API-->>C: 200 OK<br/>{token, user}
Note over C,DB: ─── 受保护请求 ───
C->>API: GET /api/todos<br/>Authorization: Bearer xxx
API->>JWT: decode_token(token)
JWT-->>API: {sub: 1, exp: ...}
API->>DB: SELECT user WHERE id=1
DB-->>API: User 对象
API->>DB: SELECT todos WHERE user_id=1
DB-->>API: [Todo, Todo, Todo]
API-->>C: 200 OK<br/>{data: [...]}
5. RESTful API 资源映射
graph LR
subgraph "HTTP 方法"
GET[GET<br/>读取]
POST[POST<br/>创建]
PUT[PUT<br/>全量更新]
PATCH[PATCH<br/>局部更新]
DELETE[DELETE<br/>删除]
end
subgraph "URL 路径"
COLL["/todos<br/>资源集合"]
ITEM["/todos/:id<br/>单个资源"]
ACT["/todos/:id/toggle<br/>动作"]
end
GET --> COLL
POST --> COLL
GET --> ITEM
PUT --> ITEM
PATCH --> ITEM
DELETE --> ITEM
POST --> ACT
COLL -->|返回列表| R1["200 + 数组"]
COLL -->|创建| R2["201 + 新资源"]
ITEM -->|读取| R3["200 + 资源"]
ITEM -->|更新| R4["200 + 资源"]
ITEM -->|删除| R5["204 无内容"]
ACT -->|操作| R6["200 + 结果"]
graph LR
subgraph "HTTP 方法"
GET[GET<br/>读取]
POST[POST<br/>创建]
PUT[PUT<br/>全量更新]
PATCH[PATCH<br/>局部更新]
DELETE[DELETE<br/>删除]
end
subgraph "URL 路径"
COLL["/todos<br/>资源集合"]
ITEM["/todos/:id<br/>单个资源"]
ACT["/todos/:id/toggle<br/>动作"]
end
GET --> COLL
POST --> COLL
GET --> ITEM
PUT --> ITEM
PATCH --> ITEM
DELETE --> ITEM
POST --> ACT
COLL -->|返回列表| R1["200 + 数组"]
COLL -->|创建| R2["201 + 新资源"]
ITEM -->|读取| R3["200 + 资源"]
ITEM -->|更新| R4["200 + 资源"]
ITEM -->|删除| R5["204 无内容"]
ACT -->|操作| R6["200 + 结果"]
RESTful URL 设计原则
集合 (Collection): /resources
单个资源: /resources/:id
子资源: /resources/:id/sub-resources
动作 (非 CRUD): /resources/:id/action
搜索/过滤: /resources?q=keyword&status=active
分页: /resources?page=1&per_page=20
6. 完整请求处理流程(TODO API)
flowchart TB
Start["HTTP 请求<br/>POST /api/v1/todos"] --> Auth{"token_required<br/>检查 Authorization"}
Auth -->|"无 Token"| Err401["401 Unauthorized<br/>{code: AUTH_REQUIRED}"]
Auth -->|"Token 无效/过期"| Err401
Auth -->|Token 有效| UserCheck["查找用户<br/>User.query.get(sub)"]
UserCheck -->|用户不存在| Err401_2["401 Unauthorized<br/>{code: USER_NOT_FOUND}"]
UserCheck -->|用户存在| Validate["参数校验<br/>validate_todo_data()"]
Validate -->|校验失败| Err422["422 Unprocessable<br/>{code: VALIDATION_ERROR,<br/>details: {...}}"]
Validate -->|校验通过| Create["创建 Todo 对象<br/>todo = Todo(**cleaned)"]
Create --> Add["db.session.add(todo)"]
Add --> Commit["db.session.commit()"]
Commit --> Response["201 Created<br/>{message, data}"]
style Err401 fill:#ff9999
style Err401_2 fill:#ff9999
style Err422 fill:#ffcc99
style Response fill:#90EE90
flowchart TB
Start["HTTP 请求<br/>POST /api/v1/todos"] --> Auth{"token_required<br/>检查 Authorization"}
Auth -->|"无 Token"| Err401["401 Unauthorized<br/>{code: AUTH_REQUIRED}"]
Auth -->|"Token 无效/过期"| Err401
Auth -->|Token 有效| UserCheck["查找用户<br/>User.query.get(sub)"]
UserCheck -->|用户不存在| Err401_2["401 Unauthorized<br/>{code: USER_NOT_FOUND}"]
UserCheck -->|用户存在| Validate["参数校验<br/>validate_todo_data()"]
Validate -->|校验失败| Err422["422 Unprocessable<br/>{code: VALIDATION_ERROR,<br/>details: {...}}"]
Validate -->|校验通过| Create["创建 Todo 对象<br/>todo = Todo(**cleaned)"]
Create --> Add["db.session.add(todo)"]
Add --> Commit["db.session.commit()"]
Commit --> Response["201 Created<br/>{message, data}"]
style Err401 fill:#ff9999
style Err401_2 fill:#ff9999
style Err422 fill:#ffcc99
style Response fill:#90EE90
7. 密码哈希原理图解
用户注册密码: "myPassword123"
│
▼
生成随机 Salt: "a1b2c3d4e5f6g7h8"
│
▼
组合: "myPassword123" + "a1b2c3d4e5f6g7h8"
│
▼
PBKDF2-HMAC-SHA256 (600000 次迭代)
│
▼
输出哈希: "pbkdf2:sha256:600000$a1b2c3d4e5f6g7h8$..."
↑
Salt 被包含在输出中
存储到数据库: password_hash 字段 ← "pbkdf2:sha256:600000$..."
──────────── 验证 ────────────
用户输入密码: "myPassword123"
check_password_hash() 执行:
1. 从存储的哈希中提取 Salt: "a1b2c3d4e5f6g7h8"
2. 用同样的 Salt + 算法重新计算哈希
3. 比较结果是否一致
如果一致 → 密码正确 ✅
如果不一致 → 密码错误 ❌
8. ORM 查询中 N+1 问题
graph TD
subgraph "❌ N+1 查询 (lazy='select')"
Q1["1 次查询: SELECT * FROM users"]
Q1 --> U1["用户 1"]
Q1 --> U2["用户 2"]
Q1 --> U3["用户 3"]
U1 --> T1["1 次查询: SELECT * FROM todos WHERE user_id=1"]
U2 --> T2["1 次查询: SELECT * FROM todos WHERE user_id=2"]
U3 --> T3["1 次查询: SELECT * FROM todos WHERE user_id=3"]
Total1["总共: 1 (用户) + N (用户数) 次查询"]
end
subgraph "✅ 预加载 (lazy='joined')"
Q2["1 次查询: SELECT * FROM users<br/>LEFT JOIN todos ON user.id=todos.user_id"]
Q2 --> R1["用户 1 + TODO 列表 ← 一次查询全部获取"]
Q2 --> R2["用户 2 + TODO 列表"]
Q2 --> R3["用户 3 + TODO 列表"]
Total2["总共: 1 次查询!"]
end
style Total1 fill:#ff9999
style Total2 fill:#90EE90
graph TD
subgraph "❌ N+1 查询 (lazy='select')"
Q1["1 次查询: SELECT * FROM users"]
Q1 --> U1["用户 1"]
Q1 --> U2["用户 2"]
Q1 --> U3["用户 3"]
U1 --> T1["1 次查询: SELECT * FROM todos WHERE user_id=1"]
U2 --> T2["1 次查询: SELECT * FROM todos WHERE user_id=2"]
U3 --> T3["1 次查询: SELECT * FROM todos WHERE user_id=3"]
Total1["总共: 1 (用户) + N (用户数) 次查询"]
end
subgraph "✅ 预加载 (lazy='joined')"
Q2["1 次查询: SELECT * FROM users<br/>LEFT JOIN todos ON user.id=todos.user_id"]
Q2 --> R1["用户 1 + TODO 列表 ← 一次查询全部获取"]
Q2 --> R2["用户 2 + TODO 列表"]
Q2 --> R3["用户 3 + TODO 列表"]
Total2["总共: 1 次查询!"]
end
style Total1 fill:#ff9999
style Total2 fill:#90EE90
9. Flask 项目分层架构(推荐)
flask-todo-api/
│
├── app.py ← 创建 Flask 实例、配置加载
│
├── config.py ← 不同环境的配置类
│
├── models/ ← 数据库模型
│ ├── __init__.py
│ ├── user.py
│ └── todo.py
│
├── routes/ ← 路由(蓝图)
│ ├── __init__.py
│ ├── auth.py ← 认证相关路由
│ └── todos.py ← TODO CRUD 路由
│
├── services/ ← 业务逻辑
│ ├── __init__.py
│ ├── auth_service.py
│ └── todo_service.py
│
├── utils/ ← 工具函数
│ ├── __init__.py
│ ├── errors.py ← 错误码和异常
│ ├── validators.py ← 参数校验
│ └── jwt_utils.py ← JWT 工具
│
├── requirements.txt
├── Dockerfile
└── .env ← 环境变量
分层职责:
路由层 (routes/) ← 接收请求、返回响应、参数提取
│
▼ 调用服务层
服务层 (services/) ← 业务逻辑、事务管理
│
▼ 调用模型层
模型层 (models/) ← 数据定义、数据库操作
│
▼
工具层 (utils/) ← JWT、校验、错误码等通用工具