Flask log request response
Tested with Python 2.7, 3.7.
Create FlaskLogRequestResponseDemo.py:
# coding: utf-8
from __future__ import absolute_import
from datetime import datetime
from flask import Flask
from flask import jsonify
from flask import request
import json
import logging
try:
from io import BytesIO
except ImportError:
from StringIO import StringIO as BytesIO
LOGGER = logging.getLogger(__name__)
FLASK_APP = Flask(__name__)
class SaveBodyWsgiApp(object):
def __init__(self, app):
self.app = app
def __call__(self, environ, start_response):
content_length = environ.get('CONTENT_LENGTH', '0')
content_length = 0 if not content_length else int(content_length)
body = environ['wsgi.input'].read(content_length)
environ['raw_body'] = body
environ['wsgi.input'] = BytesIO(body)
resp = self.app(
environ,
start_response,
)
return resp
@FLASK_APP.before_request
def before_request():
if not FLASK_APP.debug:
return
lines = []
time_now = datetime.now()
lines.append('Time: ' + time_now.strftime('%Y-%m-%d %H:%M:%S'))
lines.append('Remote IP: ' + request.remote_addr)
lines.append('Protocol: ' + str(request.scheme))
lines.append('Method: ' + request.method)
lines.append('Host: ' + request.host)
lines.append('URL: ' + request.url)
lines.append('Path: ' + request.path)
lines.append('Query: ' + repr(request.query_string))
header_texts = [
'{0}: {1}'.format(repr(x[0]), repr(x[1]))\
for x in request.headers
]
lines.append('Headers:\n```\n{0}\n```'.format('\n'.join(header_texts)))
lines.append('Query arguments:')
query_arguments = dict(request.args)
lines.append(
json.dumps(
query_arguments, ensure_ascii=False, indent=4, sort_keys=True,
)
)
lines.append('Body arguments:')
body_arguments = dict(request.form)
lines.append(
json.dumps(
body_arguments, ensure_ascii=False, indent=4, sort_keys=True,
)
)
body_data = request.environ.get('raw_body')
if body_data is None:
body_data = request.get_data()
lines.append('Body: {0}'.format(repr(body_data)))
if body_data:
content_type = request.headers.get('Content-Type')
if content_type == 'application/json':
try:
body_dict = json.loads(request.data.decode('utf-8'))
body_text = json.dumps(
body_dict, ensure_ascii=False, indent=4, sort_keys=True,
)
lines.append('Body dict:\n{0}'.format(body_text))
except Exception:
pass
message = '\n# ----- Request -----\n{0}\n'.format(
'\n'.join(lines),
)
LOGGER.debug(message)
@FLASK_APP.after_request
def after_request(response):
if not FLASK_APP.debug:
return response
lines = []
time_now = datetime.now()
lines.append('Time: ' + time_now.strftime('%Y-%m-%d %H:%M:%S'))
lines.append('Status: {0}'.format(response.status))
header_texts = [
'{0}: {1}'.format(repr(x[0]), repr(x[1]))\
for x in response.headers
]
lines.append('Headers:\n```\n{0}\n```'.format('\n'.join(header_texts)))
try:
body = response.get_data()
except Exception:
body = None
lines.append('Body: {0}'.format(repr(body)))
content_type = response.headers.get('Content-Type')
if content_type == 'application/json':
try:
body_dict = json.loads(body)
body_text = json.dumps(
body_dict, ensure_ascii=False, indent=4, sort_keys=True,
)
lines.append('Body dict:\n{0}'.format(body_text))
except Exception:
pass
message = '\n# ----- Response -----\n{0}\n'.format(
'\n'.join(lines),
)
LOGGER.debug(message)
return response
@FLASK_APP.route('/api/test', methods=['GET', 'POST'])
def api_test():
return jsonify({
'status': 200,
'code': 'SUCCESS',
})
def main(args=None):
try:
log_formatter = logging.Formatter(
'# %(asctime)s %(filename)s:L%(lineno)d %(levelname)s %(message)s'
)
log_handler = logging.StreamHandler()
log_handler.setLevel(logging.DEBUG)
log_handler.setFormatter(log_formatter)
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG)
root_logger.addHandler(log_handler)
debug = True
if debug:
FLASK_APP.wsgi_app = SaveBodyWsgiApp(FLASK_APP.wsgi_app)
FLASK_APP.run(
host='127.0.0.1',
port=8000,
debug=debug,
)
except KeyboardInterrupt:
pass
if __name__ == '__main__':
exit(main())
Run FlaskLogRequestResponseDemo.py:
python FlaskLogRequestResponseDemo.py
Send a request:
curl -v -X POST -H 'Content-Type: application/json' -d '{"x": 1, "y": 2}' \
'http://127.0.0.1:8000/api/test?a=1&b=2'
Log result:
# 2019-04-09 21:13:12,973 FlaskLogRequestResponseDemo.py:L124 DEBUG
# ----- Request -----
Time: 2019-04-09 21:13:12
Remote IP: 127.0.0.1
Protocol: http
Method: POST
Host: 127.0.0.1:8000
URL: http://127.0.0.1:8000/api/test?a=1&b=2
Path: /api/test
Query: b'a=1&b=2'
Headers:
```
'Host': '127.0.0.1:8000'
'User-Agent': 'curl/7.29.0'
'Accept': '*/*'
'Content-Type': 'application/json'
'Content-Length': '16'
```
Query arguments:
{
"a": [
"1"
],
"b": [
"2"
]
}
Body arguments:
{}
Body: b'{"x": 1, "y": 2}'
Body dict:
{
"x": 1,
"y": 2
}
# 2019-04-09 21:13:12,977 FlaskLogRequestResponseDemo.py:L172 DEBUG
# ----- Response -----
Time: 2019-04-09 21:13:12
Status: 200 OK
Headers:
```
'Content-Type': 'application/json'
'Content-Length': '42'
```
Body: b'{\n "code": "SUCCESS", \n "status": 200\n}\n'
Body dict:
{
"code": "SUCCESS",
"status": 200
}
Comments: