ライブラリを充実させる
This commit is contained in:
parent
748434e2cf
commit
42eec81c46
1
.gitignore
vendored
1
.gitignore
vendored
@ -1 +1,2 @@
|
||||
venv/
|
||||
__pycache__
|
19
example/debug.py
Normal file
19
example/debug.py
Normal file
@ -0,0 +1,19 @@
|
||||
import sys
|
||||
import os
|
||||
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src")))
|
||||
|
||||
from libs.custom_logger import get_logger
|
||||
from libs.time_watch import TimeWatch
|
||||
|
||||
# level = 10 # DEBUG level
|
||||
logger = get_logger(level=10)
|
||||
logger.debug("Debug message")
|
||||
|
||||
@TimeWatch.debug(logger=logger)
|
||||
def app_run():
|
||||
import time
|
||||
time.sleep(2)
|
||||
return "Function executed."
|
||||
|
||||
if __name__ == "__main__":
|
||||
app_run()
|
11
src/libs/__init__.py
Normal file
11
src/libs/__init__.py
Normal file
@ -0,0 +1,11 @@
|
||||
from .custom_logger import get_logger, CustomLogger
|
||||
from .time_watch import TimeWatch
|
||||
from .singleton import Singleton
|
||||
|
||||
__all__ = [
|
||||
"get_logger",
|
||||
"CustomLogger",
|
||||
"TimeWatch",
|
||||
"Singleton",
|
||||
]
|
||||
|
48
src/libs/custom_logger.py
Normal file
48
src/libs/custom_logger.py
Normal file
@ -0,0 +1,48 @@
|
||||
import logging
|
||||
import functools
|
||||
from .singleton import Singleton
|
||||
|
||||
class CustomLogger(Singleton):
|
||||
def __init__(self, name='main', log_file=None, level=logging.INFO):
|
||||
if hasattr(self, '_initialized') and self._initialized:
|
||||
self.logger.setLevel(level)
|
||||
return # すでに初期化済みなら何もしない
|
||||
|
||||
self.logger = logging.getLogger(name)
|
||||
self.logger.setLevel(level)
|
||||
self.logger.propagate = False
|
||||
|
||||
formatter = logging.Formatter(
|
||||
'%(asctime)s %(levelname)s [%(filename)s:%(lineno)3d]: %(message)s'
|
||||
)
|
||||
|
||||
# Console handler
|
||||
ch = logging.StreamHandler()
|
||||
ch.setFormatter(formatter)
|
||||
self.logger.addHandler(ch)
|
||||
|
||||
# File handler
|
||||
if log_file:
|
||||
fh = logging.FileHandler(log_file, encoding='utf-8')
|
||||
fh.setFormatter(formatter)
|
||||
self.logger.addHandler(fh)
|
||||
|
||||
self._initialized = True
|
||||
|
||||
|
||||
def get_logger(self):
|
||||
return self.logger
|
||||
|
||||
def log_entry_exit(self, func):
|
||||
@functools.wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
self.logger.info(f"Enter: {func.__qualname__}")
|
||||
result = func(*args, **kwargs)
|
||||
self.logger.info(f"Exit: {func.__qualname__}")
|
||||
return result
|
||||
return wrapper
|
||||
|
||||
|
||||
def get_logger(name='main', log_file=None, level=logging.INFO):
|
||||
custom_logger = CustomLogger(name, log_file, level)
|
||||
return custom_logger.get_logger()
|
20
src/libs/singleton.py
Normal file
20
src/libs/singleton.py
Normal file
@ -0,0 +1,20 @@
|
||||
"""Singleton pattern implementation in Python.
|
||||
This implementation is thread-safe and ensures that only one instance of the class is created.
|
||||
|
||||
Singleton が提供するのは「同じインスタンスを返す仕組み」
|
||||
* __init__() は毎回呼ばれる(多くの人が意図しない動作)
|
||||
* __init__の2回目は_initialized というフラグは 使う側で管理する必要がある。
|
||||
"""
|
||||
|
||||
import threading
|
||||
|
||||
class Singleton(object):
|
||||
_instances = {}
|
||||
_lock = threading.Lock()
|
||||
|
||||
def __new__(cls, *args, **kwargs):
|
||||
if cls not in cls._instances:
|
||||
with cls._lock:
|
||||
if cls not in cls._instances: # ダブルチェック
|
||||
cls._instances[cls] = super(Singleton, cls).__new__(cls)
|
||||
return cls._instances[cls]
|
@ -36,20 +36,22 @@ class TimeWatch:
|
||||
self.end_time = None
|
||||
self.logger.debug("Timer reset.")
|
||||
|
||||
@classmethod
|
||||
def debug(cls, func):
|
||||
def wrapper(*args, **kwargs):
|
||||
instance = cls() # インスタンスを作成
|
||||
instance.start()
|
||||
result = func(*args, **kwargs)
|
||||
instance.stop()
|
||||
instance.elapsed_time()
|
||||
return result
|
||||
return wrapper
|
||||
@staticmethod
|
||||
def debug(logger=None):
|
||||
def decorator(func):
|
||||
def wrapper(*args, **kwargs):
|
||||
timer = TimeWatch(logger=logger)
|
||||
timer.start()
|
||||
result = func(*args, **kwargs)
|
||||
timer.stop()
|
||||
timer.elapsed_time()
|
||||
return result
|
||||
return wrapper
|
||||
return decorator
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
@TimeWatch.debug
|
||||
@TimeWatch.debug()
|
||||
def test_function():
|
||||
time.sleep(2)
|
||||
return "Function executed."
|
Loading…
x
Reference in New Issue
Block a user