1、一些装饰器,可以减少重复编写。比较常用的。
用的时候函数上面加上装饰器就可以。这是一些装饰器,加在函数或者方法上,减少了很多重复代码。
除此之外工作中也用一些mixin类大幅减少代码。
import sys import traceback from functools import wraps import threading import time import unittest from app.utils_ydf import LogManager from tomorrow3 import threads as tomorrow_threads from functools import lru_cache # NOQA handle_exception_log = LogManager('function_error').get_logger_and_add_handlers() run_times_log = LogManager('run_many_times').get_logger_and_add_handlers() class CustomException(Exception): def __init__(self, err=''): err0 = 'fatal exception ' Exception.__init__(self, err0 + err) def run_many_times(times=1): """把函数运行times次的装饰器 :param times:运行次数 没有捕获错误,出错误就中断运行,可以配合handle_exception装饰器不管是否错误都运行n次。 """ def _run_many_times(func): @wraps(func) def __run_many_times(*args, **kwargs): for i in range(times): LogManager('run_many_times').get_logger_without_handlers().debug('* ' * 50 + '当前是第 {} 次运行[ {} ]函数'.format(i + 1, func.__name__)) func(*args, **kwargs) return __run_many_times return _run_many_times def handle_exception(retry_times=0, error_detail_level=0, is_throw_error=False): """捕获函数错误的装饰器,重试并打印日志 :param retry_times : 重试次数 :param error_detail_level :为0打印exception提示,为1打印3层深度的错误堆栈,为2打印所有深度层次的错误堆栈 :param is_throw_error : 在达到最大次数时候是否重新抛出错误 :type error_detail_level: int """ if error_detail_level not in [0, 1, 2]: raise Exception('error_detail_level参数必须设置为0 、1 、2') def _handle_exception(func): @wraps(func) def __handle_exception(*args, **keyargs): for i in range(0, retry_times + 1): try: result = func(*args, **keyargs) if i: LogManager('function_error').get_logger_without_handlers().debug( u'%s 调用成功,调用方法--> [ %s ] 第 %s 次重试成功' % ('# ' * 40, func.__name__, i)) return result except Exception as e: error_info = '' if error_detail_level == 0: error_info = str(e) elif error_detail_level == 1: error_info = traceback.format_exc(limit=3) elif error_detail_level == 2: error_info = traceback.format_exc() LogManager('function_error').get_logger_without_handlers().error( u'%s 记录错误日志,调用方法--> [ %s ] 第 %s 次错误重试,错误原因是: %s ' % ('- ' * 40, func.__name__, i, error_info)) if i == retry_times and is_throw_error: # 达到最大错误次数后,重新抛出错误 raise e return __handle_exception return _handle_exception def keep_circulating(time_sleep=0.001): """间隔一段时间,一直循环运行某个方法的装饰器 :param time_sleep :循环的间隔时间 """ if not hasattr(keep_circulating, 'keep_circulating_log'): keep_circulating.log = LogManager('keep_circulating').get_logger_and_add_handlers() def _keep_circulating(func): @wraps(func) def __keep_circulating(*args, **kwargs): while 1: try: func(*args, **kwargs) except Exception: msg = func.__name__ + ' 运行出错 ' + traceback.format_exc(limit=2) keep_circulating.log.error(msg) finally: time.sleep(time_sleep) return __keep_circulating return _keep_circulating def singleton(cls): """单例模式装饰器 """ _instance = {} @wraps(cls) def _singleton(*args, **kwargs): if cls not in _instance: _instance[cls] = cls(*args, **kwargs) return _instance[cls] return _singleton def timer(func): """计时器装饰器,只能用来计算函数运行时间""" if not hasattr(timer, 'log'): timer.log = LogManager('timer').get_logger_and_add_handlers() @wraps(func) def _timer(*args, **kwargs): t1 = time.time() result = func(*args, **kwargs) t2 = time.time() t_spend = t2 - t1 timer.log.debug('执行[ {} ]方法用时 {} 秒'.format(func.__name__, t_spend)) return result return _timer class TimerContextManager(object): """ 用上下文管理器计时,可对代码片段计时 """ log = LogManager('TimerContext').get_logger_and_add_handlers() def __init__(self, is_print_log=True): self._is_print_log = is_print_log self.t_spend = None self._line = None self._file_name = None self.time_start = None def __enter__(self): self._line = sys._getframe().f_back.f_lineno # 调用此方法的代码的函数 self._file_name = sys._getframe(1).f_code.co_filename # 哪个文件调了用此方法 self.time_start = time.time() return self def __exit__(self, exc_type, exc_val, exc_tb): self.t_spend = time.time() - self.time_start if self._is_print_log: self.log.debug(f'对下面代码片段进行计时: 执行"{self._file_name}:{self._line}" 用时 {self.t_spend} 秒') class ExceptionContextManager(): """ 用上下文管理器捕获异常,可对代码片段进行错误捕捉,比装饰器更细腻 """ def __init__(self, logger_name='ExceptionContextManager', verbose=100, donot_raise__exception=True, ): """ :param _verbose: 打印错误的深度,对应traceback对象的limit,为正整数 :param donot_raise__exception:是否不重新抛出错误,为Fasle则抛出,为True则不抛出 """ self.logger = LogManager(logger_name).get_logger_and_add_handlers() self._verbose = verbose self._donot_raise__exception = donot_raise__exception def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): # print(exc_val) # print(traceback.format_exc()) exc_str = str(exc_type) + ' : ' + str(exc_val) exc_str_color = '