Flask中的上下文管理：从请求隔离到线程安全的底层机制,全面剖析Flask四大上下文（请求/应用/会话/全局）的运行原理

# 危险示例：全局变量污染
current_request = None  # 全局变量

@app.route('/')
def index():
    global current_request
    current_request = request  # 所有请求共享同一个变量
    # 当多个请求同时到达时数据会互相覆盖

# Flask实际工作原理（伪代码）
from werkzeug.local import LocalStack

_request_ctx_stack = LocalStack()  # 线程安全的栈结构

class RequestContext:
    def __init__(self, request):
        self.request = request

def handle_request(request):
    ctx = RequestContext(request)
    _request_ctx_stack.push(ctx)  # 将上下文压入栈
    process_response = dispatch_request()
    _request_ctx_stack.pop()      # 请求结束后弹出
    return process_response

@app.route('/user/<id>')
def get_user(id):
    # 从请求上下文中获取请求对象
    current_request = request  # 自动关联当前请求
    print(f"正在处理请求路径：{current_request.path}")
    
    # 访问查询参数（示例URL：/user/123?debug=true）
    debug_mode = request.args.get('debug', 'false')
    return f"用户{id}的调试模式：{debug_mode}"

# 初始化阶段
app = Flask(__name__)
app.config['API_KEY'] = 'secret123'

# 请求处理中
@app.route('/config')
def show_config():
    # 访问应用级配置
    print(f"当前应用名称：{current_app.name}")
    return f"API密钥长度：{len(current_app.config['API_KEY'])}"

@app.route('/login')
def login():
    session['user'] = 'john@example.com'  # 写入加密的Cookie
    return "登录成功"

@app.route('/profile')
def profile():
    if 'user' not in session:
        return redirect('/login')
    return f"欢迎回来，{session['user']}"

@app.before_request
def load_user():
    # 模拟数据库查询
    user = {'id': 123, 'name': 'John'}
    g.user = user  # 存储到全局上下文

@app.route('/dashboard')
def dashboard():
    return f"欢迎进入控制台，{g.user['name']}"

# 应用初始化
app = Flask(__name__)

# 手动创建应用上下文
with app.app_context():
    print(current_app.name)  # 可以访问应用配置

# Flask内部处理逻辑（伪代码）
def wsgi_app(environ, start_response):
    # 创建请求上下文
    ctx = app.request_context(environ)
    ctx.push()
    
    try:
        # 创建应用上下文（如果不存在）
        if not current_app:
            app_ctx = app.app_context()
            app_ctx.push()
            
        # 执行请求处理
        response = full_dispatch_request()
    finally:
        # 清理上下文
        ctx.pop()
        if app_ctx:
            app_ctx.pop()

from flask import copy_current_request_context

@app.route('/long-task')
def long_task():
    @copy_current_request_context
    def background_task():
        # 即使原请求结束，仍能访问原有上下文
        print(f"任务参数：{request.args.get('id')}")
    
    threading.Thread(target=background_task).start()
    return "任务已启动"

# 模拟LocalStack工作原理
import threading
from collections import defaultdict

class SimpleLocalStack:
    def __init__(self):
        self._storage = defaultdict(list)
    
    def push(self, item):
        ident = threading.get_ident()
        self._storage[ident].append(item)
    
    def pop(self):
        ident = threading.get_ident()
        return self._storage[ident].pop()
    
    @property
    def top(self):
        ident = threading.get_ident()
        return self._storage[ident][-1] if self._storage[ident] else None

# 使用示例
local_stack = SimpleLocalStack()

def worker(num):
    local_stack.push(num)
    print(f"线程{threading.get_ident()}的栈顶：{local_stack.top}")

threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

@app.before_request
def connect_db():
    # 每个请求获取新连接
    g.db_conn = db.engine.connect()

@app.teardown_request
def close_db(exc):
    # 请求结束后释放连接
    if hasattr(g, 'db_conn'):
        g.db_conn.close()

@app.before_request
def identify_tenant():
    # 根据子域名识别租户
    subdomain = request.host.split('.')[0]
    g.tenant = Tenant.query.filter_by(subdomain=subdomain).first()

@app.route('/data')
def get_data():
    # 自动过滤当前租户数据
    return Data.query.filter_by(tenant=g.tenant).all()

# 错误示例
cache = {}

def bad_cache():
    cache[request.url] = time.time()  # 在非请求上下文中访问request

# 正确做法
from flask import has_request_context

def safe_cache():
    if has_request_context():
        cache[request.url] = time.time()

# 错误的手动上下文管理
ctx = app.test_request_context()
ctx.push()
# 忘记调用 ctx.pop()

# 使用with语句自动管理
with app.test_request_context():
    # 自动执行push/pop
    print(request.url)