# -*- coding: utf-8 -*-
from __future__ import absolute_import, division, print_function, unicode_literals
from json import JSONEncoder
from logging import getLogger, INFO, Handler, Formatter, StreamHandler, DEBUG
from pprint import pformat
from sys import stderr
log = getLogger(__name__)
root_log = getLogger()
DEBUG_FORMATTER = Formatter(
"[%(levelname)s] [%(asctime)s.%(msecs)03d] %(process)d %(name)s:%(funcName)s(%(lineno)d):\n"
"%(message)s\n",
"%Y-%m-%d %H:%M:%S")
INFO_FORMATTER = Formatter(
"[%(levelname)s] [%(asctime)s.%(msecs)03d] %(process)d %(name)s(%(lineno)d): %(message)s\n",
"%Y-%m-%d %H:%M:%S")
[docs]def set_root_level(level=INFO):
root_log.setLevel(level)
[docs]def attach_stderr(level=INFO):
has_stderr_handler = any(handler.name == 'stderr' for handler in root_log.handlers)
if not has_stderr_handler:
handler = StreamHandler(stderr)
handler.name = 'stderr'
if level is not None:
handler.setLevel(level)
handler.setFormatter(DEBUG_FORMATTER if level == DEBUG else INFO_FORMATTER)
root_log.addHandler(handler)
return True
else:
return False
[docs]def detach_stderr():
for handler in root_log.handlers:
if handler.name == 'stderr':
root_log.removeHandler(handler)
return True
return False
[docs]def initialize_logging(level=INFO):
attach_stderr(level)
[docs]class NullHandler(Handler):
[docs] def emit(self, record):
pass
[docs]class DumpEncoder(JSONEncoder):
[docs] def default(self, obj):
if hasattr(obj, 'dump'):
return obj.dump()
# Let the base class default method raise the TypeError
return super(DumpEncoder, self).default(obj)
_DUMPS = DumpEncoder(indent=2, ensure_ascii=False, sort_keys=True).encode
[docs]def jsondumps(obj):
return _DUMPS(obj)
[docs]def fullname(obj):
return obj.__module__ + "." + obj.__class__.__name__
request_header_sort_dict = {
'Host': '\x00\x00',
'User-Agent': '\x00\x01',
}
response_header_sort_dict = {
'Content-Length': '\x7e\x7e\x61',
'Connection': '\x7e\x7e\x62',
}
[docs]def stringify(obj):
def bottle_builder(builder, bottle_object):
builder.append("{0} {1}{2} {3}".format(bottle_object.method,
bottle_object.path,
bottle_object.environ.get('QUERY_STRING', ''),
bottle_object.get('SERVER_PROTOCOL')))
builder += ["{0}: {1}".format(key, value) for key, value in bottle_object.headers.items()]
builder.append('')
body = bottle_object.body.read().strip()
if body:
builder.append(body)
def requests_models_PreparedRequest_builder(builder, request_object):
builder.append("> {0} {1} {2}".format(request_object.method, request_object.path_url,
request_object.url.split(':', 1)[0].upper()))
builder.extend("> {0}: {1}".format(key, value)
for key, value in sorted(request_object.headers.items(),
key=request_header_sort_key))
builder.append('')
if request_object.body:
builder.append(request_object.body)
def requests_models_Response_builder(builder, response_object):
builder.append("< {0} {1} {2}".format(response_object.url.split(':', 1)[0].upper(),
response_object.status_code, response_object.reason))
builder.extend("> {0}: {1}".format(key, value)
for key, value in sorted(response_object.headers.items(),
key=response_header_sort_key))
builder.append('')
content_type = response_object.headers.get('Content-Type')
if content_type == 'application/json':
builder.append(pformat(response_object.json, indent=2))
builder.append('')
elif content_type is not None and content_type.startswith('text/'):
builder.append(response_object.text)
try:
name = fullname(obj)
builder = [''] # start with new line
if name.startswith('bottle.'):
bottle_builder(builder, obj)
elif name.endswith('requests.models.PreparedRequest'):
requests_models_PreparedRequest_builder(builder, obj)
elif name.endswith('requests.models.Response'):
requests_models_PreparedRequest_builder(builder, obj.request)
requests_models_Response_builder(builder, obj)
else:
return None
builder.append('') # end with new line
return "\n".join(builder)
except Exception as e:
log.exception(e)