Discussed on falconry/user: how to implement a basic instrumentation plugin using middleware

instrumentation.py

import random
import time
import uuid

import falcon


class RequestMonitor:

    def __init__(self, req, resp):
        self._start = time.time()

    def send_metrics(self, req, resp, req_succeeded):
        taken = time.time() - self._start
        print(f'time taken for {req.context.requestid}: {taken}')
        print(f'{req.context.requestid} succeeded: {req_succeeded}')


class InstrumentationMiddleware:

    def process_request(self, req, resp):
        req.context.requestid = str(uuid.uuid4())
        req.context.monitor = RequestMonitor(req, resp)

    def process_response(self, req, resp, resource, req_succeeded):
        req.context.monitor.send_metrics(req, resp, req_succeeded)

        # NOTE(vytas): The above does not take resp.stream into account.
        #   In the case resp.stream is used, one would need to hook into it
        #   explicitly, see e.g.
        # https://github.com/vytas7/falcon-sqla/blob/master/falcon_sqla/util.py


class InstrumentedApp(falcon.App):

    def __init__(self, *args, **kwargs):
        # TODO: This won't work if someone passes `middleware` as a positional
        # argument.
        middleware = kwargs.pop('middleware', [])

        # TODO: This doesn't support legacy shim for passing a single
        # middleware object instead of list.
        middleware = list(middleware)
        middleware.insert(0, InstrumentationMiddleware())
        kwargs['middleware'] = middleware

        super().__init__(*args, **kwargs)


class HelloWorld:

    def on_get(self, req, resp):
        resp.media = {'message': 'Hello, World!'}


class Sleepy:
    def on_get(self, req, resp):
        value = random.random()
        time.sleep(value)

        resp.media = value


app = InstrumentedApp()
app.add_route('/hello', HelloWorld())
app.add_route('/sleepy', Sleepy())