Request/Response Logging in Python

Sometimes you just don't like the logging options other packages provide. At JBS we've found that existing logging around external requests and responses didn't quite fit our needs. This is a simplified version of our requests libs monkey patch that will show you how you can wrap your own logging around any requests that happen through requests lib.

The ShowExtra log formatter will help you leverage this data, by printing formatted JSON to the console. And the `setup_logging` method will wire it up for you.

# requests_monkey_patch.py:

import json
import logging
import os
from datetime import datetime

import requests

request_logger = logging.getLogger('request_logger')
original_request_method = requests.request
PRETTY_OUTPUT = True


class ShowExtra(logging.Formatter):
    def format(self, record):
        log_data = {
            'msg': super().format(record),
            'start_time': getattr(record, 'start_time', ''),
            'end_time': getattr(record, 'end_time', ''),
            'response_time_ms': getattr(record, 'response_time_ms', ''),
            'request': record.request,
            'response': getattr(record, 'response', None),
            **getattr(record, 'extra', {}),
        }
        kwargs = {}
        if PRETTY_OUTPUT:
            kwargs['indent'] = 4
        resp = json.dumps(log_data, **kwargs)
        return resp


def logged_request(method, url, **kwargs):
    # Yes, or True in the environment will enable
    enabled = os.environ.get('ENABLE_REQUEST_LOGGING', 'false').lower()[0] in ('y', 't')
    if not enabled:
        return original_request_method(method, url, **kwargs)

    request = None
    response = None
    start_time = datetime.now()

    try:
        response = original_request_method(method, url, **kwargs)
        request = response.request
    finally:
        end_time = datetime.now()
        log_request_response(
            start_time=start_time,
            end_time=end_time,
            request=request,
            response=response,
            method=method,
            url=url,
        )

    return response


def is_protected(key):
    key_lower = key.lower()
    for test in ('key', 'secret'):
        if test in key_lower:
            return True
    return False


def sanitize_data(data: dict):
    # This would be a good place to remove or sensor certain keys or values
    return {
        k: '*****' if is_protected(k) else v
        for k, v in data.items()
    }


def log_request_response(
        *,  # Only Keyword arguments
        start_time,
        end_time,
        request=None,
        response=None,
        method=None,
        url=None,
):
    request_info = {}
    response_info = {}

    log_msg = f'{method} {url} (Failed)'

    if request is not None:
        request_info = {
            'method': request.method,
        }

        # TODO: parse url to niceify the query string
        request_info['url'] = request.url
        request_info['path'] = request.path_url
        request_info['headers'] = sanitize_data(dict(request.headers))

        if request.body:
            if isinstance(request.body, bytes):
                body = request.body.decode('utf-8', 'replace')
            else:
                body = request.body

            try:
                request_info['data'] = sanitize_data(json.loads(body))
            except json.JSONDecodeError:
                request_info['data'] = body

        log_msg = f'{request.method} {request.url}'

        if response is not None:
            log_msg += f' ({response.status_code})'

            response_info['status_code'] = response.status_code

            response_text = getattr(response, 'text', '')
            try:
                response_info['data'] = sanitize_data(json.loads(response_text))
            except json.JSONDecodeError:
                response_info['data'] = response_text

    extra = {
        'start_time': start_time.strftime('%Y-%m-%d %H:%M:%S.%f'),
        'end_time': end_time and end_time.strftime('%Y-%m-%d %H:%M:%S.%f'),
        'response_time_ms': end_time and int((end_time - start_time).total_seconds() * 1000),
        'request': request_info,
        'response': response_info,
    }

    request_logger.debug(log_msg, extra=extra)


def patcher():
    # Import this early and call it to monkey patch the two places where this method might be imported
    requests.api.request = logged_request
    requests.request = logged_request

def setup_logging():
    logging.basicConfig(level=logging.DEBUG)
    ch = logging.StreamHandler()
    ch.setFormatter(ShowExtra())
    request_logger.addHandler(ch)

# Example Usage

import requests_monkey_patch as mp
import requests
mp.patcher()
mp.setup_logging()
x = requests.get('http://xkcd.com/info.0.json')