flask——4：请求与响应

一，请求(Request)对象

在Flask中，客户端(通常是浏览器)发送的所有请求信息都可以通过request对象访问。使用前需要从flask模块导入：

from flask import request

（一）常用请求属性

from urllib import request

from flask import Flask

app = Flask(__name__)


@app.route('/example')
def example():
    # 请求方法 (GET, POST等)
    method = request.method

    # 获取URL参数 (?key=value)
    arg_value = request.args.get('key', 'default_value')  # GET参数
    # 获取URL 中“？”之后原始值
    raw_args = request.query_string.decode('utf-8')

    # 获取表单数据 (Content-Type: application/x-www-form-urlencoded)
    form_value = request.form.get('key', 'default_value')  # POST表单数据

    # 获取JSON数据 (Content-Type: application/json)
    if request.is_json:
        try:
            json_data = request.get_json()
            json_key_value = json_data.get('key', 'default_value')  # JSON数据
        except Exception as e:
            # 处理JSON解析失败的情况
            json_data = request.on_json_loading_failed()  # 如果内容类型不是 application/json，则它将为 None

    # 获取原始请求体数据 (Content-Type: text/plain 或其他)
    data_value = request.data.decode('utf-8') # 原始请求体数据
    data = request.get_data(as_text=True)   # 获取原始请求体数据，as_text=True 将其转换为字符串

    # 获取上传的文件(Content-Type: multipart/form-data)
    uploaded_file = request.files.get('file_key')

    # 请求头
    user_agent = request.headers.get('User-Agent')

    # Cookies
    cookie_value = request.cookies.get('cookie_name')

    # 请求的URL信息
    path = request.path  # /example
    full_path = request.full_path  # /example?key=value
    url = request.url  # http://localhost:5000/example?key=value

    # 获取请求的IP地址
    ip_address = request.remote_addr

    return f"Method: {method}, Arg: {arg_value}"


if __name__ == '__main__':
    app.run()

二，响应(Response)对象

Flask视图函数可以返回多种形式的响应

（一）基本响应

@app.route('/')
def index():
    # 返回字符串 (自动转为Response对象)
    return "Hello, World!"
    
    # 返回元组：(response, status_code, headers)
    # return "Not Found", 404, {'X-Custom-Header': 'Value'}

（二）构建自定义响应

from flask import make_response

@app.route('/custom')
def custom_response():
    # 使用make_response构建响应
    response = make_response("Custom Response", 201)
    response.headers['X-Custom-Header'] = 'Something'
    response.set_cookie('username', 'john_doe')
    return response

（三）返回JSON响应

from flask import jsonify

@app.route('/data')
def get_data():
    data = {
        'name': 'John',
        'age': 30,
        'skills': ['Python', 'Flask']
    }
    return jsonify(data)  # 自动设置Content-Type为application/json

（四）重定向

from flask import redirect, url_for

@app.route('/old')
def old_endpoint():
    return redirect(url_for('new_endpoint'))  # 重定向到new_endpoint路由

@app.route('/new')
def new_endpoint():
    return "This is the new endpoint"

（五）错误处理

错误处理是Web应用开发中非常重要的一部分，良好的错误处理机制可以提升用户体验和应用的健壮性。在Flask中，错误处理主要通过abort()函数和errorhandler装饰器来实现。

from flask import abort

@app.route('/admin')
def admin_panel():
    if not user_is_admin():
        abort(403)  # 返回403 Forbidden错误
    return "Admin Panel"

常见HTTP状态码：

• 400 - Bad Request (客户端请求错误)
• 401 - Unauthorized (未认证)
• 403 - Forbidden (无权限)
• 404 - Not Found (资源不存在)
• 405 - Method Not Allowed (请求方法不允许)
• 500 - Internal Server Error (服务器内部错误)

使用@app.errorhandler装饰器可以为特定错误码或异常注册自定义处理函数。

# 自定义错误处理——处理HTTP错误码
@app.errorhandler(404)
def page_not_found(error):
    return "This page does not exist", 404

# 自定义错误处理——处理Python异常
@app.errorhandler(ZeroDivisionError)
def handle_zero_division(error):
    return "Oops! Divided by zero!", 500

错误处理器可以返回与普通视图函数相同的响应类型：

• 字符串
• 响应对象
• 元组：(response, status_code)
• 元组：(response, headers)
• 元组：(response, status_code, headers)

@app.errorhandler(404)
def not_found(error):
    # 返回JSON格式的错误信息
    return jsonify({'error': 'Not found', 'message': str(error)}), 404

三，钩子（Before/After Request）

Flask 提供了多种请求钩子（也称为装饰器），允许开发者在请求处理的不同阶段插入自定义逻辑。这些钩子对于执行预处理、后处理、请求验证、资源管理等任务非常有用。

（一）请求钩子概述

Flask 提供了四个主要的请求钩子：

• before_request - 在每个请求之前执行
• after_request - 在每个请求之后执行（如果没有未处理的异常）
• teardown_request - 在每个请求之后执行（即使有未处理的异常）
• before_first_request - 在第一个请求之前执行（Flask 2.3+ 已弃用）

（二）@app.before_request

功能：在每个请求处理之前执行

特点：

• 如果该函数返回非 None 值，Flask 将跳过常规的路由处理，直接使用该返回值
• 常用于身份验证、请求预处理、数据库连接等
• 可以注册多个函数，按注册顺序执行

@app.before_request
def load_user():
    # 从session中加载用户信息
    user_id = session.get('user_id')
    if user_id:
    	# 设置全局数据
        g.user = User.query.get(user_id)
    else:
        g.user = None

@app.before_request
def check_maintenance():
    # 检查是否在维护模式
    if app.config['MAINTENANCE_MODE']:
        return render_template('maintenance.html'), 503

（三）@app.after_request

功能：在每个请求处理之后执行（仅当没有未处理的异常时）

特点：

• 接收并必须返回一个响应对象
• 常用于添加HTTP头、修改响应内容、记录日志等
• 可以注册多个函数，按注册的相反顺序执行

@app.after_request
def add_security_headers(response):
    # 添加安全相关的HTTP头
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'SAMEORIGIN'
    response.headers['X-XSS-Protection'] = '1; mode=block'
    return response

@app.after_request
def log_request(response):
    # 记录请求信息
    app.logger.info(f"{request.method} {request.path} - {response.status_code}")
    return response

（四）@app.teardown_request

功能：在每个请求结束时执行（无论是否有异常）

特点：

• 接收一个异常参数（如果没有异常则为None）
• 不修改响应对象，主要用于清理工作
• 即使请求处理过程中抛出异常也会执行
• 常用于释放资源、关闭数据库连接等

@app.teardown_request
def close_db_connection(exception):
    # 关闭数据库连接
    if hasattr(g, 'db'):
        g.db.close()
    
    # 如果有异常可以记录
    if exception:
        app.logger.error(f"Request teardown with exception: {str(exception)}")

四，文件上传与下载

文件上传：

from flask import request, redirect, url_for
from werkzeug.utils import secure_filename
import os

UPLOAD_FOLDER = '/path/to/uploads'
ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg'}

app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER

def allowed_file(filename):
    return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        if 'file' not in request.files:
            return redirect(request.url)
        file = request.files['file']
        if file.filename == '':
            return redirect(request.url)
        if file and allowed_file(file.filename):
        	# secure_filename是Flask（实际上是其依赖的Werkzeug库）提供的一个实用函数，专门用于处理用户上传的文件名，确保文件名是安全且符合文件系统要求的
            filename = secure_filename(file.filename)
            file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
            return 'File uploaded successfully'
    return '''
    <form method="post" enctype="multipart/form-data">
      <input type="file" name="file">
      <input type="submit" value="Upload">
    </form>
    '''

文件下载：

from flask import send_from_directory

@app.route('/downloads/<filename>')
def download_file(filename):
    return send_from_directory(app.config['UPLOAD_FOLDER'], filename, as_attachment=True)

五，开发restful接口

（一）一些 restful 实践

1，使用名词表示资源

在REST架构中，资源是Web上任何可以被标识和操作的事物，它可以是一个文档、一张图片、一组数据、一项服务等任何可以命名的信息。

GET    /posts                 # 获取所有文章
POST   /posts                 # 创建新文章
GET    /posts/{post_id}       # 获取特定文章
PUT    /posts/{post_id}       # 更新文章
DELETE /posts/{post_id}       # 删除文章

GET    /posts/{post_id}/comments  # 获取文章评论
POST   /posts/{post_id}/comments  # 添加评论

GET    /categories            # 获取所有分类
GET    /categories/{cat_id}/posts  # 获取某分类下的文章

HTTP方法（GET、POST、PUT、DELETE等）已经表达了操作类型，URL只需标识操作的目标（资源）即可。
REST的核心约束之一，通过标准化HTTP方法（动词）来操作资源（名词），保持接口一致性。

2，使用合理的请求方法与状态码

HTTP请求方法：

HTTP状态码分类表：

成功场景：

GET /users/123
→ 200 OK
{ "id": 123, "name": "Alice" }

POST /users
→ 201 Created
Location: /users/124
{ "id": 124, "name": "Bob" }

DELETE /users/123
→ 204 No Content

错误场景：

GET /users/999
→ 404 Not Found
{ "error": "User not found" }

POST /users
Content-Type: application/json
{}
→ 400 Bad Request
{ "error": "Missing required field: name" }

PUT /users/123
→ 403 Forbidden
{ "error": "You can only update your own profile" }

3，正确使用 restful

RESTful架构的无状态性(Statelessness)是其核心约束之一——每个从客户端到服务器的请求都必须包含处理该请求所需的全部信息，服务器不应存储任何客户端上下文信息在请求之间。

协议层面——HTTP本身是无状态协议，RESTful API不应破坏这一特性：

• 无会话：不应依赖服务器端session
• 无上下文：请求不依赖先前请求的处理结果

资源操作层面——资源表示应包含所有必要状态信息：

• 操作不应隐含前置状态要求

认证授权层面——每个请求应携带完整认证信息：

• 不应依赖服务器存储的认证状态

 正确方式：每个请求自包含认证令牌
GET /orders HTTP/1.1
Authorization: Bearer xxxxxx

错误方式：依赖服务器会话
GET /orders HTTP/1.1
Cookie: JSESSIONID=xxxxx  # 依赖服务器存储的会话状态

✅ 正确方式：客户端提供完整分页参数
GET /products?page=2&size=20

❌ 错误方式：服务器记住分页状态
GET /products/next-page  # 依赖服务器记住当前页码

4，嵌套对象序列化

常用的 JSON API规范，将关联的信息内联在一起：

{
  "data": {
    "type": "articles",
    "id": "1",
    "relationships": {
      "author": {
        "data": { "type": "people", "id": "9" }
      }
    }
  },
  "included": [  // 分离的关联资源
    {
      "type": "people",
      "id": "9",
      "attributes": {
        "name": "王作者"
      }
    }
  ]
}

嵌套策略决策树
是否高频访问？

• 是 → 考虑嵌套
• 否 → 使用链接

是否关键业务数据？

• 是 → 嵌套核心字段
• 否 → 仅嵌套ID

是否可能单独更新？

• 是 → 减少嵌套
• 否 → 可适当嵌套

是否影响主要性能？

• 是 → 限制嵌套深度
• 否 → 可完整嵌套

（二）基础RESTful API开发

安装依赖：

pip install flask flask-restful flask-cors

配置跨域支持(CORS)：

# 方法一：使用 Flask-CORS 扩展（推荐）

from flask import Flask
from flask_cors import CORS

app = Flask(__name__)

# 1. 全局CORS配置 - 允许所有域名访问
CORS(app)

# 2. 全局CORS配置 - 指定允许的域名
# CORS(app, origins=['http://localhost:3000', 'https://mydomain.com'])

# 3. 全局CORS配置 - 详细配置
# CORS(app, 
#      origins=['http://localhost:3000'],
#      methods=['GET', 'POST', 'PUT', 'DELETE'],
#      allow_headers=['Content-Type', 'Authorization'],
#      supports_credentials=True)

@app.route('/api/data')
def get_data():
    return {'message': '数据获取成功'}

# 4. 针对特定路由的CORS配置
from flask_cors import cross_origin

@app.route('/api/special')
@cross_origin(origins=['http://localhost:3000'])
def special_endpoint():
    return {'message': '特定路由的跨域配置'}

if __name__ == '__main__':
    app.run(debug=True)


# 方法二：手动配置CORS响应头

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.after_request
def after_request(response):
    # 允许所有域名访问
    response.headers['Access-Control-Allow-Origin'] = '*'
    
    # 或者指定特定域名
    # response.headers['Access-Control-Allow-Origin'] = 'http://localhost:3000'
    
    # 允许的HTTP方法
    response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE, OPTIONS'
    
    # 允许的请求头
    response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization, X-Requested-With'
    
    # 是否允许发送cookies
    response.headers['Access-Control-Allow-Credentials'] = 'true'
    
    # 预检请求的缓存时间（秒）
    response.headers['Access-Control-Max-Age'] = '3600'
    
    return response

# 处理OPTIONS预检请求
@app.before_request
def handle_preflight():
    if request.method == "OPTIONS":
        response = jsonify()
        response.headers['Access-Control-Allow-Origin'] = '*'
        response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE, OPTIONS'
        response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization, X-Requested-With'
        return response

@app.route('/api/data')
def get_data():
    return jsonify({'message': '数据获取成功'})

if __name__ == '__main__':
    app.run(debug=True)


# 方法三：使用装饰器自定义CORS

from functools import wraps
from flask import Flask, jsonify, make_response

app = Flask(__name__)

def add_cors_headers(origin="*"):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            response = make_response(f(*args, **kwargs))
            response.headers['Access-Control-Allow-Origin'] = origin
            response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE, OPTIONS'
            response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization'
            return response
        return decorated_function
    return decorator

@app.route('/api/data')
@add_cors_headers(origin='http://localhost:3000')
def get_data():
    return jsonify({'message': '使用装饰器的跨域配置'})

if __name__ == '__main__':
    app.run(debug=True)


# 安装 Flask-CORS 扩展的命令：
# pip install flask-cors

基础API结构：

from flask import Flask, request, jsonify
from flask_restful import Api, Resource
from flask_cors import CORS

app = Flask(__name__)
CORS(app)  # 启用CORS
api = Api(app)

# 模拟数据库
users = [
    {"id": 1, "name": "Alice", "email": "alice@example.com"},
    {"id": 2, "name": "Bob", "email": "bob@example.com"}
]

class UserListResource(Resource):
    def get(self):
        return jsonify({"users": users})
    
    def post(self):
        if not request.is_json:
            return {"message": "Missing JSON in request"}, 400
        
        data = request.get_json()
        if not all(k in data for k in ["name", "email"]):
            return {"message": "Missing required fields"}, 400
        
        new_user = {
            "id": len(users) + 1,
            "name": data["name"],
            "email": data["email"]
        }
        users.append(new_user)
        return jsonify(new_user), 201

class UserResource(Resource):
    def get(self, user_id):
        user = next((u for u in users if u["id"] == user_id), None)
        if user:
            return jsonify(user)
        return {"message": "User not found"}, 404
    
    def put(self, user_id):
        user = next((u for u in users if u["id"] == user_id), None)
        if not user:
            return {"message": "User not found"}, 404
        
        data = request.get_json()
        if not data:
            return {"message": "No data provided"}, 400
        
        user.update({
            "name": data.get("name", user["name"]),
            "email": data.get("email", user["email"])
        })
        return jsonify(user)
    
    def delete(self, user_id):
        global users
        users = [u for u in users if u["id"] != user_id]
        return {"message": "User deleted"}, 204

api.add_resource(UserListResource, '/api/users')
api.add_resource(UserResource, '/api/users/<int:user_id>')

if __name__ == '__main__':
    app.run(debug=True)

使用Flask-RESTX（Flask-RESTful的增强版），提供Swagger文档支持：

from flask import Flask, request
from flask_restx import Api, Resource, fields

# 初始化Flask应用
app = Flask(__name__)
api = Api(app, doc='/', version='1.0', title='Todo API', description='一个简单的待办事项API')

# 模拟数据存储
todos = [
    {"id": 1, "task": "学习Flask-RESTX", "completed": False},
    {"id": 2, "task": "编写API文档", "completed": False}
]

# 定义数据模型
todo_model = api.model('Todo', {
    'id': fields.Integer(readonly=True),
    'task': fields.String(required=True),
    'completed': fields.Boolean
})

# 获取所有待办事项
@api.route('/todos')
class TodoList(Resource):
    @api.doc('list_todos')
    @api.marshal_list_with(todo_model)
    def get(self):
        """获取所有待办事项"""
        return todos

    @api.doc('create_todo')
    @api.expect(todo_model)
    @api.marshal_with(todo_model, code=201)
    def post(self):
        """创建新待办事项"""
        new_todo = api.payload
        new_todo['id'] = len(todos) + 1
        todos.append(new_todo)
        return new_todo, 201

# 操作单个待办事项
@api.route('/todos/<int:todo_id>')
@api.param('todo_id', '待办事项ID')
@api.response(404, '待办事项未找到')
class Todo(Resource):
    @api.doc('get_todo')
    @api.marshal_with(todo_model)
    def get(self, todo_id):
        """获取单个待办事项"""
        for todo in todos:
            if todo['id'] == todo_id:
                return todo
        api.abort(404)

    @api.doc('update_todo')
    @api.expect(todo_model)
    @api.marshal_with(todo_model)
    def put(self, todo_id):
        """更新待办事项"""
        for todo in todos:
            if todo['id'] == todo_id:
                todo.update(api.payload)
                return todo
        api.abort(404)

    @api.doc('delete_todo')
    @api.response(204, '待办事项已删除')
    def delete(self, todo_id):
        """删除待办事项"""
        for i, todo in enumerate(todos):
            if todo['id'] == todo_id:
                del todos[i]
                return '', 204
        api.abort(404)

if __name__ == '__main__':
    app.run(debug=True)

访问 http://127.0.0.1:5000