Hướng dẫn how do you retry after python? - làm thế nào để bạn thử lại sau khi python?

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

Trê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

Trong 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.

Làm thế nào để bạn thử lại trong Python?

Điều gì xảy ra sau khi ngoại lệ bị bắt Python?

Python có tiếp tục sau khi thử ngoại trừ không?

Điều gì xảy ra nếu một lỗi không bị bắt trong Python?