Python3中Sanic中间件的使用

2025-01-09 04:50
文章标签 使用 python3 中间件 sanic

本文主要是介绍Python3中Sanic中间件的使用，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

《Python3中Sanic中间件的使用》Sanic框架中的中间件是一种强大的工具,本文就来介绍Python3中Sanic中间件的使用,具有一定的参考价值,感兴趣的可以了解一下...

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

Sanic 中间件的工作流程

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

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

中间件的使用

在 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 Sanhttp://www.chinasem.cnicException

@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 会捕获到并返回一个自定义的错误响应。

China编程

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/Ojavascript 操作而不会阻塞事件循环。

通过合理使用中间件，你可以在 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. 代码解析

  • record_start_time 请求中间件：

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

  • calculate_execution_time 响应中间件：

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

  • 示例路由：

    • 在 / 路由中，我们返回一个简单的 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")
jsasync 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 中的 中间件 来统计请求的执行时间是非常简便的。通过记录请求开始和结束的时间，可以轻松计算出请求的执行时间。
  • 你可以根据自己的需求扩展这个功能，如统计最慢的请求、记录日志等。

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

到此这篇关于python3中Sanic中间件的使用的文章就介绍到这了,更多相关Python3 Sanic中间件内容请搜索编程China编程(www.chinasem.cn)以前的文章或继续浏览下面的相关文章希望大家以后多多支持China编程(www.chinasem.cn)！

这篇关于Python3中Sanic中间件的使用的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1152987。

相关文章

Qt spdlog日志模块的使用详解

Qt spdlog日志模块的使用详解

《Qtspdlog日志模块的使用详解》在Qt应用程序开发中,良好的日志系统至关重要,本文将介绍如何使用spdlog1.5.0创建满足以下要求的日志系统,感兴趣的朋友一起看看吧... 目录版本摘要例子logmanager.cpp文件main.cpp文件版本spdlog版本：1.5.0采用1.5.0版本主要
阅读更多...
Java中使用Hutool进行AES加密解密的方法举例

Java中使用Hutool进行AES加密解密的方法举例

《Java中使用Hutool进行AES加密解密的方法举例》AES是一种对称加密,所谓对称加密就是加密与解密使用的秘钥是一个,下面：本文主要介绍Java中使用Hutool进行AES加密解密的相关资料... 目录前言一、Hutool简介与引入1.1 Hutool简介1.2 引入Hutool二、AES加密解密基础
阅读更多...
使用Python将JSON,XML和YAML数据写入Excel文件

使用Python将JSON,XML和YAML数据写入Excel文件

《使用Python将JSON,XML和YAML数据写入Excel文件》JSON、XML和YAML作为主流结构化数据格式,因其层次化表达能力和跨平台兼容性,已成为系统间数据交换的通用载体,本文将介绍如何... 目录如何使用python写入数据到Excel工作表用Python导入jsON数据到Excel工作表用
阅读更多...
鸿蒙中@State的原理使用详解(HarmonyOS 5)

鸿蒙中@State的原理使用详解(HarmonyOS 5)

《鸿蒙中@State的原理使用详解(HarmonyOS5)》@State是HarmonyOSArkTS框架中用于管理组件状态的核心装饰器,其核心作用是实现数据驱动UI的响应式编程模式,本文给大家介绍... 目录一、@State在鸿蒙中是做什么的？二、@Spythontate的基本原理1. 依赖关系的收集2.
阅读更多...
Python基础语法中defaultdict的使用小结

Python基础语法中defaultdict的使用小结

《Python基础语法中defaultdict的使用小结》Python的defaultdict是collections模块中提供的一种特殊的字典类型,它与普通的字典（dict）有着相似的功能,本文主要... 目录示例1示例2python的defaultdict是collections模块中提供的一种特殊的字
阅读更多...
C++ Sort函数使用场景分析

C++ Sort函数使用场景分析

《C++Sort函数使用场景分析》sort函数是algorithm库下的一个函数,sort函数是不稳定的,即大小相同的元素在排序后相对顺序可能发生改变,如果某些场景需要保持相同元素间的相对顺序,可使... 目录C++ Sort函数详解一、sort函数调用的两种方式二、sort函数使用场景三、sort函数排序
阅读更多...
Java String字符串的常用使用方法

Java String字符串的常用使用方法

《JavaString字符串的常用使用方法》String是JDK提供的一个类,是引用类型,并不是基本的数据类型,String用于字符串操作,在之前学习c语言的时候,对于一些字符串,会初始化字符数组表... 目录一、什么是String二、如何定义一个String1. 用双引号定义2. 通过构造函数定义三、St
阅读更多...
Pydantic中Optional 和Union类型的使用

Pydantic中Optional 和Union类型的使用

《Pydantic中Optional和Union类型的使用》本文主要介绍了Pydantic中Optional和Union类型的使用,这两者在处理可选字段和多类型字段时尤为重要,文中通过示例代码介绍的... 目录简介Optional 类型Union 类型Optional 和 Union 的组合总结简介Pyd
阅读更多...
Vue3使用router,params传参为空问题

Vue3使用router,params传参为空问题

《Vue3使用router,params传参为空问题》：本文主要介绍Vue3使用router,params传参为空问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐... 目录vue3使用China编程router,params传参为空1.使用query方式传参2.使用 Histo
阅读更多...
使用Python自建轻量级的HTTP调试工具

使用Python自建轻量级的HTTP调试工具

《使用Python自建轻量级的HTTP调试工具》这篇文章主要为大家详细介绍了如何使用Python自建一个轻量级的HTTP调试工具,文中的示例代码讲解详细,感兴趣的小伙伴可以参考一下... 目录一、为什么需要自建工具二、核心功能设计三、技术选型四、分步实现五、进阶优化技巧六、使用示例七、性能对比八、扩展方向建
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2