python 裝飾器實踐，實現定時函數和失敗異常重復調用

執行請求或函數，出現異常情況下指定重復執行次數
可以作為一個包調用
方法 get()和post 做請求，execcunt = 指定請求失敗再次請求次數
方法loopExecution作為裝飾器，循環執行函數，execcunt = 指定異常再次執行次數

#函數異常情況下再次執行，執行次數

import traceback def loopExecution(func):def wrapper(*args,**kwargs):count = kwargs.pop("execcount", 1)try:res = func(*args,**kwargs)except Exception as err:for i in range(count):try:res = func(*args,**kwargs)if isinstance(res,Exception):return resexcept Exception as err:pass#失敗到達限定值返回異常信息return traceback.format_exc()# return Nonereturn resreturn wrapper@loopExecution def func():print("我是一個函數")raise Exception("執行失敗了")if __name__ == '__main__':data = func(execcount=6)print("多次失敗情況下返回的數據：",data)

通用請求，指定請求次數，將其作為一個包調用，如包名為my_request
import my_reqeust
response = my_request.get(url)

import time import requests import traceback from requests import Session from functools import partialsession = Session() traceback_info = Nonedef outer(func):def wrapper(*args,**kwargs):count = kwargs.pop("execcount", 1)res = func(*args,**kwargs)if isinstance(res, Exception):for i in range(count):print(i)res = func(*args,**kwargs)if not isinstance(res, Exception):breakif isinstance(res, Exception):return Nonereturn resreturn wrapper@outer def publicRequsts(*args,**kwargs):"""通用請求方法"""method = args[0] if len(args) is 1 else kwargs.pop("method", "GET")if not method in ("GET","POST"):raise Exception("The request method is illegal")url = args[1] if len(args) is 2 else kwargs.pop("url", None)if url is None or not isinstance(url,str) or not "http" in url:raise Exception("The URL is illegal")try:response = session.request(method, url,**kwargs)print("="*10)except Exception as err:global traceback_infotraceback_info = traceback.format_exc()return errreturn responseget = partial(publicRequsts, "GET") post = partial(publicRequsts, "POST")

使用線程對執行函數限制執行時間，可不加定時為非阻塞，加定時間到會強制停止線程

from threading import Thread import threading import inspect import ctypesdef _async_raise(tid, exctype):"""raises the exception, performs cleanup if needed"""tid = ctypes.c_long(tid)if not inspect.isclass(exctype):exctype = type(exctype)res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))if res == 0:raise ValueError("invalid thread id")elif res != 1:# """if it returns a number greater than one, you're in trouble,# and you should call it again with exc=NULL to revert the effect"""ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)raise SystemError("PyThreadState_SetAsyncExc failed")def stop_thread(thread):_async_raise(thread.ident, SystemExit)def timelimited(exectime=None):def decorator(function):def decorator2(*args, **kwargs):time_out = exectime if not exectime is None else kwargs.pop("exectime",None)if time_out is None:return Exception("exec time param missing ")class TimeLimited(threading.Thread):def __init__(self, _error=None, ):Thread.__init__(self)self._error = _errorself._result = Nonedef run(self):try:result = function(*args, **kwargs)if result is None:self._result = Trueelse:self._result = resultexcept Exception as err:self._error = Exception(err)t = TimeLimited()t.setDaemon(True)t.start()if not time_out is False:t.join(time_out)if isinstance(t._error, Exception):return t._errorelse:if t._result is None:stop_thread(t)return Exception("Time out!")else:return t._resultreturn decorator2return decorator@timelimited(exectime=3)# 設置運行超時時間S，優先使用，exectime=False是不加定時 def func():time.sleep(5)print("==========")print("==========")print("==========")print("==========")return "aaaa"if __name__ == "__main__":print(func())time.sleep(4)print(threading.enumerate())time.sleep(6)

總結

以上是生活随笔為你收集整理的python 装饰器实践，实现定时函数和失败异常重复调用的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。