Skip to content

middleware.py

Middleware to handle requests from smee server.

SmeeMiddleware

Bases: BaseHTTPMiddleware

Middleware to process requests from smee server before reaching the API endpoints.

Source code in taglyatelle/exposition/middleware.py
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class SmeeMiddleware(BaseHTTPMiddleware):
    """Middleware to process requests from smee server before reaching the API endpoints."""

    @tracing_request(enabled=USE_TRACING_REQUEST)
    async def dispatch(self, request: Request, call_next: Callable) -> Request | JSONResponse:
        """
        Process incoming requests to verify signatures and log details.

        Parameters
        ----------
        request
            The incoming request object

        call_next
            The next middleware or endpoint to call

        Returns
        -------
        The response from the next middleware or endpoint
        """
        payload_body = await request.body()

        webhook_sender = WebhookSender(request)
        webhook_sender.verify_signature(payload_body, WEBHOOK_SECRET)
        logger.info("Signature verified.")

        response = await call_next(request)

        return response

dispatch(request, call_next) async

Process incoming requests to verify signatures and log details.

Parameters

request The incoming request object

call_next The next middleware or endpoint to call

Returns

The response from the next middleware or endpoint

Source code in taglyatelle/exposition/middleware.py
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@tracing_request(enabled=USE_TRACING_REQUEST)
async def dispatch(self, request: Request, call_next: Callable) -> Request | JSONResponse:
    """
    Process incoming requests to verify signatures and log details.

    Parameters
    ----------
    request
        The incoming request object

    call_next
        The next middleware or endpoint to call

    Returns
    -------
    The response from the next middleware or endpoint
    """
    payload_body = await request.body()

    webhook_sender = WebhookSender(request)
    webhook_sender.verify_signature(payload_body, WEBHOOK_SECRET)
    logger.info("Signature verified.")

    response = await call_next(request)

    return response