Pythonアプリケーションのミニマルなひな型

研究で使っているスクリプトのシステム周りの実装がほぼテンプレート化してきたので載せてみる。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from abc import ABCMeta, abstractmethod
import argparse
from collections.abc import Callable
from datetime import datetime
from dateutil.tz import tzlocal
import logging
import os
import signal
import sys
import time
import uuid
class Singleton(metaclass=ABCMeta):
__instance = None
def __new__(cls, *args, **kwargs):
if cls.__instance is None:
cls.__instance = super(Singleton, cls).__new__(cls, *args, **kwargs)
return cls.__instance
def _gen_hexid():
return uuid.uuid4().hex[:6]
class _Formatter(logging.Formatter):
def formatTime(self, record, datefmt=None):
ct = self.converter(record.created)
if datefmt:
t = datefmt.replace('%f', str(int(record.msecs)))
s = time.strftime(t, ct)
else:
t = time.strftime(self.default_time_format, ct)
s = self.default_msec_format % (t, record.msecs)
return s
class Logger:
__instance = None
FATAL = logging.CRITICAL
ERROR = logging.ERROR
WARNING = logging.WARNING
INFO = logging.INFO
DEBUG = logging.DEBUG
VERBOSE = 5
LABELS = {
ERROR: 'error',
WARNING: 'warn',
INFO: 'info',
DEBUG: 'debug',
VERBOSE: 'trace',
}
# global config
_verbose = False
_logdir = './logs'
_loglevel = INFO
_format = "%(asctime)-15s\t%(accessid)s\t[%(priority)s]\t%(message)s"
_dateformat = "%Y-%m-%d %H:%M:%S.%f %Z"
def __init__(self):
raise NotImplementedError()
@classmethod
def finalize(cls):
try:
cls._get_instance()._stop()
except FileNotFoundError:
pass
finally:
Logger.__instance = None
def _initialize(self):
if hasattr(self, '_initialized') and self._initialized:
return
self._logger = logging.getLogger(__name__)
self._logger.setLevel(Logger.VERBOSE)
logdir = os.path.abspath(os.path.expanduser(Logger._logdir))
if not os.path.isdir(logdir):
raise FileNotFoundError("logdir was not found: '%s'" % logdir)
logfile = logdir + '/' + datetime.now().strftime("%Y%m%d") + '.log'
file_handler = logging.FileHandler(filename=logfile)
file_handler.setLevel(Logger._loglevel)
file_handler.setFormatter(_Formatter(Logger._format, Logger._dateformat))
self._add_handler(file_handler)
if Logger._verbose:
stream_handler = logging.StreamHandler()
stream_handler.setLevel(Logger.VERBOSE)
stream_handler.setFormatter(_Formatter(Logger._format, Logger._dateformat))
self._add_handler(stream_handler)
self._start()
self._initialized = True
def _add_handler(self, hdlr):
self._logger.addHandler(hdlr)
def _remove_handler(self, hdlr):
hdlr.close()
self._logger.removeHandler(hdlr)
@classmethod
def _get_instance(cls):
if cls.__instance is None:
instance = object.__new__(cls)
instance._initialize()
cls.__instance = instance
return cls.__instance
@classmethod
def configure(cls, loglevel=_loglevel, verbose=_verbose, logdir=_logdir):
cls._loglevel = loglevel
cls._verbose = verbose
cls._logdir = logdir
cls._get_instance()
def _start(self):
now = datetime.now(tzlocal())
self._accessid = _gen_hexid()
self._uniqueid = "UNIQID"
self._accesssec = now
self._accesstime = now.strftime(Logger._dateformat)
message = "LOG Start with ACCESSID=[%s] UNIQUEID=[%s] ACCESSTIME=[%s]"
self._log(Logger.INFO, message % (self._accessid, self._uniqueid, self._accesstime))
def _stop(self):
processtime = '%3.9f' % (datetime.now(tzlocal()) - self._accesssec).total_seconds()
message = "LOG End with ACCESSID=[%s] UNIQUEID=[%s] ACCESSTIME=[%s] PROCESSTIME=[%s]\n"
self._log(Logger.INFO, message % (self._accessid, self._uniqueid, self._accesstime, processtime))
while len(self._logger.handlers) > 0:
self._remove_handler(self._logger.handlers[0])
def _log(self, level, message, exc_info=False, stack_info=False):
extras = {
'accessid': self._accessid,
'priority': Logger.LABELS[level]
}
self._logger.log(level, message, extra=extras, exc_info=exc_info, stack_info=stack_info)
@classmethod
def e(cls, message):
cls._get_instance()._log(Logger.ERROR, message, exc_info=True, stack_info=True)
@classmethod
def w(cls, message):
cls._get_instance()._log(Logger.WARNING, message)
@classmethod
def i(cls, message):
cls._get_instance()._log(Logger.INFO, message)
@classmethod
def d(cls, message):
cls._get_instance()._log(Logger.DEBUG, message)
@classmethod
def v(cls, message):
cls._get_instance()._log(Logger.VERBOSE, message)
class App(Singleton, Callable):
_basedir = os.path.dirname(os.path.realpath(__file__))
_logdir = _basedir + '/logs'
_loglevel = Logger.DEBUG
_verbose = True
_debug = True
__defined_args = []
@abstractmethod
def main(self):
raise NotImplementedError()
@classmethod
def configure(
cls,
logdir=_logdir,
loglevel=_loglevel,
verbose=_verbose,
debug=_debug):
cls._loglevel = loglevel
cls._verbose = verbose
cls._logdir = logdir
cls._debug = debug
@classmethod
def exec(cls):
app = cls()
app()
@classmethod
def _def_arg(cls, *args, **kwargs):
cls.__defined_args.append((args, kwargs))
def _initialize(self):
pass
def __initialize(self):
self._initialize()
self._def_arg('--debug', type=str, default=self._debug,
help='Enable debug mode')
self._def_arg('--logdir', type=str, default=self._logdir,
help='Log directory')
self._def_arg('--silent', '--quiet', action='store_true', default=not(self._verbose),
help='Silent execution: does not print any message')
parser = argparse.ArgumentParser()
[parser.add_argument(*_args, **_kwargs) for (_args, _kwargs) in self.__defined_args]
args = parser.parse_args()
self.configure(args.logdir, Logger.DEBUG if args.debug else Logger.INFO, not(args.silent), args.debug)
Logger.configure(loglevel=self._loglevel, verbose=self._verbose, logdir=self._logdir)
if not self._verbose:
sys.stdout = sys.stderr = open(os.devnull, 'w')
self._args = args
def __call__(self):
self.__initialize()
try:
def handler(signum, frame):
raise SystemExit("Signal(%d) received: The program %s will be closed" % (signum, __file__))
signal.signal(signal.SIGINT, handler)
signal.signal(signal.SIGTERM, handler)
self.main()
except Exception:
Logger.e("Exception occurred during execution:")
except SystemExit as e:
Logger.w(e)
finally:
Logger.finalize()
sys.exit(0)
if __name__ == "__main__":
Log = Logger
class MyApp(App):
def _initialize(self):
self._def_arg('--iter', '-i', type=int, default=10,
help='Number of iteration')
def main(self):
for i in range(self._args.iter):
Log.i("Hello World!")
MyApp.exec()

https://raw.githubusercontent.com/chantera/pyapp/master/base.py

LINEで送る
Pocket

コメントを残す

*