The use of Sanic middleware in Python3

existSanicIn , middleware refers to the code executed between a request and a response. They are a very powerful tool for handling tasks such as preprocessing of requests, post-processing of responses, global error handling, logging, authentication, permission verification, cross-domain resource sharing (CORS) and other tasks. Middleware is usually used for the global logic of an application and is executed before or after the request is processed.

Workflow of Sanic Middleware

The Sanic middleware is executed in the following two stages:

Before the request is processed: Process the request data and execute before the request is routed. It can be used for verification, modification of requests and other operations.
After the response process: Process the response data and execute before the response is sent to the client. It can be used to modify responses, record logs and other operations.

Use of middleware

existSanicIn , the middleware is defined by a decorator. You can define middleware for the entire application (global middleware) or for some specific routes.

1. Global middleware

Global middleware is called throughout the life cycle of the application and does not depend on specific routes. They are executed at all stages of the request and response.

Global middleware for the request phase

Global middleware is usually used to handle preprocessing of requests, such as authentication, logging, setting CORS, etc.

from sanic import Sanic
from  import json

app = Sanic("MyApp")

# Global middleware for the request phase@("request")
async def log_request(request):
    print(f"Request received: {} {}")
    # You can modify the request here, such as authentication, setting CORS, etc.    # If the request requires authentication, you can verify it here
# Global middleware for the response phase@("response")
async def log_response(request, response):
    print(f"Response status: {}")
    # You can modify the response here, such as adding CORS headers, logging, etc.    # Process before the response is returned to the client    ['X-Custom-Header'] = 'Sanic'  # Can modify the response header    return response

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

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

In this example:

log_requestThe middleware will be called when each request arrives at the application, printing the requested method and URL.
log_responseThe middleware will be called before each response is returned to the client, print the status code of the response, and add a custom header to the response header.

2. Routing middleware

You can also define middleware for specific routes or routing groups. This is the flexibility provided by Sanic that allows you to add extra logic to certain routes without affecting others.

@("request", route="/user/&lt;user_id&gt;")
async def check_user_auth(request, user_id):
    # Suppose we authenticate through header    if ("Authorization") != "Bearer my_token":
        return json({"error": "Unauthorized"}, status=401)
    print(f"Authenticated request for user {user_id}")

In this example,check_user_authMiddleware will only be accessed/user/<user_id>Execute during routing, check the request headerAuthorizationWhether the field contains a valid token.

3. Exception handling middleware

SanicThe exception handling middleware in allows you to catch unhandled exceptions in the application and provide users with customized error information.

You can use@The decorator catches exceptions of a specific type, and can also use middleware to handle errors globally.

Catch specific exceptions

from  import SanicException

@(SanicException)
async def handle_sanic_exception(request, exception):
    return json({"error": f"Sanic error: {exception}"}, status=500)

Catch all exceptions

You can also define a global middleware that catches all exceptions and returns custom error messages.

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

In this example, if an exception is thrown by a request and is not caught by other middleware,handle_all_exceptionsA custom error response will be captured and returned.

4. Asynchronous middleware

becauseSanicIt is based onasyncioAsynchronous framework, you can write asynchronous middleware so that they do not block the event loop. Middleware is asynchronous, which means you can perform I/O operations (such as database queries or HTTP requests) without blocking other requests.

@("request")
async def async_middleware(request):
    # Asynchronous operations, such as querying databases asynchronously    await (1)
    print("Async operation completed")

5. Priority and middleware order

Multiple middleware will be executed in the defined order. existSanicIn , the priority of the middleware is fixed and executed in the following order:

Request middleware: Execute in the defined order, and the first defined request middleware is executed first.
Response middleware: executes in the defined order, and the first defined response middleware is executed first.

6. Middleware running mechanism

Request phase

Request phase middleware refers to those middleware executed before the request route matches, which are usually used for request preprocessing, such as requesting data parsing, verification, authentication, etc.

Response phase

The middleware of the response phase is executed before the response is sent back to the client and is usually used to modify the response, such as adding additional response headers, modifying the response data, etc.

7. Use middleware for authentication

In web development, authentication is a very common requirement. Middleware can be used to implement authentication logic. For example, check whether the tokens or cookies in the user request header are valid.

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

8. Implement CORS using middleware

Cross-domain resource sharing (CORS) is a mechanism that allows servers to control which domains can access their resources. You can use middleware to add CORS headers.

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

9. Multiple decorators for middleware

Sanic supports multiple middleware decorator applications on the same request or response phase. For example, you can apply multiple request middleware at the same time.

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

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

In this example,middleware_oneWill execute first, thenmiddleware_two。

Summarize

SanicThe middleware in it is very powerful and flexible, and they allow you to insert custom logic into the request processing flow of your web application. Middleware usage scenarios are very wide, including request and response processing, authentication, permission control, logging, CORS processing, error processing, etc.

Main features:

Request Middleware: Execute before requesting routing processing.
Response Middleware: Execute before the response is returned to the client.
Global middleware: Suitable for the entire application.
Routing middleware: Applicable to specific routes only.
Asynchronous operation: Supports asynchronous middleware, which can perform I/O operations without blocking event loops.

By using middleware rationally, you canSanicFlexible handling of various needs in the application to improve the maintainability and scalability of the code.

Statistics the time spent per request

To count the execution time of a request, you can useSanicMiddleware to implement it. During the life cycle of request processing, we can record the timestamp at the beginning of the request, calculate and output the execution time of the request at the end of the request.

1. Use middleware to count request execution time

We can useRequest MiddlewareTo record the timestamp at the beginning of each request, and then useResponse MiddlewareTo calculate and output the execution time of the request.

Sample code

from sanic import Sanic
from  import json
import time

app = Sanic("TimingApp")

# Request Middleware: Record the time when the request starts@("request")
async def record_start_time(request):
    .start_time = ()  # Store the start time into the request context
# Response middleware: calculate and output the request execution time@("response")
async def calculate_execution_time(request, response):
    if hasattr(, 'start_time'):
        end_time = ()
        execution_time = end_time - .start_time  # Calculate execution time        print(f"Request to {} took {execution_time:.4f} seconds")
    return response

# Sample routing@("/")
async def hello(request):
    return json({"message": "Hello, Sanic!"})

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

2. Code parsing

record_start_timeRequest Middleware：
- Triggered when the request reaches the application, use()Get the current time and store it in.start_timemiddle.ctxyesSanicThe provided context object is used to store data during the life of the request.
calculate_execution_timeResponse middleware：
- Triggered after the request is completed, and the time for the request to be executed is calculated. pass()Get the current time and with.start_timeCompare to get the execution time of the request.
- The execution time of the output request is in seconds.
Sample routing：
- exist/In the routing, we return a simple JSON response to test the statistics of the execution time of the request.

3. Test

When you run the above code and accesshttp://localhost:8000/When the terminal outputs something similar to the following:

Request to / took 0.0023 seconds

This means that the request has been reached/The execution time is0.0023Second.

4. More complex statistical methods

If you need more granular control (for example, counting the average execution time of multiple requests, the slowest request, the shortest request, etc.), you can further expand this function to record statistics into a log file or database.

For example, you can log execution time into a global statistics:

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

app = Sanic("TimingApp")
request_stats = defaultdict(list)  # Used to store execution time for each route
# Request Middleware: Record the time when the request starts@("request")
async def record_start_time(request):
    .start_time = ()

# Response middleware: calculate the request execution time and save it@("response")
async def calculate_execution_time(request, response):
    if hasattr(, 'start_time'):
        end_time = ()
        execution_time = end_time - .start_time
        print(f"Request to {} took {execution_time:.4f} seconds")
        request_stats[].append(execution_time)  # Record execution time
    return response

# Get the route for average execution time@("/stats")
async def stats(request):
    stats = {url: sum(times) / len(times) for url, times in request_stats.items()}
    return json(stats)

# Sample routing@("/")
async def hello(request):
    return json({"message": "Hello, Sanic!"})

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

5. Explanation of extended functions

Record the request execution time: We use onedefaultdict(list)To record the execution time of each URL request. After each request is completed, we add the execution time to the list of the corresponding URLs.
Statistics on average execution time: We created one/statsRouting, used to return the average execution time of each route.

6. Summary

useSanicInmiddlewareIt is very easy to count the execution time of the request. By recording the time of the start and end of the request, the execution time of the request can be easily calculated.
You can expand this function according to your needs, such as counting the slowest requests, recording logs, etc.

This approach can help you monitor the performance of your application, especially in high concurrency, and quickly identify potential performance bottlenecks.

This is the end of this article about the use of Sanic middleware in Python3. For more related content of Python3 Sanic middleware, please search for my previous articles or continue browsing the related articles below. I hope everyone will support me in the future!