Created
February 27, 2017 01:29
-
-
Save gchina/7fb1be2c0f11c57a9cd8e51c9f8d8c1b to your computer and use it in GitHub Desktop.
A decorator that provides input/output debug logging for function calls.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
This module provides a decorator that can be added to any function to provide | |
debug output that shows the values of the arguments passed in as well as the | |
return value. If a Logger object is passed in to the decorator, all debug | |
output will go to that Logger object. Otherwise, the output will go to the | |
default Logger object. | |
Tip: If using the default Logger object, don't forget to set the logging level | |
to DEBUG if you want to see output on stdout. | |
""" | |
import sys | |
import os | |
import functools | |
from logging import Logger, getLogger | |
import inspect | |
def funclog(logger): | |
"""A decorator function that provides debug input/output logging.""" | |
# If a Logger object is passed in, use that. Otherwise, get default Logger. | |
real_logger = logger if isinstance(logger, Logger) else getLogger() | |
# __qualname__ is prettier but it didn't get added until 3.5 | |
name_attr = '__name__' if sys.version_info < (3, 5) else '__qualname__' | |
def get_arg_string(args, kwargs): | |
"""Convert args and kwargs to a pretty string.""" | |
return ', '.join(["'{}'".format(a) if type(a) == str else | |
'{}'.format(a) for a in args] + | |
["{}='{}'".format(a, v) if type(v) == str else | |
'{}={}'.format(a, v) for a, v in kwargs.items()]) | |
def real_decorator(fn): | |
@functools.wraps(fn) | |
def wrapper(*args, **kwargs): | |
frame = inspect.stack()[1][0] | |
frame_info = inspect.getframeinfo(frame) | |
filename = os.path.basename(frame_info.filename) | |
lineno = frame_info.lineno | |
func_name = getattr(fn, name_attr) | |
arg_string = get_arg_string(args, kwargs) | |
source_info = '{}:{}:{}({})'.format(filename, lineno, func_name, | |
arg_string) | |
real_logger.debug('calling {}'.format(source_info)) | |
try: | |
res = fn(*args, **kwargs) | |
except Exception as e: | |
real_logger.exception('{} threw exception:\n{}'.format(source_info, e)) | |
raise | |
real_logger.debug('{} returned: {}'.format(source_info, res)) | |
return res | |
return wrapper | |
if type(logger) == type(real_decorator): | |
return real_decorator(logger) | |
return real_decorator |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import logging | |
from funclog import funclog | |
logging.basicConfig(level=logging.DEBUG) | |
@funclog | |
def foo(a, b, c=None): | |
return (a * c) + b if c else a - b | |
class Foo(object): | |
def __init__(self, d, e): | |
self.d = d | |
self.e = e | |
@funclog | |
def instance_method(self): | |
return self.d + self.e | |
foo(5, 12, c=3) | |
for i in range(3): | |
foo(3 * i, 8) | |
Foo(12, 19).instance_method() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
Unit tests for funclog decorator. | |
Currently the method tests expected output is formatted for Python >=3.5. | |
TODO: Right now the expected line numbers are hard-coded, which is just stupid. | |
""" | |
import pytest | |
import logging | |
import io | |
from funclog import funclog | |
logfmt = '%(name)s %(levelname)s %(message)s' | |
root_buffer = io.StringIO() | |
logging.basicConfig(format=logfmt, level=logging.DEBUG, | |
handlers=[logging.StreamHandler(root_buffer)]) | |
log_buffer = io.StringIO() | |
logger = logging.getLogger('test') | |
logger.setLevel(logging.DEBUG) | |
fh = logging.StreamHandler(log_buffer) | |
fh.setFormatter(logging.Formatter(logfmt)) | |
logger.addHandler(fh) | |
@funclog(logger) | |
def foo(a, b, c=None): | |
"""A simple function with args and kwargs.""" | |
return (a + c) / b if c else a / b | |
@funclog | |
def bar(a=None, b=None): | |
"""A simple function with only kwargs.""" | |
return a == b | |
class Foo(object): | |
"""Test class with instance, static and class methods.""" | |
@funclog(logger) | |
def instance_method(self, a, b): | |
"""Test instance method.""" | |
return a + b | |
@staticmethod | |
@funclog(logger) | |
def static_method(a, b): | |
"""Test static method.""" | |
return a - b | |
@classmethod | |
@funclog(logger) | |
def class_method(cls, a, b): | |
"""Test class method.""" | |
return a * b | |
def test_simple_functions(): | |
root_buffer.truncate(0) | |
root_buffer.seek(0) | |
log_buffer.truncate(0) | |
log_buffer.seek(0) | |
expected_buf = [ | |
'root DEBUG calling test_funclog.py:70:bar()', | |
'root DEBUG test_funclog.py:70:bar() returned: True' | |
] | |
bar() | |
temp_buf = root_buffer.getvalue().strip().split('\n') | |
assert expected_buf == temp_buf | |
assert len(log_buffer.getvalue()) == 0 | |
root_buffer.truncate(0) | |
root_buffer.seek(0) | |
log_buffer.truncate(0) | |
log_buffer.seek(0) | |
expected_buf = [ | |
"test DEBUG calling test_funclog.py:83:foo(12, 3, c=6)", | |
"test DEBUG test_funclog.py:83:foo(12, 3, c=6) returned: 6.0" | |
] | |
foo(12, 3, c=6) | |
temp_buf = root_buffer.getvalue().strip().split('\n') | |
assert expected_buf == temp_buf | |
temp_buf = log_buffer.getvalue().strip().split('\n') | |
assert expected_buf == temp_buf | |
root_buffer.truncate(0) | |
root_buffer.seek(0) | |
log_buffer.truncate(0) | |
log_buffer.seek(0) | |
expected_buf = [ | |
"test DEBUG calling test_funclog.py:99:foo(12, 3, c='hello')", | |
"test ERROR test_funclog.py:99:foo(12, 3, c='hello') threw exception:", | |
"unsupported operand type(s) for +: 'int' and 'str'" | |
] | |
with pytest.raises(TypeError): | |
foo(12, 3, c='hello') | |
temp_buf = root_buffer.getvalue().strip().split('\n') | |
assert expected_buf == temp_buf[:3] | |
temp_buf = log_buffer.getvalue().strip().split('\n') | |
assert expected_buf == temp_buf[:3] | |
root_buffer.truncate(0) | |
root_buffer.seek(0) | |
log_buffer.truncate(0) | |
log_buffer.seek(0) | |
expected_buf = [ | |
"root DEBUG calling test_funclog.py:113:bar('a', 1)", | |
"root DEBUG test_funclog.py:113:bar('a', 1) returned: False" | |
] | |
bar('a', 1) | |
temp_buf = root_buffer.getvalue().strip().split('\n') | |
assert expected_buf == temp_buf | |
assert len(log_buffer.getvalue()) == 0 | |
root_buffer.truncate(0) | |
root_buffer.seek(0) | |
log_buffer.truncate(0) | |
log_buffer.seek(0) | |
expected_buf = [ | |
"test DEBUG calling test_funclog.py:128:foo('a', 3)", | |
"test ERROR test_funclog.py:128:foo('a', 3) threw exception:", | |
"unsupported operand type(s) for /: 'str' and 'int'" | |
] | |
with pytest.raises(TypeError): | |
foo('a', 3) | |
temp_buf = root_buffer.getvalue().strip().split('\n') | |
assert expected_buf == temp_buf[:3] | |
temp_buf = log_buffer.getvalue().strip().split('\n') | |
assert expected_buf == temp_buf[:3] | |
# Testing kwargs-only arguments. This is tricky because the kwargs | |
# dictionary could be randomly re-ordered. | |
root_buffer.truncate(0) | |
root_buffer.seek(0) | |
log_buffer.truncate(0) | |
log_buffer.seek(0) | |
expected_buf = [ | |
"root DEBUG calling test_funclog.py:144:bar(a='a', b=1)", | |
"root DEBUG test_funclog.py:144:bar(a='a', b=1) returned: False", | |
] | |
bar(a='a', b=1) | |
temp_buf = root_buffer.getvalue().strip().split('\n') | |
assert expected_buf == temp_buf | |
assert len(log_buffer.getvalue()) == 0 | |
root_buffer.truncate(0) | |
root_buffer.seek(0) | |
log_buffer.truncate(0) | |
log_buffer.seek(0) | |
expected_buf = [ | |
"root DEBUG calling test_funclog.py:157:bar(a='a', b=<function foo at ", | |
"root DEBUG test_funclog.py:157:bar(a='a', b=<function foo at " | |
] | |
bar(a='a', b=foo) | |
temp_buf = root_buffer.getvalue().strip().split('\n') | |
assert temp_buf[0].startswith(expected_buf[0]) | |
assert temp_buf[1].startswith(expected_buf[1]) | |
assert len(log_buffer.getvalue()) == 0 | |
root_buffer.truncate(0) | |
root_buffer.seek(0) | |
log_buffer.truncate(0) | |
log_buffer.seek(0) | |
expected_buf = [ | |
"test DEBUG calling test_funclog.py:173:foo(4, 2)", | |
"test DEBUG test_funclog.py:173:foo(4, 2) returned: 2.0", | |
"root DEBUG calling test_funclog.py:173:bar(a='a', b=2.0)", | |
"root DEBUG test_funclog.py:173:bar(a='a', b=2.0) returned: False", | |
] | |
bar(a='a', b=foo(4, 2)) | |
temp_buf = root_buffer.getvalue().strip().split('\n') | |
assert expected_buf == temp_buf | |
temp_buf = log_buffer.getvalue().strip().split('\n') | |
assert expected_buf[:2] == temp_buf | |
def test_simple_methods(): | |
root_buffer.truncate(0) | |
root_buffer.seek(0) | |
expected_buf = [ | |
"test DEBUG calling test_funclog.py:187:Foo.instance_method(<test_funclog.Foo object at", | |
"test DEBUG test_funclog.py:187:Foo.instance_method(<test_funclog.Foo object at" | |
] | |
Foo().instance_method(1, 2) | |
temp_buf = root_buffer.getvalue().strip().split('\n') | |
assert temp_buf[0].startswith(expected_buf[0]) | |
assert temp_buf[1].startswith(expected_buf[1]) | |
root_buffer.truncate(0) | |
root_buffer.seek(0) | |
expected_buf = [ | |
"test DEBUG calling test_funclog.py:198:Foo.static_method(12, 3)", | |
"test DEBUG test_funclog.py:198:Foo.static_method(12, 3) returned: 9" | |
] | |
Foo.static_method(12, 3) | |
temp_buf = root_buffer.getvalue().strip().split('\n') | |
assert expected_buf == temp_buf | |
root_buffer.truncate(0) | |
root_buffer.seek(0) | |
expected_buf = [ | |
"test DEBUG calling test_funclog.py:208:Foo.class_method(<class 'test_funclog.Foo'>, 5, 6)", | |
"test DEBUG test_funclog.py:208:Foo.class_method(<class 'test_funclog.Foo'>, 5, 6) returned: 30", | |
] | |
Foo.class_method(5, 6) | |
temp_buf = root_buffer.getvalue().strip().split('\n') | |
assert expected_buf == temp_buf |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment