verbose_proxy.py 1.6 KB

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