Source code for pyquickhelper.loghelper.custom_log

# -*- coding: utf-8 -*-
"""
Creates a custom log (open a text file and flushes everything in it).


:githublink:`%|py|6`
"""
import datetime
import os


[docs]class CustomLog: """ Implements a custom logging function. This class is not protected against multithreading. Usage: :: clog = CustomLog("folder") clog('[fct]', info) :githublink:`%|py|20` """
[docs] def __init__(self, folder=None, filename=None, create=True, parent=None): """ initialisation :param folder: folder (created if not exists) :param filename: new filename :param create: force the creation :param parent: logging function (called after this one if not None) :githublink:`%|py|30` """ folder = os.path.abspath(folder) self._folder = folder self._parent = parent if not os.path.exists(folder): os.makedirs(folder) typstr = str if filename is None: i = 0 filename = "log_custom_%03d.txt" % i fullpath = os.path.join(folder, filename) while os.path.exists(fullpath): i += 1 filename = "custom_log_%03d.txt" % i fullpath = os.path.join(folder, filename) self._filename = filename self._fullpath = fullpath self._handle = open(self._fullpath, "w", encoding="utf-8") self._close = True elif isinstance(filename, typstr): self._filename = filename self._fullpath = os.path.join(folder, filename) self._handle = open(self._fullpath, "w", encoding="utf-8") self._close = True else: self._handle = filename self._close = False self._filename = None self._fullpath = None
@property def filename(self): """ returns *_filename* :githublink:`%|py|64` """ return self._filename @property def fullpath(self): """ returns *_fullpath* :githublink:`%|py|71` """ return self._fullpath
[docs] def __del__(self): """ Closes the stream if needed. :githublink:`%|py|77` """ if self._close: self._handle.close()
[docs] def __call__(self, *l, **p): """ Log anything. :githublink:`%|py|84` """ self.fLOG(*l, **p) if self._parent is not None: self._parent(*l, **p)
[docs] def fLOG(self, *l, **p): """ Builds a message on a single line with the date, it deals with encoding issues. :param l: list of fields :param p: dictionary of fields :githublink:`%|py|95` """ dt = datetime.datetime(2009, 1, 1).now() typstr = str if len(l) > 0: def _str_process(s): if isinstance(s, str): return s elif isinstance(s, bytes): return s.decode("utf8") else: try: return str(s) except Exception as e: raise Exception( "unable to convert s into string: type(s)=" + str(type(s))) from e message = str(dt).split( ".")[0] + " " + " ".join([_str_process(s) for s in l]) + "\n" self._handle.write(message) st = " " else: st = typstr(dt).split(".")[0] + " " for k, v in p.items(): message = st + \ "%s = %s%s" % ( typstr(k), typstr(v), "\n") if "INNER JOIN" in message: break self._handle.write(message) self._handle.flush()