Created
June 4, 2019 12:40
-
-
Save ScheerMT/8a4016af2d5206edc16a17a525e8ac7d to your computer and use it in GitHub Desktop.
requires six to be installed to be able to run
This file contains hidden or 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
from functools import wraps, partial | |
import logging | |
import six | |
# Testing function decorator version | |
def log_func(): | |
return log_func_internal | |
def log_func_internal(func): | |
@wraps(func) | |
def wrapper(*args, **kwargs): | |
args_repr = [repr(a) for a in args] | |
#kwargs_repr = [f"{k}={v!r}" for k, v in kwargs.items()] | |
kwargs_repr = ["{0}={1}".format(k,v) for k, v in kwargs.items()] | |
kwargs_repr_dict = {k: v for k, v in kwargs.items()} | |
signature = ", ".join(args_repr + kwargs_repr) | |
print(kwargs_repr_dict) | |
print("Calling {0}({1})".format(func.__name__, signature)) | |
print(func.__module__) | |
returned_value = func(*args, **kwargs) | |
print("{0} returned {1}".format(func.__name__, returned_value)) | |
return returned_value | |
return wrapper | |
# TODO: maybe implement a function to allow better optional handling (right now default decorator use forces a set of closing parenthesis? | |
# https://chase-seibert.github.io/blog/2013/12/17/python-decorator-optional-parameter.html# | |
# TODO: Rename class lol | |
class LogMe(object): | |
def __init__(self, *args, **kwargs): | |
self.__wfunc = None | |
self.service_name = kwargs.pop('service_name', None) | |
self.logger = logging.getLogger('default') | |
self.other_args = kwargs | |
@property | |
def service_name(self): | |
"""Each logger has should be associated with a service for ease of tracking in logs. | |
If no name is defined, we fallback onto a default named 'SERVICE_DEFAULT_NAME' | |
""" | |
return self._service_name | |
@service_name.setter | |
def service_name(self, value): | |
DEFAULT_SERVICE_NAME = 'SERVICE_DEFAULT_NAME' | |
self._service_name = value if value else DEFAULT_SERVICE_NAME | |
@property | |
def __wfunc(self): | |
"""The wrapped function we will be working with""" | |
return self.___wfunc if self.___wfunc else None | |
@__wfunc.setter | |
def __wfunc(self, value): | |
self.___wfunc = value | |
@property | |
def __function_name(self): | |
"""The __name__ of the wrapped function we are working with. | |
We fallback to an empoty string if we do not have a function | |
""" | |
return self.__wfunc.__name__ if self.__wfunc else '' | |
def debug(self, msg, **kwargs): | |
"""Log a debug level message with any extra keyword parameters | |
Arguments: | |
msg {string} -- A string that will be logged out | |
Keyword Arguments: | |
These will be appended to the logged message as key value pairs output to a string representation | |
""" | |
self.log(level=logging.DEBUG, msg=msg, **kwargs) | |
def info(self, msg, **kwargs): | |
"""Log an info level message with any extra keyword parameters | |
Arguments: | |
msg {string} -- A string that will be logged out | |
Keyword Arguments: | |
These will be appended to the logged message as key value pairs output to a string representation | |
""" | |
self.log(level=logging.INFO, msg=msg, **kwargs) | |
def warn(self, msg, **kwargs): | |
"""Log a warning level message with any extra keyword parameters | |
Arguments: | |
msg {string} -- A string that will be logged out | |
Keyword Arguments: | |
These will be appended to the logged message as key value pairs output to a string representation | |
""" | |
self.log(level=logging.WARNING, msg=msg, **kwargs) | |
def error(self, msg, **kwargs): | |
"""Log an error level message with any extra keyword parameters | |
Arguments: | |
msg {string} -- A string that will be logged out | |
Keyword Arguments: | |
These will be appended to the logged message as key value pairs output to a string representation | |
""" | |
self.log(level=logging.ERROR, msg=msg, **kwargs) | |
def exception(self, msg, **kwargs): | |
"""Log an error level message with any extra keyword parameters | |
Arguments: | |
msg {string} -- A string that will be logged out | |
Keyword Arguments: | |
These will be appended to the logged message as key value pairs output to a string representation | |
""" | |
self.error(msg=msg, exc_info=True, **kwargs) | |
def log(self, msg, **kwargs): | |
"""Log a message, at a specified level | |
Arguments: | |
msg {string} -- A string that will be logged out | |
Keyword Arguments: | |
level {Logging.DEBUG/INFO/etc} -- one of the defined logging levels | |
""" | |
# Determine level with a sane default | |
# TODO: get defensive about log levels passed in here - what should be allowed? | |
level = kwargs.pop('level', logging.DEBUG) | |
# Passing these through to properly allow stack traces/info | |
exc_info = kwargs.pop('exc_info', False) | |
# This is a py3 only option! | |
stack_info = kwargs.pop('stack_info', False) | |
extra = kwargs if kwargs else {} | |
if extra != {}: | |
extra.update(self.other_args) | |
else: | |
# If we don't have any arguments passed in, we need to still attach the other args passed in on init | |
extra = self.other_args if self.other_args else '' | |
# Keep it defensive - ensure we do not fail logging on string conversion | |
try: | |
extra_str = str(extra) | |
except: | |
extra_str = {} | |
if six.PY2: | |
# Log that message! | |
self.logger.log( | |
level=level, | |
exc_info=exc_info, | |
extra=extra, | |
msg= self.service_name + '::' + self.__function_name + '::' + msg + ':::' + extra_str | |
) | |
else: | |
# Log that message! | |
self.logger.log( | |
level=level, | |
exc_info=exc_info, | |
stack_info=stack_info, | |
extra=extra, | |
msg= self.service_name + '::' + self.__function_name + '::' + msg + ':::' + extra_str | |
) | |
# Callable! this is what helps us turn this class into a decorator | |
# an attempt is made to ensure, if attempted to be used as a callable class outside of a function call, we have a fallback | |
def __call__(self, func=None): | |
# Ensure we update our stored function that we are working with wrapping | |
# This is needed for logging output of function name etc | |
self.__wfunc = func | |
""" Determine how best to use this?""" | |
def wrapped(*args, **kwargs): | |
self.log('bare call' + str(kwargs)) | |
# This needs to appear first because the @wraps call bombsa out if you are not wrapping via a decorator call | |
if func is None: | |
return wrapped | |
# Decorator magic~ | |
@wraps(func) | |
def wrapped_func(*args, **kwargs): | |
args_repr = [repr(a) for a in args] | |
kwargs_repr = ["{0}={1}".format(k,v) for k, v in kwargs.items()] | |
signature = ", ".join(args_repr + kwargs_repr) | |
# Pre function call logging | |
self.log( | |
'Calling the function with the following', | |
func_signature="{0}({1})".format(self.__function_name, signature), | |
args_list=", ".join(args_repr), | |
kwargs_list=", ".join(kwargs_repr), | |
kwargs_dict=kwargs | |
) | |
# Ensure we run the wrapped func and capture its return to be logged and returned | |
returned_value = func(*args, **kwargs) | |
# Keep it defensive - ensure we do not fail logging on string conversion | |
try: | |
returned_value_str = str(returned_value) | |
except: | |
returned_value_str = '' | |
# Post function call logging | |
self.log('WhatWasTheReturn', return_value_str=returned_value_str) | |
# Returning value from function call | |
return returned_value | |
# Decorator magic~ | |
return wrapped_func | |
if __name__ == '__main__': | |
daServiceLogger = LogMe(service_name='daService', other='default', info=123) | |
daServiceExceptionLogger = LogMe(service_name='daService', type='exception') | |
@log_func() | |
def do_the_thing(one, two, three='four'): | |
print('hello!') | |
return 1 | |
@daServiceLogger | |
def do_the_thing_log_default(one, two, three='four'): | |
print('hello!') | |
return 1 | |
@LogMe(service_name='daService') | |
def do_the_thing_log(one, two, three='four'): | |
print('hello!') | |
return 1 | |
@LogMe() | |
def do_the_thing_log_bare(one, two, three='four'): | |
print('hello! - bare') | |
return 1 | |
class testOne(object): | |
def __init__(self): | |
self.one = 'yes' | |
self.two = {'one': 'three'} | |
self.three = 12345 | |
@LogMe() | |
def do_the_thing_log_class(): | |
return testOne() | |
##### TESTING ##### | |
logging.basicConfig(level=logging.DEBUG) | |
print('\n') | |
# Utilizes a default logger we setup and subsequently used as a decorator | |
do_the_thing_log_default(0,1,three=2) | |
print('\n') | |
# Uses decorator class directly and sets it up in the decorator | |
do_the_thing_log(1,2,three=3) | |
print('\n\n') | |
# Utilizes the decorator class with defaults | |
do_the_thing_log_bare(2,3,three=4) | |
print('\n\n') | |
# Defult class decorator setup - returning a class this time | |
do_the_thing_log_class() | |
print('\n\n') | |
# Calling the logger directly | |
LogMe(service_name='callWithDatInfo', other=1).info('heya!', one=2) | |
print('\n\n') | |
# testing directly... | |
daServiceLogger.debug('debug me!') | |
daServiceLogger.info('info me!') | |
daServiceLogger.warn('warn me!') | |
daServiceLogger.error('error me!') | |
daServiceLogger.log('critical me!', level=logging.CRITICAL) | |
try: | |
raise Exception('thing') | |
except Exception as e: | |
# Will log at error level and, by default, include exception information | |
daServiceExceptionLogger.exception('Catching this exception!') | |
# This methis allows you to still log exception info while logging at a lower level | |
daServiceExceptionLogger.warn('Catching this exception!', exc_info=True) | |
# ????? - default class, callable then calling that? very odd but wanted to make this safe too | |
LogMe()()(one=2) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment