Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
requires six to be installed to be able to run
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
You can’t perform that action at this time.