Skip to content

Instantly share code, notes, and snippets.

@ckesanapalli
Last active March 28, 2022 14:13
Show Gist options
  • Save ckesanapalli/2cbdbf75c690726ec0762753f0cfdbb8 to your computer and use it in GitHub Desktop.
Save ckesanapalli/2cbdbf75c690726ec0762753f0cfdbb8 to your computer and use it in GitHub Desktop.
Logging configuration code with wrapper that log entry, exit and runtime of functions with a decorator using logging module
import logging
from datetime import datetime
from pathlib import Path
import functools
from pprint import pformat
def get_func_name(func):
"""Return the name of the given function including module name"""
if func.__class__.__module__ == "builtins":
mod_name = func.__module__ if hasattr(func, '__module__') and \
func.__module__ != None else ""
qual_name = func.__qualname__ if hasattr(func, '__qualname__') and \
func.__qualname__ != None else func.__name__
else:
class_func = func.__class__
mod_name = class_func.__module__ if hasattr(class_func, '__module__') \
and class_func.__module__ != None else ""
qual_name = (class_func.__qualname__ if hasattr(class_func, '__qualname__') \
and class_func.__qualname__ != None else class_func.__name__) \
+ '.' + func.__name__
return '.'.join([mod_name, qual_name])
def get_func_small_doc(func):
"""Return first two lines from the function documentation"""
try:
split_doc = func.__doc__.split("\n")
if len(split_doc) < 2:
doc = ' '.join(split_doc).strip()
else:
doc = ' '.join(split_doc[:2]).strip()
except:
doc = None
return doc
def get_logger_wrap(logger, indent=4, sort_dicts=False, **kwargs):
var2str = functools.partial(pformat, indent=indent,
sort_dicts=sort_dicts, **kwargs)
def logger_wrap(*, func_entry=True,
func_exit=True, level=logging.DEBUG, debug_io=False, debug_args=False,
debug_kwargs=False, debug_result=False, log_full_doc=False):
"""
A wrapper function to log the function or class entry, exit and runtime
outputting arguments and results
Parameters
----------
logger: logging.Logger
Logger
func_entry : bool, optional
Whether to log the entry of function. The default is True.
func_exit : bool, optional
Whether to log the exit of function. The default is True.
level : int, optional
logging level. The default is 0 (DEBUG).
debug_io : bool, optional
Whether to log function inputs and output parameters. The default is False.
debug_args : bool, optional
Whether to log function arguments. The default is False.
debug_kwargs : bool, optional
Whether to log function keyword arguments. The default is False.
debug_result : bool, optional
Whether to log function result. The default is False.
is_full_doc : bool, optional
Whether to log full documentation of function.
If False, first two lines of function documentation are logged.
The default is False.
Returns
-------
decorator:
An object that can be used to decorate a function or a class.
"""
def wrapper(func):
name = get_func_name(func)
doc = func.__doc__ if log_full_doc else get_func_small_doc(func)
@functools.wraps(func)
def wrapped(*args, **kwargs):
if func_entry:
entry_dict = {"State": "ENTRY", "Name": name, "Description": doc}
if debug_args or debug_io:
entry_dict["Args"] = args
if debug_kwargs or debug_io:
entry_dict["Kwargs"] = kwargs
entry_str = var2str(entry_dict)
logger.log(level, entry_str, stacklevel=2)
start = datetime.now()
result = func(*args, **kwargs)
end = datetime.now()
runtime = str(end - start)
if func_exit:
exit_dict = {"State": "EXIT", "Name": name, "Runtime": runtime}
if debug_result or debug_io:
exit_dict["Result"] = result
exit_str = var2str(exit_dict)
logger.log(level, exit_str, stacklevel=2)
return result
return wrapped
return wrapper
return logger_wrap
class CustomFormatter(logging.Formatter):
"""Logging colored formatter, adapted from https://stackoverflow.com/a/56944256/3638629"""
grey = '\x1b[38;21m'
blue = '\x1b[38;5;39m'
yellow = '\x1b[38;5;226m'
red = '\x1b[38;5;196m'
bold_red = '\x1b[31;1m'
reset = '\x1b[0m'
def __init__(self, fmt):
super().__init__()
self.fmt = fmt
self.FORMATS = {
logging.DEBUG: self.grey + self.fmt + self.reset,
logging.INFO: self.blue + self.fmt + self.reset,
logging.WARNING: self.yellow + self.fmt + self.reset,
logging.ERROR: self.red + self.fmt + self.reset,
logging.CRITICAL: self.bold_red + self.fmt + self.reset
}
def format(self, record):
log_fmt = self.FORMATS.get(record.levelno)
formatter = logging.Formatter(log_fmt)
return formatter.format(record)
def updated_logger(logger, file_name, log_dir, level=logging.DEBUG, timed_path=True, **kwargs):
"""
Return updated logger with the log file path and the logging level
Parameters
----------
logger : logging.logger
Logger.
file_name : str
name of the logger file.
log_dir : str or Path
Path of the log directory.
level : int, optional
Level of the logging. The default is logging.DEBUG.
timed_path : bool, optional
whether to add datetime to the file suffix. The default is True.
Returns
-------
logger: logging.logger
Updated Logger
logger_wrap: decorator
Log decorator for functions
"""
log_dir = Path(log_dir)
log_dir.mkdir(parents=True, exist_ok=True)
logger.setLevel(level)
time_str = datetime.strftime(datetime.now(), '%Y_%m_%d_%H_%M_%S')
log_file_name = f"{file_name}_{time_str}.log" if timed_path else f"{file_name}.log"
log_file_path = log_dir / log_file_name
fmt_str = "%(asctime)s | %(levelname)-8s | \
%(module)s:%(name)s:%(funcName)s:%(lineno)d - %(message)s"
formatter = logging.Formatter(fmt_str)
# create console handler with a higher log level
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
ch.setFormatter(CustomFormatter(fmt_str))
logger.addHandler(ch)
# create file handler which logs even debug messages
fh = logging.FileHandler(log_file_path, mode='w')
fh.setLevel(logging.DEBUG)
fh.setFormatter(formatter)
logger.addHandler(fh)
logger_wrap = get_logger_wrap(logger, **kwargs)
return logger, logger_wrap
def get_heading(logger, title, level=0, total_len=79):
"""
Converts 'string' to a heading ('== string ==')
Parameters
----------
logger : logging.logger
logger object
title : str
title string before formatting
level : int, optional
Heading type ranging from 0 to 2, by default 0
total_len : int, optional
Maximum desired length of heading, by default 79
Returns
-------
logger.info
Outputs logger.info(string)
"""
title_len = len(title)
if title_len > total_len:
logger.warning(
f"Title length ({title_len}) is greater than Total length ({total_len})"
)
title_len = total_len - 2
line1_len = (total_len - title_len) // 2
line2_len = total_len - (title_len + line1_len)
if level == 0:
line_chr = "="
elif level == 1:
line_chr = "-"
elif level == 2:
line_chr = ":"
else:
line_chr = "."
line1 = line_chr * line1_len
line2 = line_chr * line2_len
formated_heading = " ".join([line1, title, line2])
return logger.info(formated_heading)
import time
import logging
from log_conf import updated_logger
from dataclasses import dataclass
logger, logger_wrap = updated_logger(logging.getLogger(__name__), "test",
indent=4, sort_dicts=False)
@logger_wrap(debug_io=True)
def my_func(a, b, c=1, d="Text"):
"""A test function"""
logger.info("Sleeping for 1 sec")
time.sleep(1)
return str([a, b, c, d])
@logger_wrap(debug_io=True)
@dataclass
class my_class():
"""A test class"""
a : int
area: float
test_class = my_class(2, 3)
my_func("hello", "logger", c="print this", d="keyword arg")
# Logging Existing library functions
import numpy as np
logger_wrap(debug_io=True)(np.random.rand)(10,10)
logger_wrap(log_full_doc=True)(np.sin)(np.random.rand(10,10))
2022-03-28 02:00:01,643 | DEBUG | main:__main__:<module>:23 - { State: ENTRY,
Name: __main__.my_class,
Description: A test class,
Args: (2, 3),
Kwargs: {}}
2022-03-28 02:00:01,644 | DEBUG | main:__main__:<module>:23 - { State: EXIT,
Name: __main__.my_class,
Runtime: 0:00:00,
Result: my_class(a=2, area=3)}
2022-03-28 02:00:01,644 | WARNING | main:__main__:<module>:24 - Not Sleeping
2022-03-28 02:00:01,644 | DEBUG | main:__main__:<module>:26 - { State: ENTRY,
Name: __main__.my_func,
Description: A test function,
Args: (hello, logger),
Kwargs: {c: print this, d: keyword arg}}
2022-03-28 02:00:01,645 | INFO | main:__main__:my_func:12 - Sleeping for 1 sec
2022-03-28 02:00:02,653 | DEBUG | main:__main__:<module>:26 - { State: EXIT,
Name: __main__.my_func,
Runtime: 0:00:01.008160,
Result: [hello, logger, print this, keyword arg]}
2022-03-28 02:00:02,654 | DEBUG | main:__main__:<module>:30 - { State: ENTRY,
Name: .RandomState.rand,
Description: rand(d0, d1, ..., dn),
Args: (10, 10),
Kwargs: {}}
2022-03-28 02:00:02,655 | DEBUG | main:__main__:<module>:30 - { State: EXIT,
Name: .RandomState.rand,
Runtime: 0:00:00,
Result: [[0.2363461 0.48550162 0.13387681 0.87350022 0.73049686
0.92323771
0.26544497 0.62579515 0.64877539 0.85464577]
[0.64606151 0.83008665 0.49724045 0.5762315 0.21503162
0.75508816
0.36645548 0.97247661 0.35752656 0.66461037]
[0.52730874 0.77483066 0.43904472 0.76365511 0.60058024
0.89924877
0.36866417 0.38233081 0.68954406 0.25537451]
[0.3137653 0.87982569 0.05163364 0.97376399 0.26522001
0.2569054
0.39716299 0.36478841 0.15514908 0.9514152 ]
[0.17787851 0.75302585 0.53634242 0.93452775 0.32741251
0.6441435
0.67265579 0.21157967 0.71253889 0.50326617]
[0.84149008 0.37644909 0.42836803 0.99342584 0.73844244
0.25168303
0.29965424 0.49836129 0.81708351 0.19190296]
[0.70483546 0.66344476 0.44611424 0.49046865 0.97799542
0.92617753
0.04034599 0.83111146 0.26974899 0.47880171]
[0.0913788 0.2426281 0.33022209 0.25033408 0.9195304
0.82793309
0.8876626 0.71317787 0.14346796 0.33335825]
[0.34448748 0.62689749 0.64079311 0.51878668 0.9918069
0.85796244
0.70466754 0.55603669 0.95307835 0.92907495]
[0.43503599 0.16428707 0.41947888 0.10935243 0.32389202
0.59091364
0.29246639 0.44738868 0.56157866 0.38954625]]}
2022-03-28 02:00:02,656 | DEBUG | main:__main__:<module>:31 - { State: ENTRY,
Name: numpy.ufunc.sin,
Description: sin(x, /, out=None, *, where=True, casting=same_kind,
order=K, dtype=None, subok=True[, signature, extobj])
Trigonometric sine, element-wise.
Parameters
----------
x : array_like
Angle, in radians (:math:`2 \\pi` rad equals 360
degrees).
out : ndarray, None, or tuple of ndarray and None,
optional
A location into which the result is stored. If
provided, it must have
a shape that the inputs broadcast to. If not provided
or None,
a freshly-allocated array is returned. A tuple
(possible only as a
keyword argument) must have length equal to the number
of outputs.
where : array_like, optional
This condition is broadcast over the input. At
locations where the
condition is True, the `out` array will be set to the
ufunc result.
Elsewhere, the `out` array will retain its original
value.
Note that if an uninitialized `out` array is created
via the default
``out=None``, locations within it where the condition
is False will
remain uninitialized.
**kwargs
For other keyword-only arguments, see the
:ref:`ufunc docs <ufuncs.kwargs>`.
Returns
-------
y : array_like
The sine of each element of x.
This is a scalar if `x` is a scalar.
See Also
--------
arcsin, sinh, cos
Notes
-----
The sine is one of the fundamental functions of
trigonometry (the
mathematical study of triangles). Consider a circle of
radius 1
centered on the origin. A ray comes in from the
:math:`+x` axis, makes
an angle at the origin (measured counter-clockwise from
that axis), and
departs from the origin. The :math:`y` coordinate of the
outgoing
rays intersection with the unit circle is the sine of
that angle. It
ranges from -1 for :math:`x=3\\pi / 2` to +1 for
:math:`\\pi / 2.` The
function has zeroes where the angle is a multiple of
:math:`\\pi`.
Sines of angles between :math:`\\pi` and :math:`2\\pi` are
negative.
The numerous properties of the sine and related functions
are included
in any standard trigonometry text.
Examples
--------
Print sine of one angle:
>>> np.sin(np.pi/2.)
1.0
Print sines of an array of angles given in degrees:
>>> np.sin(np.array((0., 30., 45., 60., 90.)) * np.pi /
180. )
array([ 0. , 0.5 , 0.70710678, 0.8660254
, 1. ])
Plot the sine function:
>>> import matplotlib.pylab as plt
>>> x = np.linspace(-np.pi, np.pi, 201)
>>> plt.plot(x, np.sin(x))
>>> plt.xlabel(Angle [rad])
>>> plt.ylabel(sin(x))
>>> plt.axis(tight)
>>> plt.show()}
2022-03-28 02:00:02,658 | DEBUG | main:__main__:<module>:31 - {State: EXIT, Name: numpy.ufunc.sin, Runtime: 0:00:00}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment