199 lines
5.9 KiB
Markdown
199 lines
5.9 KiB
Markdown
# Flask g 对象与请求上下文
|
||
|
||
:::tip 本章概要
|
||
Flask 的 `g` 对象是请求级别的全局变量容器，用于在同一请求的不同函数间共享数据。本文介绍 `g` 对象的使用方式、请求上下文机制，以及项目中 `g` 对象的实际应用场景。
|
||
:::
|
||
|
||
## 什么是 g 对象
|
||
|
||
`g`（global 的缩写）是 Flask 提供的请求级全局变量对象。它的生命周期与请求绑定：
|
||
|
||
```
|
||
请求进入 → 创建 g 对象
|
||
↓
|
||
路由处理 → 通过 g 存取数据
|
||
↓
|
||
响应返回 → 销毁 g 对象
|
||
```
|
||
|
||
:::warning 线程安全
|
||
每个请求有独立的 `g` 对象，不同请求之间的 `g` 互不影响。在多线程环境下，每个线程处理的请求都有自己的 `g`，不存在数据竞争。
|
||
:::
|
||
|
||
## 基本用法
|
||
|
||
```python
|
||
from flask import g
|
||
|
||
# 在请求开始时设置
|
||
g.user_id = 1
|
||
g.username = 'admin'
|
||
|
||
# 在请求处理过程中获取
|
||
user_id = g.user_id
|
||
```
|
||
|
||
## 项目中的实际应用
|
||
|
||
### 1. 事务状态管理
|
||
|
||
AOP 事务拦截器通过 `g` 对象管理事务状态：
|
||
|
||
```python
|
||
# src/middleware/transaction.py
|
||
from flask import g, request
|
||
from src.core.database import db
|
||
|
||
|
||
def register_transaction(app):
|
||
"""注册 AOP 事务拦截器"""
|
||
|
||
@app.before_request
|
||
def begin_transaction():
|
||
"""请求开始：记录开始时间，开启事务标记"""
|
||
g.request_start_time = time.time()
|
||
g.transaction_active = True
|
||
|
||
@app.after_request
|
||
def commit_transaction(response):
|
||
"""请求结束：自动提交事务"""
|
||
if request.method in ("POST", "PUT", "DELETE", "PATCH"):
|
||
if 200 <= response.status_code < 300:
|
||
data = response.get_json(silent=True)
|
||
if data and data.get("code", 0) == 0:
|
||
db.session.commit()
|
||
else:
|
||
db.session.rollback()
|
||
return response
|
||
|
||
@app.teardown_request
|
||
def cleanup_transaction(exception=None):
|
||
"""请求结束：清理资源"""
|
||
if exception:
|
||
db.session.rollback()
|
||
g.pop("transaction_active", None)
|
||
g.pop("request_start_time", None)
|
||
```
|
||
|
||
### 2. 当前用户信息
|
||
|
||
登录校验中间件将当前用户信息存入 `g` 对象，业务代码通过 `get_user_id()` 等函数获取：
|
||
|
||
```python
|
||
# src/middleware/auth.py
|
||
from flask import g, request
|
||
from src.core.jwt import decode_token
|
||
|
||
|
||
def register_middleware(app):
|
||
"""注册登录校验中间件（实际位于 src/middleware/auth.py）"""
|
||
|
||
@app.before_request
|
||
def login_required():
|
||
"""登录校验：解析 JWT Token，注入用户信息到 g 对象"""
|
||
# 白名单路径跳过校验
|
||
if _is_whitelist_path(request.path):
|
||
return
|
||
|
||
# 提取 Token
|
||
token = request.headers.get('Authorization', '').replace('Bearer ', '')
|
||
if not token:
|
||
return R.failed(msg="未登录", code=401)
|
||
|
||
# 解析 Token
|
||
payload = decode_token(token)
|
||
if not payload:
|
||
return R.failed(msg="Token 无效", code=401)
|
||
|
||
# 注入用户信息到 g 对象
|
||
g.user_id = payload.get('user_id')
|
||
g.username = payload.get('username')
|
||
g.realname = payload.get('realname')
|
||
```
|
||
|
||
```python
|
||
# src/core/security.py
|
||
from flask import g
|
||
|
||
|
||
def get_user_id():
|
||
"""获取当前登录用户ID"""
|
||
return getattr(g, 'user_id', None)
|
||
|
||
|
||
def get_username():
|
||
"""获取当前登录用户名"""
|
||
return getattr(g, 'username', None)
|
||
|
||
|
||
def get_realname():
|
||
"""获取当前登录用户真实姓名"""
|
||
return getattr(g, 'realname', None)
|
||
```
|
||
|
||
业务代码通过这些函数获取当前用户信息：
|
||
|
||
```python
|
||
# src/modules/system/position/service.py
|
||
from src.core.security import get_realname
|
||
|
||
class PositionService(BaseService):
|
||
@classmethod
|
||
def add(cls, data: dict) -> None:
|
||
"""新增岗位"""
|
||
form = cls._validate_form(data)
|
||
create_data = cls._clean_form_data(form)
|
||
create_data["create_user"] = get_realname() # 从 g 对象获取
|
||
create_data["update_user"] = get_realname()
|
||
cls.repository.create(**create_data)
|
||
```
|
||
|
||
### 3. 性能监控
|
||
|
||
事务拦截器通过 `g` 记录请求开始时间，用于计算响应耗时：
|
||
|
||
```python
|
||
# src/middleware/transaction.py
|
||
@app.after_request
|
||
def commit_transaction(response):
|
||
elapsed = time.time() - g.get("request_start_time", time.time())
|
||
logger.debug(
|
||
f"请求结束, method: {request.method}, path: {request.path}, "
|
||
f"status: {response.status_code}, elapsed: {elapsed:.3f}s"
|
||
)
|
||
return response
|
||
```
|
||
|
||
## g 对象 vs 其他存储方式
|
||
|
||
| 存储方式 | 生命周期 | 线程安全 | 适用场景 |
|
||
|:---|:---|:---|:---|
|
||
| `g` | 单个请求 | 是 | 请求级数据（当前用户、事务状态） |
|
||
| `session` | 用户会话 | 是 | 用户级数据（登录状态、偏好设置） |
|
||
| `current_app.config` | 应用生命周期 | 是 | 应用级配置（数据库 URI、密钥） |
|
||
| 模块级变量 | 进程生命周期 | 否 | 单例对象（扩展实例） |
|
||
|
||
:::warning 注意事项
|
||
- `g` 对象只能在请求上下文中访问，不能在请求外使用
|
||
- `g` 对象在请求结束后自动销毁，不要用于持久化数据
|
||
- 需要跨请求共享的数据应使用 Redis 或数据库
|
||
:::
|
||
|
||
## 请求上下文
|
||
|
||
Flask 的请求上下文机制确保 `g`、`request`、`session` 等对象在请求处理期间可用：
|
||
|
||
```python
|
||
# 在请求上下文中可以访问 g
|
||
with app.test_request_context():
|
||
g.user_id = 1
|
||
print(g.user_id) # 正常访问
|
||
|
||
# 在请求上下文外访问 g 会报错
|
||
# print(g.user_id) # RuntimeError: Working outside of application context.
|
||
```
|
||
|
||
## 总结
|
||
|
||
`g` 对象是 Flask 请求级全局变量容器，项目中用于存储当前用户信息（`user_id` / `username` / `realname`）、事务状态（`transaction_active`）、请求开始时间（`request_start_time`）等请求级数据。通过 `get_user_id()` / `get_realname()` 等安全函数封装访问，业务代码无需直接操作 `g` 对象。
|