verbose_proxy.py 1.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657
  1. import functools
  2. import logging
  3. import pprint
  4. from itertools import chain
  5. import six
  6. def format_call(args, kwargs):
  7. args = (repr(a) for a in args)
  8. kwargs = ("{0!s}={1!r}".format(*item) for item in six.iteritems(kwargs))
  9. return "({0})".format(", ".join(chain(args, kwargs)))
  10. def format_return(result, max_lines):
  11. if isinstance(result, (list, tuple, set)):
  12. return "({0} with {1} items)".format(type(result).__name__, len(result))
  13. if result:
  14. lines = pprint.pformat(result).split('\n')
  15. extra = '\n...' if len(lines) > max_lines else ''
  16. return '\n'.join(lines[:max_lines]) + extra
  17. return result
  18. class VerboseProxy(object):
  19. """Proxy all function calls to another class and log method name, arguments
  20. and return values for each call.
  21. """
  22. def __init__(self, obj_name, obj, log_name=None, max_lines=10):
  23. self.obj_name = obj_name
  24. self.obj = obj
  25. self.max_lines = max_lines
  26. self.log = logging.getLogger(log_name or __name__)
  27. def __getattr__(self, name):
  28. attr = getattr(self.obj, name)
  29. if not six.callable(attr):
  30. return attr
  31. return functools.partial(self.proxy_callable, name)
  32. def proxy_callable(self, call_name, *args, **kwargs):
  33. self.log.info("%s %s <- %s",
  34. self.obj_name,
  35. call_name,
  36. format_call(args, kwargs))
  37. result = getattr(self.obj, call_name)(*args, **kwargs)
  38. self.log.info("%s %s -> %s",
  39. self.obj_name,
  40. call_name,
  41. format_return(result, self.max_lines))
  42. return result