当前位置:首页>python>Python 零基础100天—Day57 Flask 进阶

Python 零基础100天—Day57 Flask 进阶

  • 2026-09-08 22:00:54
Python 零基础100天—Day57 Flask 进阶

Flask 进阶 — 从能跑到能上线

🕐 预计用时:2-3 小时 | 🎯 目标:掌握中间件、错误处理、日志、CORS 和配置管理


📖 今日目录

  1. 配置管理
  2. 中间件与请求钩子
  3. 错误处理
  4. 日志系统
  5. CORS 跨域
  6. Flask 扩展生态
  7. 今日练习
  8. 今日小结

1. 配置管理

⚙️ 为什么要管理配置?

开发环境和生产环境需要不同的配置——开发用 debug 模式 + SQLite,生产用关闭 debug + MySQL + HTTPS。

# 方式1:直接配置(简单项目)app = Flask(__name__)app.config['SECRET_KEY'] = 'dev-secret-key'app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'app.config['DEBUG'] = True# 方式2:从文件加载(推荐)app.config.from_object('config.ProductionConfig')  # 或 DevelopmentConfig# 方式3:从环境变量加载(最安全)app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'fallback-secret')

📁 配置文件模式

# config.pyimport osclass Config:    """基础配置"""    SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-secret-key')    SQLALCHEMY_TRACK_MODIFICATIONS = False    JSON_AS_ASCII = False  # JSON 支持中文class DevelopmentConfig(Config):    """开发环境"""    DEBUG = True    SQLALCHEMY_DATABASE_URI = 'sqlite:///dev.db'class ProductionConfig(Config):    """生产环境"""    DEBUG = False    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL', 'sqlite:///prod.db')    SESSION_COOKIE_SECURE = True    SESSION_COOKIE_HTTPONLY = Trueclass TestingConfig(Config):    """测试环境"""    TESTING = True    SQLALCHEMY_DATABASE_URI = 'sqlite:///test.db'
# app.pyfrom flask import Flaskapp = Flask(__name__)# 根据环境变量选择配置env = os.environ.get('FLASK_ENV', 'development')if env == 'production':    app.config.from_object('config.ProductionConfig')else:    app.config.from_object('config.DevelopmentConfig')

⚠️ 配置安全规则:• SECRET_KEY、数据库密码等绝不写死在代码中• 用环境变量或 .env 文件存储敏感配置• .env 文件要加入 .gitignore,不能提交到 Git


2. 中间件与请求钩子

🔗 什么是请求钩子?

请求钩子是在请求处理的不同阶段自动执行的函数。就像流水线上的"质检站"——每个请求都经过。

# 请求生命周期:# before_request → 路由匹配 → 视图函数 → after_request → 返回响应@app.before_requestdef before_request_func():    """每次请求前执行(类似中间件)"""    # 记录请求开始时间    g.start_time = time.time()    # 检查 API Key    if request.path.startswith('/api/'):        api_key = request.headers.get('X-API-Key')        if api_key != 'my-secret-key':            return jsonify({'error': 'Invalid API key'}), 401@app.after_requestdef after_request_func(response):    """每次请求后执行(可以修改响应)"""    # 添加 CORS 头    response.headers['X-Custom'] = 'MyValue'    # 记录请求耗时    if hasattr(g, 'start_time'):        elapsed = time.time() - g.start_time        response.headers['X-Response-Time'] = f'{elapsed:.3f}s'    return response@app.teardown_appcontextdef teardown(exception):    """请求结束时执行(即使出错也会执行)"""    # 清理资源,关闭数据库连接等    pass

📋 四种钩子对比

钩子
执行时机
用途
before_request
请求处理前
认证、日志、计时
after_request
请求处理后(成功时)
添加响应头、修改响应
teardown_request
请求结束时(无论成功与否)
清理资源
teardown_appcontext
应用上下文结束时
关闭数据库连接

💡 before_request 的常见用途:• 用户认证检查(未登录 → 重定向)• 请求频率限制(rate limiting)• 请求日志记录(哪个 IP 访问了哪个 URL)• 数据库连接初始化• API Key 验证


3. 错误处理

🚨 全局错误处理器

# 定义统一的错误响应格式def error_response(status_code, message, errors=None):    response = {        'code': status_code,        'message': message,    }    if errors:        response['errors'] = errors    return jsonify(response), status_code# 404 Not Found@app.errorhandler(404)def not_found(error):    if request.accept_mimetypes.accept_json:        return error_response(404, '资源不存在')    return render_template('404.html'), 404# 400 Bad Request@app.errorhandler(400)def bad_request(error):    return error_response(400, '请求参数有误')# 403 Forbidden@app.errorhandler(403)def forbidden(error):    return error_response(403, '权限不足')# 500 Internal Server Error@app.errorhandler(500)def internal_error(error):    db.session.rollback()  # 回滚数据库事务    app.logger.error(f'Server Error: {error}')    return error_response(500, '服务器内部错误')# 自定义异常class APIError(Exception):    """自定义 API 异常"""    def __init__(self, status_code, message):        self.status_code = status_code        self.message = message@app.errorhandler(APIError)def handle_api_error(error):    return error_response(error.status_code, error.message)# 在视图中使用@app.route('/api/users/<int:user_id>')def get_user(user_id):    user = User.query.get(user_id)    if not user:        raise APIError(404, '用户不存在')    return jsonify(user.to_dict())

4. 日志系统

📝 为什么不用 print?

# ❌ print 的问题print("用户登录了")  # 只在终端显示,生产环境看不到print(f"错误:{e}")  # 没有时间戳,没有级别,不可控# ✅ logging 的优势app.logger.info("用户登录")        # 有级别、有时间戳app.logger.error(f"数据库错误:{e}")  # 可以输出到文件、发送告警

🔧 配置日志

# app.pyimport loggingfrom logging.handlers import RotatingFileHandlerimport osdef setup_logging(app):    """配置日志系统"""    # 确保日志目录存在    os.makedirs('logs', exist_ok=True)    # 文件日志(自动轮转,每个文件最大 1MB,保留 10 个备份)    file_handler = RotatingFileHandler(        'logs/app.log',         maxBytes=1024*1024,         backupCount=10,        encoding='utf-8'    )    file_handler.setFormatter(logging.Formatter(        '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'    ))    file_handler.setLevel(logging.INFO)    # 控制台日志    console_handler = logging.StreamHandler()    console_handler.setLevel(logging.DEBUG)    console_handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s: %(message)s'))    # 应用日志配置    app.logger.setLevel(logging.INFO)    app.logger.addHandler(file_handler)    app.logger.addHandler(console_handler)    app.logger.info('Flask 应用启动')

📊 日志级别

级别
场景
示例
DEBUG
调试信息(开发时)
变量值、SQL 查询
INFO
正常操作记录
用户登录、API 调用
WARNING
警告(不影响运行)
磁盘空间低、配置缺失
ERROR
错误(功能受影响)
数据库连接失败、API 超时
CRITICAL
严重错误(系统崩溃)
数据库完全不可用
# 在视图中使用日志@app.route('/api/users', methods=['POST'])def create_user():    data = request.get_json()    app.logger.info(f'创建用户请求:{data}')    try:        user = User(**data)        db.session.add(user)        db.session.commit()        app.logger.info(f'用户创建成功:{user.username}')        return jsonify(user.to_dict()), 201    except Exception as e:        app.logger.error(f'用户创建失败:{str(e)}')        db.session.rollback()        return jsonify({'error': '创建失败'}), 500

5. CORS 跨域

🌍 什么是 CORS?

浏览器的安全策略:前端页面和后端 API 不在同一个域名下时,默认不允许请求。

# 前端:http://localhost:3000  (React/Vue 开发服务器)# 后端:http://localhost:5000  (Flask)# 浏览器会阻止!报错:CORS policy: No 'Access-Control-Allow-Origin' header# CORS = Cross-Origin Resource Sharing(跨源资源共享)# 告诉浏览器:"这个跨域请求是我允许的,放行!"

🔧 Flask-CORS

# 安装pip install flask-cors# 使用from flask_cors import CORSapp = Flask(__name__)# 方式1:允许所有来源(开发时用)CORS(app)# 方式2:只允许特定来源(生产环境)CORS(app, origins=['https://myfrontend.com', 'https://www.myfrontend.com'])# 方式3:精细控制CORS(app, resources={    '/api/*': {        'origins': ['https://myfrontend.com'],        'methods': ['GET', 'POST', 'PUT', 'DELETE'],        'allow_headers': ['Content-Type', 'Authorization'],        'max_age': 3600  # 预检请求缓存 1 小时    }})

💡 CORS 什么时候需要?• 前后端分离(React/Vue + Flask API)→ 需要• 同源(前端模板由 Flask 渲染)→ 不需要• 移动端 App 调用 API → 不需要(CORS 只是浏览器限制)• 第三方 API 调用 → 不需要


6. Flask 扩展生态

🧰 常用 Flask 扩展

扩展
用途
安装
Flask-SQLAlchemy
ORM 数据库
pip install flask-sqlalchemy
Flask-Migrate
数据库迁移
pip install flask-migrate
Flask-Login
用户认证
pip install flask-login
Flask-WTF
表单处理
pip install flask-wtf
Flask-CORS
跨域支持
pip install flask-cors
Flask-Mail
发送邮件
pip install flask-mail
Flask-Caching
缓存
pip install flask-caching
Flask-Limiter
请求限流
pip install flask-limiter
Flask-Marshmallow
序列化/反序列化
pip install flask-marshmallow
Flask-RESTful
快速构建 REST API
pip install flask-restful

📦 应用工厂模式

当项目变大时,用"工厂函数"创建应用——方便测试和多环境部署。

# app/__init__.pyfrom flask import Flaskfrom flask_sqlalchemy import SQLAlchemyfrom flask_login import LoginManagerfrom flask_cors import CORSdb = SQLAlchemy()login_manager = LoginManager()cors = CORS()def create_app(config_name='development'):    """应用工厂函数"""    app = Flask(__name__)    # 加载配置    if config_name == 'production':        app.config.from_object('config.ProductionConfig')    else:        app.config.from_object('config.DevelopmentConfig')    # 初始化扩展    db.init_app(app)    login_manager.init_app(app)    cors.init_app(app)    # 注册蓝图    from api.users import users_bp    from api.posts import posts_bp    app.register_blueprint(users_bp)    app.register_blueprint(posts_bp)    # 注册错误处理器    register_error_handlers(app)    return appdef register_error_handlers(app):    @app.errorhandler(404)    def not_found(e):        return {'code': 404, 'message': 'Not Found'}, 404    @app.errorhandler(500)    def internal_error(e):        db.session.rollback()        return {'code': 500, 'message': 'Internal Error'}, 500
# run.py(启动文件)from app import create_appapp = create_app()if __name__ == '__main__':    app.run(debug=True)

💡 应用工厂模式的好处:• 测试时可以创建独立的应用实例(用 test 配置)• 可以创建多个不同配置的实例• 避免循环导入问题• 是 Flask 官方推荐的最佳实践


7. 今日练习

🏋️ 练习 1:配置管理

# 创建三套环境配置:# - DevelopmentConfig: debug=True, SQLite# - TestingConfig: testing=True, 内存数据库# - ProductionConfig: debug=False, MySQL# 通过环境变量 FLASK_ENV 切换

🏋️ 练习 2:请求日志中间件

# 用 before_request + after_request 实现:# - 记录每个请求的方法、路径、IP# - 记录响应状态码和处理时间# - 输出格式:[2026-05-12 13:00:00] GET /api/users 200 0.023s 127.0.0.1

🏋️ 练习 3:应用工厂

# 把之前 Day56 的 CRUD API 重构为应用工厂模式:# app/# ├── __init__.py    (create_app 工厂)# ├── api/# │   ├── users.py   (users_bp)# │   └── posts.py   (posts_bp)# ├── models.py# └── config.py# run.py

8. 今日小结

知识点
核心内容
配置管理
Config 类、from_object、环境变量、敏感配置不入库
请求钩子
before_request / after_request / teardown
错误处理
errorhandler + 自定义 APIError + 统一错误格式
日志系统
logging 模块、RotatingFileHandler、日志级别
CORS
Flask-CORS 解决前后端分离的跨域问题
应用工厂
create_app() 模式,方便测试和多环境部署

🚀 明日预告:Day 58 — 博客系统综合项目前 8 天学的 Flask 知识终于要"合体"了!明天我们要做一个完整的博客系统——注册登录、文章 CRUD、分页、Markdown 渲染、评论功能。把路由、模板、表单、数据库、认证全部串起来!

轻松时刻:

最新文章

随机文章