Flask 应用完整示例：对象生命周期管理

下面是一个完整的 Flask 应用示例，展示各种对象类型的生命周期管理，包括 Flask 应用实例、请求上下文对象、数据库连接、全局单例对象等。

from flask import Flask, request, session, g, current_app
from flask_sqlalchemy import SQLAlchemy
import redis
from datetime import datetime

# ----------------------------
# 1. Flask 应用实例 (单例)
# ----------------------------
app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///example.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# ----------------------------
# 2. Flask 扩展 (生命周期与app一致)
# ----------------------------
db = SQLAlchemy(app)

# 定义数据模型
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)

# ----------------------------
# 3. 全局单例对象 (Redis客户端)
# ----------------------------
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# ----------------------------
# 4. 自定义服务类 (演示不同生命周期)
# ----------------------------
class AuthService:
    """请求级别的服务 (每次请求新建实例)"""
    def __init__(self):
        self.request_time = datetime.now()
        print(f"AuthService 实例创建于 {self.request_time}")

    def get_current_user(self):
        return session.get('user_id')

class CacheService:
    """应用级别的服务 (单例)"""
    def __init__(self):
        self.init_time = datetime.now()
        print(f"CacheService 单例初始化于 {self.init_time}")
    
    def get_data(self, key):
        return redis_client.get(key)

# 初始化单例
cache_service = CacheService()

# ----------------------------
# 5. 生命周期钩子
# ----------------------------
@app.before_first_request
def initialize_data():
    """应用启动时执行一次 (Flask 2.3+ 已弃用，建议使用CLI命令)"""
    print("初始化数据库...")
    db.create_all()
    db.session.add(User(username='admin'))
    db.session.commit()

@app.before_request
def before_each_request():
    """每个请求前执行"""
    g.request_start_time = datetime.now()
    print(f"请求开始于 {g.request_start_time}")
    
    # 将AuthService实例挂载到g对象
    g.auth_service = AuthService()

@app.after_request
def after_each_request(response):
    """每个请求后执行"""
    duration = datetime.now() - g.request_start_time
    print(f"请求处理耗时: {duration.total_seconds()}秒")
    return response

@app.teardown_request
def teardown_request(exception):
    """请求结束时执行 (即使发生异常也会执行)"""
    print("清理请求资源...")

@app.teardown_appcontext
def teardown_db(exception):
    """应用上下文销毁时执行"""
    print("关闭数据库连接...")
    db.session.remove()

# ----------------------------
# 6. 路由和视图函数
# ----------------------------
@app.route('/')
def index():
    # 访问不同作用域的对象
    app_name = current_app.name  # 当前应用实例
    user_agent = request.headers.get('User-Agent')  # 请求对象
    session['visits'] = session.get('visits', 0) + 1  # 会话对象
    
    # 使用请求级别的服务
    current_user = g.auth_service.get_current_user()
    
    # 使用应用级别的服务
    cache_data = cache_service.get_data('some_key')
    
    # 使用数据库
    user_count = User.query.count()
    
    return f'''
        <h1>{app_name}</h1>
        <p>User Agent: {user_agent}</p>
        <p>访问次数: {session['visits']}</p>
        <p>当前用户: {current_user}</p>
        <p>用户总数: {user_count}</p>
    '''

# ----------------------------
# 7. 启动应用
# ----------------------------
if __name__ == '__main__':
    app.run(debug=True)

关键生命周期说明

1. Flask 应用实例 (app)

   • 整个应用运行期间只有一个实例
   • 在模块加载时创建，进程结束时销毁

2. 请求上下文对象

   • request: 每个HTTP请求创建新实例
   • session: 跨请求持久化(默认使用cookie)
   • g: 请求级别的临时存储

3. 数据库连接

   • db = SQLAlchemy(app) 与app生命周期一致
   • 使用连接池管理数据库连接
   • teardown_appcontext 确保连接正确关闭

4. 服务类

   • AuthService: 请求级别(每个请求新建)
   • CacheService: 应用级别(单例)

5. Redis客户端

   • 全局单例，所有请求共享
   • 进程结束时自动关闭连接

6. 生命周期钩子

   • before_first_request: 应用初始化(已弃用)
   • before_request: 每个请求前执行
   • after_request: 每个请求后执行
   • teardown_request: 请求结束时清理
   • teardown_appcontext: 应用上下文销毁时清理

补充知识

1. 多Worker环境

   • 在生产环境(如Gunicorn)中，每个Worker进程有自己的app实例
   • 全局变量在不同Worker间不共享

2. 异步支持

   • Flask 2.0+支持async/await
   • 需使用兼容的ASGI服务器(如Hypercorn)

3. 替代before_first_request

   @app.cli.command('init-db')
   def init_db():
       """替代before_first_request的CLI命令"""
       db.create_all()
       db.session.add(User(username='admin'))
       db.session.commit()
   

   使用方式: flask init-db

这个示例完整展示了Flask应用中各种对象的生命周期管理方式，涵盖了从请求处理到资源管理的各个方面。