An aiohttp middleware dumps HTTP I/O

aiohttp_dump_middleware.py

import json
import logging
import string
from itertools import zip_longest
from json.decoder import JSONDecodeError
from textwrap import dedent
from typing import (
    Awaitable,
    Callable,
    Iterable,
    Optional,
)

from aiohttp.web import middleware, Response, Request
from multidict import MultiMapping


logger = logging.getLogger(__name__)


def _dump_headers(headers: MultiMapping) -> str:
    return '\n'.join(f'{k}: {v}' for k, v in headers.items())


def _hexdump(bs: bytes) -> str:
    """A pure python `hexdump -C` mimic
    >>> print(hexdump(string.printable.encode('ascii')))
    000000000 30 31 32 33 34 35 36 37 38 39 61 62 63 64 65 66 |0123456789abcdef|
    000000010 67 68 69 6a 6b 6c 6d 6e 6f 70 71 72 73 74 75 76 |ghijklmnopqrstuv|
    000000020 77 78 79 7a 41 42 43 44 45 46 47 48 49 4a 4b 4c |wxyzABCDEFGHIJKL|
    000000030 4d 4e 4f 50 51 52 53 54 55 56 57 58 59 5a 21 22 |MNOPQRSTUVWXYZ!"|
    000000040 23 24 25 26 27 28 29 2a 2b 2c 2d 2e 2f 3a 3b 3c |#$%&'()*+,-./:;<|
    000000050 3d 3e 3f 40 5b 5c 5d 5e 5f 60 7b 7c 7d 7e 20 09 |=>?@[\]^_`{|}~..|
    000000060 0a 0d 0b 0c                                     |....            |
    """
    ascii = (string.digits + string.ascii_letters + string.punctuation).encode('ascii')

    def _int_to_ascii(i: int, default: str = '.') -> str:
        return chr(i) if i in ascii else default

    def _dump_line(idx: int, chunk: Iterable[Optional[bytes]]) -> str:
        idxpart = f'{idx:08x}0'
        hexpart = ' '.join('  ' if i is None else f'{i:02x}' for i in chunk)
        asciipart = ''.join(' ' if i is None else _int_to_ascii(i) for i in chunk)
        return f'{idxpart} {hexpart} |{asciipart}|'

    return '\n'.join(
        _dump_line(i, chunk)
        for i, chunk in enumerate(zip_longest(*([iter(bs)] * 16)))
    )


def _dump_body(body: bytes) -> str:
    try:
        text = body.decode('utf-8')
    except UnicodeDecodeError:
        return _hexdump(body)

    try:
        return json.dumps(
            json.loads(text),
            ensure_ascii=False,
            indent=2,
        )
    except JSONDecodeError:
        return text


@middleware
async def dump_middleware(
    request: Request,
    handler: Callable[[Request], Awaitable[Response]],
):
    logger.debug(
        dedent('''\
            REQUEST:
            %s %s HTTP/%d.%d
            %s

            %s
            '''
        ),
        request.method,
        request.rel_url,
        *request.version,
        _dump_headers(request.headers),
        _dump_body(await request.read()),
    )
    resp = await handler(request)
    logger.debug(
        dedent('''\
            RESPONSE:
            HTTP/%d.%d %d %s
            %s

            %s
            '''
        ),
        *request.version,
        resp.status,
        resp.reason,
        _dump_headers(resp.headers),
        _dump_body(resp.body),
    )
    return resp