# Flask - 轻量级 Python Web 框架 ## 概述 Flask 是一个用于 Python Web 开发的微框架，旨在构建微服务、REST API 和灵活的 Web 应用程序。其极简的核心和广泛的扩展生态系统使其成为需要轻量级架构、快速开发和对组件完全控制的项目的理想选择。 **主要特点**： * 微框架理念（最小核心，可扩展） * Flask-RESTful 用于 API 开发 * 模块化应用结构蓝图 * 通过 Flask-SQLAlchemy 集成 SQLAlchemy * Jinja2 模板引擎 * 内置开发服务器，具有自动重新加载功能 * Werkzeug WSGI 工具包基础 * 大型扩展生态系统（Flask-Login、Flask-JWT、Flask-CORS） * 使用 Gunicorn/uWSGI 进行生产部署 **安装**： # 基本烧瓶点安装烧瓶 # 具有常用扩展的 Flask pip 安装flaskflask-restfulflask-sqlalchemyflask-loginflask-cors # 有数据库支持 pip 安装flask Flask-sqlalchemy psycopg2-binary # PostgreSQL # 完整的微服务堆栈 pip 安装 Flask Flask-restful Marshmallow Flask-jwt-扩展 Redis ## 基本应用模式 ### 1\.最小烧瓶应用程序 # 应用程序.py 从烧瓶导入烧瓶，jsonify，请求应用程序=烧瓶（__名称__） @app.route('/') 定义你好（）： return jsonify({"message": "你好，世界！"}) @app.route('/api/users/') def get_user(user_id): return jsonify({"id": user_id, "name": f"用户 {user_id}"}) @app.route('/api/users', 方法=['POST']) def create_user(): 数据 = request.get_json() 返回 jsonify({"id": 123, **数据}), 201 如果 __name__ == '__main__': app.run（调试=True，主机='0.0.0.0'，端口=5000） **运行：** # 开发服务器蟒蛇应用程序.py # 或者使用 Flask CLI 导出 FLASK_APP=app.py 导出 FLASK_ENV=开发烧瓶运行 # 自定义端口烧瓶运行--端口8000--主机0.0.0.0 ### 2\.应用工厂模式 **推荐用于生产和测试：** # 应用程序/__init__.py 从烧瓶导入烧瓶从 app.config 导入配置从 app.extensions 导入 db、migrate、jwt def create_app(config_class=Config): """应用工厂模式。""" 应用程序=烧瓶（__名称__） app.config.from_object(config_class) # 初始化扩展 db.init_app(应用程序) migrate.init_app（应用程序，数据库） jwt.init_app(应用程序) # 注册蓝图从 app.routes 导入 api_bp, auth_bp app.register_blueprint(api_bp, url_prefix='/api') app.register_blueprint(auth_bp, url_prefix='/auth') 返回应用程序 # 应用程序/扩展.py 从flask_sqlalchemy导入SQLAlchemy 从flask_migrate导入迁移从 Flask_jwt_extended 导入 JWTManager 数据库 = SQLAlchemy() 迁移 = 迁移() jwt = JWTManager() # 应用程序/config.py 导入操作系统类配置： SECRET_KEY = os.environ.get('SECRET_KEY') 或 'dev-secret-key' SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') 或 'sqlite:///app.db' SQLALCHEMY_TRACK_MODIFICATIONS = False JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY') 或 'jwt-secret' 类开发配置（配置）：调试=真测试 = 错误类 ProductionConfig(配置): 调试=假测试 = 错误 # 运行.py 从应用程序导入create_app 应用程序 = 创建应用程序（）如果 __name__ == '__main__': 应用程序.run() **运行：** 导出 FLASK_APP=run.py 烧瓶运行 ### 3\.请求和响应处理从烧瓶导入烧瓶，请求，jsonify，make_response，中止应用程序=烧瓶（__名称__） @app.route('/api/data',methods=['GET', 'POST']) defhandle_data(): # 获取请求如果 request.method == 'GET': # 查询参数 page = request.args.get('page', 1, type=int) limit = request.args.get('limit', 10, type=int) 返回 jsonify({ “页面”：页面， “限制”：限制， “数据”：[...] }) # POST 请求如果 request.method == 'POST': # JSON 正文数据 = request.get_json() # 验证如果不是数据或“名称”不在数据中：中止（400，描述=“缺少必填字段：名称”） # 带标头的自定义响应响应 = make_response(jsonify({"id": 1, **数据}), 201) response.headers['X-Custom-Header'] = '值' 返回响应 # 错误处理 @app.errorhandler(404) def not_found（错误）： return jsonify({"error": "资源未找到"}), 404 @app.errorhandler(400) def bad_request（错误）：返回 jsonify({"error": str(error.description)}), 400 ## 蓝图 - 模块化应用程序 ### 蓝图结构应用程序/ ├── __init__.py # 应用工厂 ├── extensions.py # 扩展实例 ├── config.py # 配置 ├── 型号/ │ ├── __init__.py │ ├── 用户.py │ └── 产品.py ├── 路线/ │ ├── __init__.py │ ├── auth.py # 认证路由 │ ├── users.py # 用户管理路由 │ └── products.py # 产品路线 └── 服务/ ├── __init__.py ├── user_service.py └── auth_service.py ### 蓝图实施 # 应用程序/路线/users.py 从烧瓶导入蓝图、jsonify、请求从 app.models.user 导入用户从 app.extensions 导入 db users_bp = 蓝图('用户', __name__) @users_bp.route('/',methods=['GET']) def list_users(): """列出所有用户。""" page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 20, type=int) 用户 = User.query.paginate(page=page, per_page=per_page) 返回 jsonify({ "users": [u.to_dict() for u in users.items], “总计”：用户总数， “页面”：用户.页面， “页面”：用户.页面 }) @users_bp.route('/',methods=['GET']) def get_user(user_id): """通过ID获取用户。""" 用户 = User.query.get_or_404(user_id) 返回 jsonify(user.to_dict()) @users_bp.route('/',methods=['POST']) def create_user(): """创建新用户。""" 数据 = request.get_json() 用户 = 用户（电子邮件=数据['电子邮件'], 名称=数据['名称'] ） user.set_password(data['密码']) db.session.add(用户) db.session.commit() 返回 jsonify(user.to_dict()), 201 @users_bp.route('/',methods=['PUT']) def update_user(user_id): """更新用户。""" 用户 = User.query.get_or_404(user_id) 数据 = request.get_json() 如果数据中有“名称”：用户名 = 数据['姓名'] 如果数据中有“电子邮件”： user.email = 数据['电子邮件'] db.session.commit() 返回 jsonify(user.to_dict()) @users_bp.route('/',methods=['DELETE']) def delete_user(user_id): """删除用户。""" 用户 = User.query.get_or_404(user_id) db.session.delete(用户) db.session.commit() 返回 '', 204 # 应用程序/__init__.py def create_app(): 应用程序=烧瓶（__名称__） # 注册蓝图从 app.routes.users 导入 users_bp 从 app.routes.products 导入 products_bp app.register_blueprint(users_bp, url_prefix='/api/users') app.register_blueprint(products_bp, url_prefix='/api/products') 返回应用程序 ## Flask-RESTful API ### 基本 REST API # 应用程序/api/resources.py 来自烧瓶导入请求从flask_restful导入资源、Api、reqparse、字段、marshal_with 从 app.models.user 导入用户从 app.extensions 导入 db # 响应序列化用户字段 = { 'id': fields.Integer, '电子邮件': fields.String, '名称': fields.String, 'created_at': fields.DateTime(dt_format='iso8601') } 类用户列表资源（资源）： """用户收集端点。""" @marshal_with(用户字段) def get(自身): """列出所有用户。""" 用户 = User.query.all() 返回用户定义帖子（自我）： """创建新用户。""" 解析器 = reqparse.RequestParser() parser.add_argument('email', required=True, help='电子邮件为必填项') parser.add_argument('name', required=True, help='名称为必填项') parser.add_argument('密码', required=True, help='需要密码') args = parser.parse_args() 用户=用户（电子邮件=参数['电子邮件']，名称=参数['名称']） user.set_password(args['密码']) db.session.add(用户) db.session.commit() 返回 {'id': user.id, 'email': user.email, 'name': user.name}, 201 用户资源类（资源）： """单用户端点。""" @marshal_with(用户字段) def get(自身, user_id): """通过ID获取用户。""" 用户 = User.query.get_or_404(user_id) 返回用户 @marshal_with(用户字段) def put(self, user_id): """更新用户。""" 用户 = User.query.get_or_404(user_id) 解析器 = reqparse.RequestParser() parser.add_argument('名称') parser.add_argument('电子邮件') args = parser.parse_args() 如果参数['名称']： user.name = args['名称'] if args['电子邮件']： user.email = args['电子邮件'] db.session.commit() 返回用户 def 删除(自身, user_id): """删除用户。""" 用户 = User.query.get_or_404(user_id) db.session.delete(用户) db.session.commit() 返回 '', 204 # 应用程序/__init__.py 从 Flask_restful 导入 API def create_app(): 应用程序=烧瓶（__名称__） api = Api(应用程序, 前缀='/api/v1') # 注册资源从 app.api.resources 导入 UserListResource、UserResource api.add_resource(UserListResource, '/users') api.add_resource(UserResource, '/users/') 返回应用程序 ## 请求验证 ### 棉花糖模式 # 应用程序/模式/user_schema.py 从棉花糖导入架构、字段、验证、验证、ValidationError 用户模式类（模式）： """用户验证架构。""" id = fields.Int(dump_only=True) 电子邮件 = fields.Email(required=True, validate=validate.Length(max=120)) 名称 = fields.Str(required=True, validate=validate.Length(min=2, max=80)) 密码 = 字段.Str( 必需=真， load_only=真，验证=验证.长度(分钟=8) ）创建的_at = fields.DateTime(dump_only=True) @validates('电子邮件') def validate_email(自身, 值): """检查电子邮件的唯一性。""" 从 app.models.user 导入用户 if User.query.filter_by(email=value).first(): raise ValidationError('电子邮件已注册') 类 UserUpdateSchema（架构）： """用户更新架构（部分更新）。""" 电子邮件 = fields.Email(validate=validate.Length(max=120)) 名称 = fields.Str(validate=validate.Length(min=2, max=80)) 密码 = fields.Str(load_only=True, validate=validate.Length(min=8)) # 路由中的用法从棉花糖导入验证错误从 app.schemas.user_schema 导入 UserSchema、UserUpdateSchema user_schema = UserSchema() users_schema = UserSchema(很多=True) user_update_schema = UserUpdateSchema() @users_bp.route('/',methods=['POST']) def create_user(): """创建用户并进行验证。""" 尝试： # 验证和反序列化数据 = user_schema.load(request.get_json()) 除了 ValidationError 为错误：返回 jsonify({"errors": err.messages}), 400 用户 = 用户（**数据） db.session.add(用户) db.session.commit() # 序列化响应返回 user_schema.dump(用户), 201 @users_bp.route('/',methods=['GET']) def list_users(): """通过序列化列出用户。""" 用户 = User.query.all() 返回 jsonify(users_schema.dump(users)) ### Pydantic 验证 # 应用程序/模式/user_schema_pydantic.py 从 pydantic 导入 BaseModel、EmailStr、Field、验证器从输入 import 可选从日期时间导入日期时间类 UserCreate(BaseModel): """用户创建架构。""" 电子邮件：EmailStr 名称：str = 字段(..., min_length=2, max_length=80) 密码：str = Field(..., min_length=8) @validator('密码') def 密码复杂度(cls, v): """验证密码复杂性。""" 如果没有（v 中的 char.isdigit()）： raise ValueError('密码必须至少包含一位数字') 如果没有（v 中的 char.isupper()）： raise ValueError('密码必须至少包含一个大写字母') 返回v 类 UserUpdate(BaseModel): """用户更新架构。""" 电子邮件：可选[EmailStr] = 无名称：可选[str] =字段（无，min_length = 2，max_length = 80）类用户响应（基础模型）： """用户响应架构。""" id：整数电子邮件：str 名称：str 创建时间：日期时间类配置： orm_模式=真 # 用法从 pydantic 导入验证错误 @users_bp.route('/',methods=['POST']) def create_user(): """使用 Pydantic 验证创建用户。""" 尝试： user_data = UserCreate(**request.get_json()) 除了 ValidationError 为 e：返回 jsonify({"errors": e.errors()}), 400 用户 = 用户（电子邮件=用户数据.电子邮件，名称=用户数据.名称） user.set_password(user_data.password) db.session.add(用户) db.session.commit() 返回 UserResponse.from_orm(user).dict(), 201 ## SQLAlchemy 集成 ### 使用 Flask-SQLAlchemy 的模型 # 应用程序/模型/user.py 从日期时间导入日期时间从 werkzeug.security 导入generate_password_hash、check_password_hash 从 app.extensions 导入 db 用户类（db.Model）： """用户模型。""" __表名__ = '用户' id = db.Column(db.Integer, Primary_key=True) 电子邮件= db.Column（db.String（120），唯一= True，可为空= False，索引= True）名称 = db.Column(db.String(80), nullable=False) 密码哈希 = db.Column(db.String(200), nullable=False) is_active = db.Column(db.Boolean, 默认=True) create_at = db.Column(db.DateTime, 默认=datetime.utcnow) Updated_at = db.Column（db.DateTime，默认= datetime.utcnow，onupdate = datetime.utcnow） # 关系 posts = db.relationship（'Post'，backref ='author'，lazy ='dynamic'，cascade ='all，delete-orphan'） def set_password（自我，密码）： """散列并设置密码。""" self.password_hash = 生成_password_hash(密码) def check_password(自我, 密码): """验证密码。""" 返回 check_password_hash(self.password_hash, 密码) def to_dict(自我): """序列化到字典。""" 返回{ 'id': 自我 ID, '电子邮件': self.email, 'name': 自己的名字， 'is_active': self.is_active, 'created_at': self.created_at.isoformat(), 'updated_at': self.updated_at.isoformat() } def __repr__(自我): 返回 f'<用户 {self.email}>' # 应用程序/模型/post.py 类帖子（db.Model）： “”“后模型。”“” __表名__ = '帖子' id = db.Column(db.Integer, Primary_key=True) 标题 = db.Column(db.String(200), nullable=False) 内容 = db.Column(db.Text, nullable=False) user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False) create_at = db.Column(db.DateTime, 默认=datetime.utcnow) def to_dict(自我): 返回{ 'id': 自我 ID, '标题'：自我标题， '内容'：自我内容， 'user_id': self.user_id, '作者': self.作者.姓名, 'created_at': self.created_at.isoformat() } # 数据库操作 @users_bp.route('//posts',methods=['GET']) def get_user_posts(user_id): """获取带有分页的用户帖子。""" 用户 = User.query.get_or_404(user_id) page = request.args.get('page', 1, type=int) posts = user.posts.paginate(page=page, per_page=20) 返回 jsonify({ 'posts': [p.to_dict() for p in posts.items], '总计'：帖子总数， '页面': posts.page }) ### 数据库迁移 # 初始化迁移烧瓶数据库初始化 # 创建迁移 Flask db migrate -m“创建用户表” # 应用迁移烧瓶数据库升级 # 回滚烧瓶数据库降级 # migrations/versions/xxx_create_users.py （自动生成） def 升级(): op.create_table( '用户', sa.Column('id', sa.Integer(), nullable=False), sa.Column('email', sa.String(length=120), nullable=False), sa.Column('名称', sa.String(length=80), nullable=False), sa.Column('password_hash', sa.String(length=200), nullable=False), sa.Column('created_at', sa.DateTime(), nullable=True), sa.PrimaryKeyConstraint('id'), sa.UniqueConstraint('电子邮件') ） op.create_index('ix_users_email', '用户', ['电子邮件']) def降级（）： op.drop_index('ix_users_email', table_name='users') op.drop_table('用户') ## 身份验证 ### Flask-登录基于会话的身份验证 # 应用程序/扩展.py 从flask_login导入LoginManager 登录管理器 = 登录管理器() # 应用程序/__init__.py def create_app(): 应用程序=烧瓶（__名称__）登录管理器.init_app(应用程序) login_manager.login_view = 'auth.login' @login_manager.user_loader def load_user(user_id): 从 app.models.user 导入用户返回 User.query.get(int(user_id)) 返回应用程序 # 应用程序/模型/user.py 从 Flask_login 导入 UserMixin 用户类（UserMixin，db.Model）： # ...现有字段... def get_id(自身): 返回 str(self.id) # 应用程序/路由/auth.py 从烧瓶导入蓝图、请求、jsonify 从flask_login导入login_user，logout_user，login_required，current_user 从 app.models.user 导入用户 auth_bp = 蓝图('auth', __name__) @auth_bp.route('/login',methods=['POST']) def 登录(): """用户登录。""" 数据 = request.get_json() user = User.query.filter_by(email=data['email']).first() 如果不是用户或不是 user.check_password(data['password']): return jsonify({"error": "无效凭证"}), 401 login_user(用户, Remember=data.get('记住', False)) 返回 jsonify(user.to_dict()) @auth_bp.route('/logout',methods=['POST']) @login_required def 注销(): """用户注销。""" 注销_用户() return jsonify({"message": "登出成功"}) @auth_bp.route('/me',methods=['GET']) @login_required def get_current_user(): """获取当前经过身份验证的用户。""" 返回 jsonify(current_user.to_dict()) ### JWT 身份验证 # 应用程序/扩展.py 从 Flask_jwt_extended 导入 JWTManager jwt = JWTManager() # 应用程序/__init__.py def create_app(): 应用程序=烧瓶（__名称__） app.config['JWT_SECRET_KEY'] = '超级密钥' app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(小时=1) app.config['JWT_REFRESH_TOKEN_EXPIRES'] = timedelta(天=30) jwt.init_app(应用程序) 返回应用程序 # 应用程序/路由/auth.py 从flask_jwt_扩展导入（创建访问令牌，创建刷新令牌， jwt_必需，获取jwt_身份） @auth_bp.route('/login',methods=['POST']) def 登录(): """登录并返回 JWT 令牌。""" 数据 = request.get_json() user = User.query.filter_by(email=data['email']).first() 如果不是用户或不是 user.check_password(data['password']): return jsonify({"error": "无效凭证"}), 401 access_token = create_access_token(identity=user.id) fresh_token = create_refresh_token(identity=user.id) 返回 jsonify({ “访问令牌”：访问令牌， “刷新令牌”：刷新令牌， “用户”：user.to_dict() }) @auth_bp.route('/refresh',methods=['POST']) @jwt_required（刷新=真） def刷新（）： """刷新访问令牌。""" 当前用户 ID = get_jwt_identity() access_token = create_access_token(身份=当前用户id) 返回 jsonify({"access_token": access_token}) @auth_bp.route('/me',methods=['GET']) @jwt_required() def get_current_user(): """从 JWT 获取当前用户。""" 当前用户 ID = get_jwt_identity() 用户 = User.query.get_or_404(current_user_id) 返回 jsonify(user.to_dict()) # 受保护的路由示例 @users_bp.route('/',methods=['DELETE']) @jwt_required() def delete_user(user_id): """删除用户（已验证）。""" 当前用户 ID = get_jwt_identity() # 权限检查如果 current_user_id != user_id: 返回 jsonify({"error": "未经授权"}), 403 用户 = User.query.get_or_404(user_id) db.session.delete(用户) db.session.commit() 返回 '', 204 ## 配置管理 ### 基于环境的配置 # 应用程序/config.py 导入操作系统从日期时间导入时间增量类配置： “”“基本配置。”“” # 烧瓶 SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key-change-in-product') # SQLAlchemy SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL', 'sqlite:///app.db') SQLALCHEMY_TRACK_MODIFICATIONS = False SQLALCHEMY_ECHO = 假 # 智威汤逊 JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY', 'jwt-secret-change-in-生产') JWT_ACCESS_TOKEN_EXPIRES = timedelta(小时=1) JWT_REFRESH_TOKEN_EXPIRES = timedelta(天=30) # 跨域资源共享 CORS_ORIGINS = os.getenv('CORS_ORIGINS', '*').split(',') # 分页每页项目 = 20 每页最多项目数 = 100 # 日志记录 LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO') LOG_FILE = os.getenv('LOG_FILE', 'app.log') 类开发配置（配置）： """开发配置。""" 调试=真测试 = 错误 SQLALCHEMY_ECHO = True 类测试配置（配置）： """测试配置。""" 测试 = 正确 SQLALCHEMY_DATABASE_URI = 'sqlite:///:内存：' WTF_CSRF_ENABLED = 假类 ProductionConfig(配置): """生产配置。""" 调试=假测试 = 错误 # 严格的安全要求 SECRET_KEY = os.getenv('SECRET_KEY') JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY') # 生产数据库（推荐PostgreSQL） SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL') @类方法 def init_app(cls, 应用程序): """特定于生产的初始化。""" # 记录到系统日志或外部服务导入日志记录从logging.handlers导入SysLogHandler syslog_handler = SysLogHandler() syslog_handler.setLevel(日志记录.警告) app.logger.addHandler(syslog_handler) # 配置工厂配置名称 = { '开发'：开发配置， “测试”：测试配置， “生产”：生产配置， “默认”：开发配置 } def get_config(env_name='default'): """通过环境名称获取配置。""" 返回 config_by_name.get(env_name, DevelopmentConfig) # 应用程序/__init__.py def create_app(config_name='default'): 应用程序=烧瓶（__名称__） app.config.from_object(get_config(config_name)) 返回应用程序 ### .env 文件支持 # .env FLASK_APP=运行.py FLASK_ENV=开发 SECRET_KEY=您的秘密密钥-此处 JWT_SECRET_KEY=此处为您的 jwt-secret DATABASE_URL=postgresql://用户:密码@localhost/dbname CORS_ORIGINS=http://localhost:3000,http://localhost:8000 # 加载环境变量从 dotenv 导入 load_dotenv 加载_dotenv() # 应用程序/__init__.py 导入操作系统从 dotenv 导入 load_dotenv def create_app(): 加载_dotenv() config_name = os.getenv('FLASK_ENV', '开发') 应用程序=烧瓶（__名称__） app.config.from_object(get_config(config_name)) 返回应用程序 ## 错误处理和日志记录 ### 全局错误处理程序 # 应用程序/错误/handlers.py 从烧瓶导入 jsonify 从 werkzeug.exceptions 导入 HTTPException 从 sqlalchemy.exc 导入 SQLAlchemyError def register_error_handlers(应用程序): """注册全局错误处理程序。""" @app.errorhandler(404) def not_found_error（错误）：返回 jsonify({ “错误”：“未找到”， "message": "未找到请求的资源" }), 404 @app.errorhandler(400) def bad_request_error（错误）：返回 jsonify({ "error": "错误的请求", "message": str(error.description) if hasattr(error, 'description') else "无效请求" }), 400 @app.errorhandler(401) def 未经授权的错误（错误）：返回 jsonify({ "error": "未经授权", "message": "需要身份验证" }), 401 @app.errorhandler(403) def禁止错误（错误）：返回 jsonify({ “错误”：“禁止”， "message": "您无权访问该资源" }), 403 @app.errorhandler(500) def 内部错误（错误）： app.logger.error(f'内部服务器错误: {error}') db.session.rollback() 返回 jsonify({ "error": "内部服务器错误", "message": "发生意外错误" }), 500 @app.errorhandler(HTTPException) defhandle_http_exception（错误）： """处理所有 HTTP 异常。""" 返回 jsonify({ “错误”：错误名称， “消息”：错误.描述 }), 错误代码 @app.errorhandler(SQLAlchemyError) defhandle_db_error（错误）： """处理数据库错误。""" app.logger.error(f'数据库错误：{error}') db.session.rollback() 返回 jsonify({ "error": "数据库错误", "message": "数据库发生错误" }), 500 # 应用程序/__init__.py def create_app(): 应用程序=烧瓶（__名称__）从 app.errors.handlers 导入 register_error_handlers 注册错误处理程序（应用程序）返回应用程序 ### 日志配置 # 应用程序/logging_config.py 导入日志记录从logging.handlers导入RotatingFileHandler 导入操作系统 def 配置_日志记录（应用程序）： """配置应用程序日志记录。""" 如果不是 app.debug 和 app.testing： # 如果logs目录不存在则创建如果不是 os.path.exists('logs'): os.mkdir('日志') # 带旋转的文件处理程序 file_handler = 旋转文件处理程序( '日志/flask_app.log', 最大字节数=10240000, # 10MB 备份计数=10 ） file_handler.setFormatter(日志记录.Formatter( '%(asctime)s %(levelname)s: %(message)s ' '[在 %(路径名)s:%(lineno)d]' ）） file_handler.setLevel(日志记录.INFO) app.logger.addHandler(file_handler) app.logger.setLevel(logging.INFO) app.logger.info('Flask应用程序启动') # 应用程序/__init__.py def create_app(): 应用程序=烧瓶（__名称__）从 app.logging_config 导入 configure_logging 配置日志记录（应用程序）返回应用程序 # 路由中的用法 @users_bp.route('/',methods=['POST']) def create_user(): current_app.logger.info(f'创建新用户：{request.get_json()}') # ...创建用户... current_app.logger.info(f'用户创建成功：{user.id}') ## 使用 pytest 进行测试 ### 测试配置 #conftest.py 导入pytest 从应用程序导入create_app 从 app.extensions 导入 db 从 app.models.user 导入用户 @pytest.fixture(范围='会话') def 应用程序（）： """创建测试应用程序。""" 应用程序 = create_app('测试') 与 app.app_context(): db.create_all() 收益应用程序 db.drop_all() @pytest.fixture def 客户端（应用程序）： """Flask 测试客户端。""" 返回app.test_client() @pytest.fixture def 跑步者（应用程序）： """Flask CLI 测试运行器。""" 返回 app.test_cli_runner() @pytest.fixture def db_session(应用程序): """用于测试的数据库会话。""" 与 app.app_context(): db.session.begin_nested() 产生数据库会话 db.session.rollback() @pytest.fixture def 样本用户(db_session): """创建示例用户。""" user = User(email='test@example.com', name='测试用户') user.set_password('密码123') db_session.add（用户） db_session.commit() 返回用户 @pytest.fixture def auth_headers（客户端，sample_user）： """获取 JWT 身份验证标头。""" 响应 = client.post('/auth/login', json={ '电子邮件': 'test@example.com', '密码': '密码123' }) 令牌 = response.get_json()['access_token'] 返回 {'授权': f'Bearer {token}'} ### API 测试 # 测试/test_users.py 导入pytest 从 app.models.user 导入用户 def test_create_user(客户端): """测试用户创建端点。""" 响应 = client.post('/api/users', json={ '电子邮件': 'new@example.com', 'name': '新用户', '密码': '密码123' }) 断言response.status_code == 201 数据=response.get_json() 断言数据['电子邮件'] == 'new@example.com' 断言数据['名称'] == '新用户' 在数据中断言“id” def test_get_user（客户端，样本用户）： """测试获取用户端点。""" 响应 = client.get(f'/api/users/{sample_user.id}') 断言response.status_code == 200 数据=response.get_json() 断言数据['id'] ==sample_user.id 断言数据['电子邮件'] == Sample_user.email def test_list_users（客户端，样本用户）： """测试列表用户端点。""" 响应 = client.get('/api/users') 断言response.status_code == 200 数据=response.get_json() 在数据中断言“用户” 断言 len(数据['用户']) > 0 def test_update_user（客户端，sample_user，auth_headers）： """测试用户更新端点。""" 响应 = 客户端.put( f'/api/users/{sample_user.id}', json={'name': '更新后的名称'}, 标头=auth_headers ）断言response.status_code == 200 数据=response.get_json() 断言 data['name'] == '更新后的名称' def test_delete_user（客户端，sample_user，auth_headers）： """测试用户删除端点。""" 响应 = 客户端.删除( f'/api/users/{sample_user.id}', 标头=auth_headers ）断言response.status_code == 204 断言 User.query.get(sample_user.id) 为 None def test_authentication_required（客户端，sample_user）： """测试受保护端点是否需要身份验证。""" 响应 = client.delete(f'/api/users/{sample_user.id}') 断言response.status_code == 401 ## 部署 ### 使用 Gunicorn 进行生产 # 安装Gunicorn 点安装gunicorn # 与 Gunicorn 一起运行 Gunicorn --workers 4 --bind 0.0.0.0:8000 "app:create_app()" # 带环境变量 Gunicorn --workers 4 --bind 0.0.0.0:8000 --env FLASK_ENV=生产“app:create_app()” **gunicorn.conf.py：** 导入多重处理 # 服务器套接字绑定='0.0.0.0:8000' 积压 = 2048 # 工作进程工人 = multiprocessing.cpu_count() * 2 + 1 工人类别 = '同步' 工人连接数 = 1000 超时= 30 保持活动= 2 # 日志记录 accesslog = '日志/access.log' 错误日志 = '日志/错误.log' 日志级别 = '信息' # 进程命名 proc_name = 'flask-app' # 服务器机制守护进程=假 pidfile = 'gunicorn.pid' ### Docker 部署 **Docker 文件：** 来自 python:3.11-slim 工作目录/应用程序 # 安装依赖项复制requirements.txt。运行 pip install --no-cache-dir -rrequirements.txt # 复制应用程序复制。。 # 创建非root用户运行 useradd -m -u 1000 烧瓶 && chown -R 烧瓶:烧瓶 /app 用户烧瓶 # 与 Gunicorn 一起运行 CMD [“gunicorn”，“--config”，“gunicorn.conf.py”，“app：create_app（）”] **docker-compose.yml：** 版本：'3.8' 服务：应用程序：构建： . 端口： - “8000:8000” 环境： - FLASK_ENV=生产 - DATABASE_URL=postgresql://用户:密码@db:5432/flask_db - SECRET_KEY=${SECRET_KEY} - JWT_SECRET_KEY=${JWT_SECRET_KEY} 取决于： - 数据库 - 雷迪斯卷： - ./logs:/应用程序/日志数据库：图片：postgres：15-alpine 环境： - POSTGRES_USER=用户 - POSTGRES_PASSWORD=密码 - POSTGRES_DB=flask_db 卷： - postgres_data:/var/lib/postgresql/data 雷迪斯：图片：redis:7-alpine 卷： - redis_data:/数据 nginx：图片：nginx:alpine 端口： - “80:80” 卷： - ./nginx.conf:/etc/nginx/nginx.conf:ro 取决于： - 应用程序卷： postgres_数据： redis_数据： ## 微服务模式 ### 服务沟通 # 应用程序/服务/external_api.py 导入请求从烧瓶导入 current_app 类外部API服务： """外部 API 客户端服务。""" def __init__(self, base_url, 超时=10): self.base_url = base_url self.timeout = 超时 def get(自身, 端点, **kwargs): """对外部 API 的 GET 请求。""" url = f"{self.base_url}/{端点}" 尝试：响应 = requests.get(url, timeout=self.timeout, **kwargs) 响应.raise_for_status() 返回response.json() 除了 requests.RequestException 作为 e： current_app.logger.error(f'外部 API 错误：{e}') 提高 def post（自身，端点，数据，**kwargs）： """向外部 API 发出 POST 请求。""" url = f"{self.base_url}/{端点}" 尝试：响应 = requests.post(url, json=data, timeout=self.timeout, **kwargs) 响应.raise_for_status() 返回response.json() 除了 requests.RequestException 作为 e： current_app.logger.error(f'外部 API 错误：{e}') 提高 # 用法 api_service =ExternalAPIService('https://api.example.com') @app.route('/proxy/data') def proxy_data(): """将请求代理到外部服务。""" 尝试：数据 = api_service.get('数据') 返回 jsonify(数据) 除了异常 e： return jsonify({"error": "外部服务不可用"}), 503 ### 健康检查 # 应用程序/路线/health.py 从烧瓶导入蓝图，jsonify 从 app.extensions 导入 db 从 sqlalchemy 导入文本 health_bp = 蓝图('health', __name__) @health_bp.route('/health') def health_check(): “”“基本健康检查。”“” 返回 jsonify({"status": "healthy"}), 200 @health_bp.route('/health/ready') def readiness_check(): """检查依赖关系的准备情况。""" 检查= { “数据库”：错误， “状态”：“不健康” } # 检查数据库尝试： db.session.execute(text('SELECT 1')) 检查[“数据库”] = True 除了异常 e： current_app.logger.error(f'数据库健康检查失败：{e}') # 整体状态 checks["status"] = "healthy" if all([checks["database"]]) else "unhealthy" status_code = 200 如果检查["status"] == "healthy" else 503 返回 jsonify(检查), status_code ## 最佳实践 ### 1\.使用应用程序工厂模式 # ✅ 好：应用程序工厂 def create_app(config_name='default'): 应用程序=烧瓶（__名称__） app.config.from_object(get_config(config_name)) 返回应用程序 # ❌ 不好：全局应用程序实例应用程序=烧瓶（__名称__） ### 2\.使用模块化蓝图 # ✅ 好：有组织的蓝图从 app.routes.users 导入 users_bp 从 app.routes.products 导入 products_bp app.register_blueprint(users_bp, url_prefix='/api/users') app.register_blueprint(products_bp, url_prefix='/api/products') # ❌ 不好：所有路由都在单个文件中 ### 3\.验证所有输入 # ✅ 好：使用 Marshmallow/Pydantic 进行验证尝试：数据 = user_schema.load(request.get_json()) 除了 ValidationError 为 e：返回 jsonify({"errors": e.messages}), 400 # ❌ 不好：没有验证数据 = request.get_json() user = User(**data) # 不安全！ ### 4\.使用环境变量作为机密 # ✅ 好：环境变量 SECRET_KEY = os.getenv('SECRET_KEY') # ❌ 不好：硬编码的秘密 SECRET_KEY = '硬编码密钥' ### 5\.实施正确的错误处理 # ✅ 好：全局错误处理程序 @app.errorhandler（异常） def 句柄_错误（错误）： app.logger.error(f'错误: {error}') return jsonify({"error": "内部服务器错误"}), 500 # ❌ 不好：未处理的异常 ## 资源 * **官方文档**： * **Flask-RESTful** : * **Flask-SQLAlchemy** ： * **Flask-JWT-Extended** : * **棉花糖**： * **Pydantic** ： ## 相关技能使用 Flask 时，请考虑以下补充技能： * **pytest** ：使用固定装置和测试客户端测试 Flask 应用程序 * **sqlalchemy** ：与 Flask-SQLAlchemy 集成的数据库 ORM 模式 * **fastapi-local-dev** ：高性能 API 的现代异步替代方案 * **django** ：包含电池的框架，具有内置管理和 ORM ### 快速 Flask 测试模式（内联以供独立使用） # 使用 pytest 测试 Flask 导入pytest 从烧瓶导入烧瓶从应用程序导入create_app，数据库 @pytest.fixture def 应用程序（）： """创建并配置测试应用程序""" 应用程序 = create_app('测试') 与 app.app_context(): db.create_all() 收益应用程序 db.drop_all() @pytest.fixture def 客户端（应用程序）： """测试客户端发出请求""" 返回app.test_client() @pytest.fixture def 跑步者（应用程序）： """CLI 测试运行程序""" 返回 app.test_cli_runner() # 测试路线 def test_home_page(客户端): 响应 = client.get('/') 断言response.status_code == 200 在response.data中断言b'欢迎' def test_api_endpoint(客户端): 响应 = client.post('/api/users', json={ '用户名': '爱丽丝', “电子邮件”：“alice@example.com” }) 断言response.status_code == 201 数据=response.get_json() 断言数据['用户名'] == 'alice' def test_authentication（客户端）： # 登录响应 = client.post('/登录', data={ '用户名': '爱丽丝', '密码': 'secret123' }) assert response.status_code == 302 # 登录后重定向 # 访问受保护的路由响应 = client.get('/仪表板') 断言response.status_code == 200 # 使用数据库进行测试 def test_user_creation（应用程序）：与 app.app_context(): 用户=用户（用户名='bob'，电子邮件='bob@example.com'） db.session.add(用户) db.session.commit() 找到 = User.query.filter_by(用户名='bob').first() 断言发现不是 None 断言找到.email == 'bob@example.com' # 测试错误处理 def test_404_error（客户端）：响应 = client.get('/不存在') 断言response.status_code == 404 在response.data中断言b'Not Found' ### 快速 SQLAlchemy 集成（内联以供独立使用） # Flask-SQLAlchemy 设置从烧瓶导入烧瓶从flask_sqlalchemy导入SQLAlchemy 从 sqlalchemy.orm 导入 DeclarativeBase 基类（声明性基类）：通过 db = SQLAlchemy(model_class=基础) def create_app(): 应用程序=烧瓶（__名称__） app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@localhost/db' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False db.init_app(应用程序) 与 app.app_context(): db.create_all() 返回应用程序 # 定义模型用户类（db.Model）： __表名__ = '用户' id = db.Column(db.Integer, Primary_key=True) 用户名 = db.Column(db.String(50), unique=True, nullable=False) 电子邮件 = db.Column(db.String(100), unique=True, nullable=False) posts = db.relationship('Post', backref='author', lazy='dynamic') 类帖子（db.Model）： __表名__ = '帖子' id = db.Column(db.Integer, Primary_key=True) 标题 = db.Column(db.String(200), nullable=False) 内容 = db.Column(db.Text, nullable=False) user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False) # 查询模式 @app.route('/users/') def get_user(user_id): user = db.session.get(用户, user_id) 如果不是用户：中止(404) 返回 jsonify({ 'id': 用户 ID, '用户名': 用户.用户名, '电子邮件'：用户.电子邮件， 'posts': [{'title': p.title} for p in user.posts] }) @app.route('/users', 方法=['POST']) def create_user(): 数据 = request.get_json() 用户=用户（用户名=数据['用户名']，电子邮件=数据['电子邮件']） db.session.add(用户) db.session.commit() 返回 jsonify({'id': user.id}), 201 # 事务处理从 sqlalchemy.exc 导入 IntegrityError @app.route('/transfer',methods=['POST']) deftransfer_funds(): 尝试： # 单个事务中的所有操作发件人 = User.query.get_or_404(request.json['sender_id']) 接收者 = User.query.get_or_404(request.json['receiver_id']) 金额 = request.json['金额'] sender.balance -= 金额接收者.balance += 金额 db.session.commit() 返回 jsonify({'status': '成功'}) 除了完整性错误： db.session.rollback() return jsonify({'error': '交易失败'}), 400 ### 快速 FastAPI 比较（内联以供独立使用） **何时选择 FastAPI 而不是 Flask：** * 需要async/await实现高并发 * 想要自动 API 文档 (OpenAPI/Swagger) * 需要内置数据验证（Pydantic） * 构建现代微服务或 GraphQL API * 需要 WebSocket 或服务器发送事件支持 **何时坚持使用烧瓶：** * 构建传统的服务器渲染 Web 应用程序 * 现有的大型 Flask 代码库 * 需要成熟的生态系统和丰富的插件 * 团队熟悉同步Python模式 * 更简单的部署（没有异步运行时复杂性） **迁移注意事项：** # 烧瓶图案 @app.route('/users/') def get_user(user_id): 用户 = User.query.get_or_404(user_id) return jsonify({'id': user.id, 'name': user.name}) # FastAPI 等效项 @app.get('/users/{user_id}', response_model=UserResponse) async def get_user(user_id: int, db: AsyncSession = Depends(get_db)): 用户=等待db.get（用户，user_id）如果不是用户：引发 HTTPException（status_code=404，详细信息=“未找到用户”）返回用户 [如果一起部署，则在各自的技能中可以使用完整的 pytest、SQLAlchemy 和 FastAPI 模式]

烧瓶

包含技能