59 lines
1.8 KiB
Python
59 lines
1.8 KiB
Python
# -*- encoding: utf-8 -*-
|
|
from datetime import datetime, timedelta
|
|
import functools
|
|
import logging
|
|
import sys
|
|
import typing as T
|
|
|
|
|
|
def logit(logger: logging.Logger) -> T.Callable[..., T.Any]:
|
|
def wrapper(func: T.Callable[..., T.Any]) -> T.Callable[..., T.Any]:
|
|
@functools.wraps(func)
|
|
def inner(*args, **kwargs):
|
|
try:
|
|
value = func(*args, **kwargs)
|
|
logger.info(f"Success: {func.__name__}")
|
|
logger.debug(f"{func.__name__}(args={args}, kwargs={kwargs})")
|
|
return value
|
|
except Exception as e:
|
|
logger.error(f"Failure: {e}")
|
|
logger.debug(f"{func.__name__}(args={args}, kwargs={kwargs})")
|
|
sys.exit(1)
|
|
|
|
return inner
|
|
|
|
return wrapper
|
|
|
|
|
|
def retry_and_log(logger: logging.Logger, retries: int):
|
|
"""
|
|
This decorator wraps a function, logs at info in case of success and logs
|
|
error in case an exception is raised.
|
|
"""
|
|
|
|
def wrapper(func: T.Callable[..., T.Any]) -> T.Callable[..., T.Any]:
|
|
@functools.wraps(func)
|
|
def inner(*args, **kwargs):
|
|
for i in range(retries):
|
|
try:
|
|
value = func(*args, **kwargs)
|
|
logger.info(f"Success ({i+1} tentative): {func.__name__}")
|
|
logger.debug(f"{func.__name__}(args={args}, kwargs={kwargs})")
|
|
return value
|
|
except Exception as e:
|
|
logger.error(f"Failure ({i+1} tentative): {e}")
|
|
logger.debug(f"{func.__name__}(args={args}, kwargs={kwargs})")
|
|
|
|
sys.exit(2)
|
|
|
|
return inner
|
|
|
|
return wrapper
|
|
|
|
|
|
def get_date() -> T.Text:
|
|
"""
|
|
Get yesterday's date in %Y-%m-%d format.
|
|
"""
|
|
return (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
|