Trang trí là một cách tiếp cận tốt.
from functools import wraps
import time
class retry:
def __init__[self, success=lambda r:True, times=3, delay=1, raiseexception=True, echo=True]:
self.success = success
self.times = times
self.raiseexception = raiseexception
self.echo = echo
self.delay = delay
def retry[fun, *args, success=lambda r:True, times=3, delay=1, raiseexception=True, echo=True, **kwargs]:
ex = Exception[f"{fun} failed."]
r = None
for i in range[times]:
if i > 0:
time.sleep[delay*2**[i-1]]
try:
r = fun[*args, **kwargs]
s = success[r]
except Exception as e:
s = False
ex = e
# raise e
if not s:
continue
return r
else:
if echo:
print[f"{fun} failed.", "args:", args, kwargs, "\nresult: %s"%r]
if raiseexception:
raise ex
def __call__[self, fun]:
@wraps[fun]
def wraper[*args, retry=0, **kwargs]:
retry = retry if retry>0 else self.times
return self.__class__.retry[fun, *args,
success=self.success,
times=retry,
delay=self.delay,
raiseexception = self.raiseexception,
echo = self.echo,
**kwargs]
return wraper
Một số ví dụ sử dụng:
@retry[success=lambda x:x>3, times=4, delay=0.1]
def rf1[x=[]]:
x.append[1]
print[x]
return len[x]
> rf1[]
[1]
[1, 1]
[1, 1, 1]
[1, 1, 1, 1]
4
@retry[success=lambda x:x>3, times=4, delay=0.1]
def rf2[l=[], v=1]:
l.append[v]
print[l]
assert len[l]>4
return len[l]
> rf2[v=2, retry=10] #overwite times=4
[2]
[2, 2]
[2, 2, 2]
[2, 2, 2, 2]
[2, 2, 2, 2, 2]
5
> retry.retry[lambda a,b:a+b, 1, 2, times=2]
3
> retry.retry[lambda a,b:a+b, 1, "2", times=2]
TypeError: unsupported operand type[s] for +: 'int' and 'str'
Trong bài đăng này, chúng tôi sẽ thảo luận về người trang trí thử lại và cách sử dụng nó trong Python.
Nói chung, chúng tôi có thể đã gặp nhiều tình huống mà chúng tôi muốn thực thi một đoạn mã cho đến khi chúng tôi nhận được kết quả mong đợi hoặc chúng tôi có thể cần phải chờ một số hoạt động nên hoàn thành trước khi tiến hành thêm.
Trước khi trực tiếp đến việc thực hiện, tôi sẽ giải thích một trường hợp sử dụng đơn giản trong đó tôi phải sử dụng máy trang trí thử lại.
Trường hợp sử dụng là, có một API với các điểm cuối sau:-
1.api/getData
2.api/status
3.api/data
API/getData, trả về id id thay vì trả về dữ liệu trực tiếp cho mỗi yêu cầu được thực hiện [có một số tham số đầu vào].id” instead of directly returning data for every request made[has some input parameters].
API/status, lấy tham số id id đã được trả về từ API/getData và kiểm tra xem trạng thái của id id đó có thành công hay không [trả về đúng/sai].id” is success or not [returns true/false].
Cho đến khi API/Trạng thái trả về true, chúng tôi không thể thực hiện API/dữ liệu yêu cầu vì dữ liệu chưa được phân tích cú pháp. Vì vậy, API/Trạng thái cho biết nếu chúng ta có thể thực hiện yêu cầu đến API/dữ liệu để lấy dữ liệu chấp nhận tham số ID ID.id” as parameter.
Vì vậy, trong trường hợp này, chúng tôi đã thực hiện các yêu cầu liên tục cho điểm cuối API/Status để kiểm tra trạng thái.
Hãy xem làm thế nào chúng ta có thể đạt được điều này, với và không sử dụng bộ trang trí thử lại.
Không có người trang trí thử lại
def check_status[id]:
status = requests.get[status_url]
if status == True:
return True
else:
return Falsedef get_data[id]:status = None
while status is None or status == False:
time.sleep[5]
status = check_status[id]response = requests.get[url]
return response;
Trong mã giả trên, tôi đang thực hiện cuộc gọi đến phương thức Check Check_status, trong khi vòng lặp và cho mọi yêu cầu, tôi sẽ trì hoãn nó trong 5 giây và nó phải đợi cho đến khi trạng thái là True True, chỉ sau đó chúng ta có thể tìm nạp dữ liệu bằng cách sử dụng api/data điểm cuối.
endpoint.
Vì vậy, đó là một loại mẫu thử lại trong đó chúng ta liên tục kiểm tra trạng thái của ID bằng cách trì hoãn nó trong 5 giây.
Điều tương tự cũng có thể được thực hiện bằng cách sử dụng máy trang trí thử lại trên mạng.
Với người trang trí thử lại
Để cài đặt bộ trang trí thử lại, sử dụng lệnh bên dưới.
pip install retry
Để sử dụng nó trong mã của bạn, bao gồm câu lệnh bên dưới trong mã của bạn.
from retry import retry
Để thực hiện bất kỳ chức năng nào như một logic thử lại, chỉ cần thêm câu lệnh @retry [] trên định nghĩa chức năng của bạn như dưới đây.
@retry[]
def check_status:
Người trang trí thử lại chấp nhận một số thông số,
@retry[success=lambda x:x>3, times=4, delay=0.1]
def rf1[x=[]]:
x.append[1]
print[x]
return len[x]
> rf1[]
[1]
[1, 1]
[1, 1, 1]
[1, 1, 1, 1]
4
0Trên đây là một số thông số mà người trang trí thử lại chấp nhận, để biết thêm thông tin tham khảo tài liệu.
Implementation:
Thay vì sử dụng thời gian.s ngủ [5], tôi đã sử dụng trình trang trí thử lại được khai báo ở trên chức năng của Wait Wait_Sfor_Success_Status với 3 loại ngoại lệ cho biết loại chức năng ngoại lệ cụ thể đó phải thử lại, hãy trì hoãn với 5 giây làm chậm quá trình thực thi chức năng bằng cách thực thi chức năng bằng cách thực thi chức năng bằng cách thực thi chức năng bằng cách thực hiện chức năng bằng cách 5 giây, thử với giá trị 6 cho thấy chức năng có thể được thử lại tối đa là 6 lần.wait_for_success_status” function with 3 parameters Exception type which indicates that on that specific type of exception function has to retry , delay with 5 seconds which delays the function execution by 5 seconds , tries with value 6 which indicates that function can be retried maximum of 6 times.
Chúng tôi cũng có thể thêm nhiều loại ngoại lệ.
@retry[success=lambda x:x>3, times=4, delay=0.1]
def rf1[x=[]]:
x.append[1]
print[x]
return len[x]
> rf1[]
[1]
[1, 1]
[1, 1, 1]
[1, 1, 1, 1]
4
1Trong trường hợp này, chức năng sẽ được thử lại nếu có một ngoại lệ xảy ra.