当前位置： 首页 > article >正文

[Python3] Sanic中间件

article 2024/12/26 14:40:48

在 Sanic 中，中间件（middleware）是指在请求和响应之间执行的代码。它们是一个非常强大的工具，用于处理请求的预处理、响应的后处理、全局错误处理、日志记录、认证、权限校验、跨域资源共享（CORS）等任务。中间件通常用于应用程序的全局逻辑，在请求处理前或响应处理后执行。

在这里插入图片描述

Sanic 中间件的工作流程

Sanic 中间件会在以下两个阶段执行：

  1. 请求处理前：处理请求数据，在请求被路由处理之前执行。可以用于校验、修改请求等操作。
  2. 响应处理后：处理响应数据，在响应被发送给客户端之前执行。可以用于修改响应、记录日志等操作。

中间件的使用

在 Sanic 中，中间件通过装饰器来定义。你可以为整个应用程序（全局中间件）定义中间件，也可以为某些特定的路由定义中间件。

1. 全局中间件

全局中间件是在整个应用程序的生命周期中被调用的，不依赖于特定的路由。它们在请求和响应的各个阶段都会执行。

请求阶段的全局中间件

全局中间件通常用于处理请求的预处理，如身份验证、记录日志、设置 CORS 等。

from sanic import Sanic
from sanic.response import json

app = Sanic("MyApp")

# 请求阶段的全局中间件
@app.middleware("request")
async def log_request(request):
    print(f"Request received: {request.method} {request.url}")
    # 你可以在这里对请求进行修改，例如身份验证、设置 CORS 等
    # 如果请求需要认证，可以在这里进行验证

# 响应阶段的全局中间件
@app.middleware("response")
async def log_response(request, response):
    print(f"Response status: {response.status}")
    # 你可以在这里修改响应，例如添加 CORS 头部、记录日志等
    # 在响应返回给客户端之前处理
    response.headers['X-Custom-Header'] = 'Sanic'  # 可以修改响应头
    return response

@app.route("/")
async def hello(request):
    return json({"message": "Hello, Sanic!"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

在这个例子中：

  • log_request 中间件会在每个请求到达应用时被调用，打印请求的方法和 URL。
  • log_response 中间件会在每个响应返回给客户端之前被调用，打印响应的状态码，并在响应头中添加一个自定义的 header。

2. 路由中间件

你还可以为特定的路由或路由组定义中间件。这是 Sanic 提供的灵活性，可以让你为某些路由添加额外的逻辑，而不影响其他路由。

@app.middleware("request", route="/user/<user_id>")
async def check_user_auth(request, user_id):
    # 假设我们通过 header 进行身份验证
    if request.headers.get("Authorization") != "Bearer my_token":
        return json({"error": "Unauthorized"}, status=401)
    print(f"Authenticated request for user {user_id}")

在这个示例中，check_user_auth 中间件只会在访问 /user/<user_id> 路由时执行，检查请求头中的 Authorization 字段是否包含有效的 Token。

3. 异常处理中间件

Sanic 中的异常处理中间件允许你捕获应用程序中的未处理异常，并为用户提供定制化的错误信息。

你可以使用 @app.exception 装饰器捕获特定类型的异常，也可以使用中间件全局处理错误。

捕获特定异常
from sanic.exceptions import SanicException

@app.exception(SanicException)
async def handle_sanic_exception(request, exception):
    return json({"error": f"Sanic error: {exception}"}, status=500)
捕获所有异常

你也可以定义一个捕获所有异常的全局中间件，并返回自定义的错误信息。

@app.middleware("response")
async def handle_all_exceptions(request, response):
    if hasattr(request, 'exception') and request.exception:
        return json({"error": f"Unhandled error: {request.exception}"}, status=500)
    return response

在这个例子中，如果某个请求抛出了异常，并且没有被其他中间件捕获，handle_all_exceptions 会捕获到并返回一个自定义的错误响应。

4. 异步中间件

由于 Sanic 是基于 asyncio 的异步框架，你可以编写异步的中间件，这样它们不会阻塞事件循环。中间件是异步的，这意味着你可以执行 I/O 操作（例如数据库查询或 HTTP 请求）而不会阻塞其他请求的处理。

@app.middleware("request")
async def async_middleware(request):
    # 异步操作，例如异步查询数据库
    await asyncio.sleep(1)
    print("Async operation completed")

5. 优先级和中间件顺序

多个中间件会按照定义的顺序依次执行。在 Sanic 中，中间件的优先级是固定的，按照以下顺序执行：

  • 请求中间件：按定义的顺序执行，最先定义的请求中间件最先执行。
  • 响应中间件：按定义的顺序执行，最先定义的响应中间件最先执行。

6. 中间件的运行机制

请求阶段

请求阶段中间件是指那些在请求路由匹配之前执行的中间件，通常用于请求预处理，例如请求数据解析、验证、认证等。

响应阶段

响应阶段的中间件是在响应发送回客户端之前执行的，通常用于对响应进行修改，例如添加额外的响应头、修改响应数据等。

7. 使用中间件进行身份验证

在 Web 开发中，身份验证是非常常见的需求。可以使用中间件来实现身份验证逻辑。比如，检查用户请求头中的 Token 或 Cookies 是否有效。

@app.middleware("request")
async def check_auth(request):
    token = request.headers.get("Authorization")
    if token != "Bearer valid_token":
        return json({"error": "Unauthorized"}, status=401)

8. 使用中间件实现 CORS

跨域资源共享（CORS）是一种允许服务器控制哪些域可以访问其资源的机制。你可以使用中间件来添加 CORS 头部。

@app.middleware("response")
async def add_cors_headers(request, response):
    response.headers["Access-Control-Allow-Origin"] = "*"
    response.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS"
    response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
    return response

9. 中间件的多个装饰器

Sanic 支持多个中间件装饰器应用在同一个请求或响应阶段上。例如，你可以同时应用多个请求中间件。

@app.middleware("request")
async def middleware_one(request):
    print("Middleware one executed")

@app.middleware("request")
async def middleware_two(request):
    print("Middleware two executed")

在这个例子中，middleware_one 会先执行，然后是 middleware_two。

总结

Sanic 中的中间件是非常强大和灵活的，它们使得你能够在 Web 应用的请求处理流程中插入自定义逻辑。中间件的使用场景非常广泛，包括请求和响应处理、身份验证、权限控制、日志记录、CORS 处理、错误处理等。

主要特点：
  • 请求中间件：在请求路由处理之前执行。
  • 响应中间件：在响应返回给客户端之前执行。
  • 全局中间件：适用于整个应用。
  • 路由中间件：只适用于特定的路由。
  • 异步操作：支持异步中间件，可以执行 I/O 操作而不会阻塞事件循环。

通过合理使用中间件，你可以在 Sanic 应用中灵活地处理各种需求，提高代码的可维护性和扩展性。

统计每个请求耗时

要统计一个请求的执行时间，你可以利用 Sanic 中间件来实现。在请求处理的生命周期中，我们可以在请求开始时记录时间戳，在请求结束时计算并输出请求的执行时间。

1. 使用中间件统计请求执行时间

我们可以使用 请求中间件 来记录每个请求开始时的时间戳，然后使用 响应中间件 来计算并输出请求的执行时间。

示例代码
from sanic import Sanic
from sanic.response import json
import time

app = Sanic("TimingApp")

# 请求中间件：记录请求开始的时间
@app.middleware("request")
async def record_start_time(request):
    request.ctx.start_time = time.time()  # 将开始时间存储到请求上下文中

# 响应中间件：计算并输出请求执行时间
@app.middleware("response")
async def calculate_execution_time(request, response):
    if hasattr(request.ctx, 'start_time'):
        end_time = time.time()
        execution_time = end_time - request.ctx.start_time  # 计算执行时间
        print(f"Request to {request.url} took {execution_time:.4f} seconds")
    return response

# 示例路由
@app.route("/")
async def hello(request):
    return json({"message": "Hello, Sanic!"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

2. 代码解析

  1. record_start_time 请求中间件：

    • 在请求到达应用时触发，使用 time.time() 获取当前时间并将其存储在 request.ctx.start_time 中。ctx 是 Sanic 提供的上下文对象，用来在请求的生命周期内存储数据。

  2. calculate_execution_time 响应中间件：

    • 在请求完成后触发，计算请求执行的时间。通过 time.time() 获取当前时间，并与 request.ctx.start_time 进行比较，得到请求的执行时间。
    • 输出请求的执行时间，单位是秒。

  3. 示例路由：

    • 在 / 路由中，我们返回一个简单的 JSON 响应，测试请求执行时间的统计。

3. 测试

当你运行上面的代码并访问 http://localhost:8000/ 时，终端输出类似以下内容：

Request to / took 0.0023 seconds

这表示请求到 / 的执行时间是 0.0023 秒。

4. 更复杂的统计方法

如果你需要更精细的控制（例如，统计多个请求的平均执行时间、最慢的请求、最短的请求等），你可以进一步扩展这个功能，记录统计数据到日志文件或者数据库中。

例如，你可以将执行时间记录到一个全局的统计数据中：

import time
from sanic import Sanic
from sanic.response import json
from collections import defaultdict

app = Sanic("TimingApp")
request_stats = defaultdict(list)  # 用于存储每个路由的执行时间

# 请求中间件：记录请求开始的时间
@app.middleware("request")
async def record_start_time(request):
    request.ctx.start_time = time.time()

# 响应中间件：计算请求执行时间并保存
@app.middleware("response")
async def calculate_execution_time(request, response):
    if hasattr(request.ctx, 'start_time'):
        end_time = time.time()
        execution_time = end_time - request.ctx.start_time
        print(f"Request to {request.url} took {execution_time:.4f} seconds")
        request_stats[request.url].append(execution_time)  # 记录执行时间

    return response

# 获取平均执行时间的路由
@app.route("/stats")
async def stats(request):
    stats = {url: sum(times) / len(times) for url, times in request_stats.items()}
    return json(stats)

# 示例路由
@app.route("/")
async def hello(request):
    return json({"message": "Hello, Sanic!"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

5. 扩展功能的解释

  • 记录请求执行时间：我们使用一个 defaultdict(list) 来记录每个 URL 请求的执行时间。在每次请求完成后，我们将执行时间添加到对应 URL 的列表中。
  • 统计平均执行时间：我们创建了一个 /stats 路由，用于返回每个路由的平均执行时间。

6. 总结

  • 使用 Sanic 中的 中间件 来统计请求的执行时间是非常简便的。通过记录请求开始和结束的时间，可以轻松计算出请求的执行时间。
  • 你可以根据自己的需求扩展这个功能，如统计最慢的请求、记录日志等。

这种方法能够帮助你监控应用的性能，尤其是在高并发的情况下，快速识别出潜在的性能瓶颈。


http://www.kler.cn/a/452267.html

相关文章：

  • NS3学习——tcpVegas算法代码详解（2）
  • C++模板：编译时模拟Duck Typing
  • (echarts)数据地图散点类型根据条件设置不同的标记图片
  • 攻防世界 cookie
  • 查询 MySQL 默认的存储引擎（SELECT @@default_storage_engine；）
  • 帧缓存的分配
  • 你比AI更有价值吗？
  • 微信小程序-基于Vant Weapp UI 组件库的Area 省市区选择
  • ESLint （10）
  • 低空经济的地理信息支撑：构建安全、高效的飞行管理体系
  • CSS|15 CSS3结构伪类border-collapse伪元素
  • PHP 微信棋牌开发全解析：高级教程
  • 解决Ascend上vllm运行时出现urllib3.exceptions.SSLError: [SSL: CERTIFICATE_VERIFY_FAILED]
  • redis和mysql的区别
  • 威尔克斯(Wilks)分布
  • 基于Pycharm与数据库的新闻管理系统（3）MongoDB
  • shell拓展知识
  • React State(状态)
  • Qt实现Android的图案密码（图形解锁）源码分享
  • 计算机的错误计算（一百九十一）
  • 信息抽取（NLP)是什么技术有哪些应用？
  • 大数据-256 离线数仓 - Atlas 数据仓库元数据管理 正式安装 启动服务访问 Hive血缘关系导入
  • 《Web 搜索引擎优化》
  • CI/CD在前端项目的应用：实现自动化与持续交付
  • Go入门篇：（二）基础知识之结构，包，变量初探
  • my-sql编写技巧