[python] tenacity 라이브러리를 이용해 코드 재실행하는 방법

들어가며

 

파이썬으로 운영을 위한 프로그램을 만들었습니다. 프로그램을 동작시키던 도중 예기치 못한 오류가 발생하여 재실행이 필요한 때는 언제든 발생할 수 있죠. 실제 운영 프로그램을 설계하면 API라던지 DB라던지 파이썬의 메인 프로그램과 다른 여러 환경들이 얽혀있다는 점을 잘 아실 겁니다. 이렇게 파이썬 밖에 있는 환경이 얽혀있을 때는 특히 통신 오류가 발생할 수 있는데요. 정확한 원인을 알 수 없는 오류가 발생하여 잠시 대기한 후에 재실행을 하면 다시 잘 되는 경우도 있죠. 이런 경우를 대비하기 위해 Tenacity 라이브러리를 적용하여 쉽게 코드를 재실행 할 수 있습니다.

 

 

 

Tenacity 사용하기

 

Tenacity는 예외가 발생하는 경우에 다시 함수를 실행시켜서 사용자가 원하는 결과를 받고 안정적으로 운영을 할 수 있게 해줍니다. Tenacity의 뜻은 우리말로 끈기, 집념이라고 하네요. 값을 반환할 때까지 끈기 있게 시도한다는 의미에서 그렇게 이름이 붙었나봅니다.🤣

 

먼저 아래 코드로 라이브러리를 설치합니다. 저는 파이썬은 3.8.10, tenacity는 8.2.3 버전을 사용했습니다.

 

pip install tenacity

 

반복문과 time.sleep을 이용하여 대기시간을 주고 함수를 재실행하는 것보다 좀 더 다양한 옵션을 줄 수 있으니 아래 간단한 예시들을 적용해보겠습니다.

 

 

Tenacity를 적용하지 않은 경우

 

예시를 위해 랜덤한 인덱스를 이용해 리스트에서 값을 얻는 함수를 작성하였습니다. 먼저 tenacity를 적용하지 않았을 때의 결과입니다. 랜덤 인덱스 4는 리스트의 범위를 벗어나기 때문에 인덱스 에러가 발생하였습니다.

 

import numpy as np
from datetime import datetime
import traceback

def get_value(lst):
    idx = np.random.randint(len(lst), len(lst) + 5)
    print(idx, datetime.now())
    value = lst[idx]
    return value

if __name__ == '__main__':
    try:
        lst = ['a', 'b', 'c', 'd']
        get_value(lst)
    except:
        print(traceback.format_exc())

4 2024-04-13 18:53:57.174375
Traceback (most recent call last):
  File "tenacity_ex.py", line 17, in <module>
    get_value(lst)
  File "tenacity_ex.py", line 10, in get_value
    value = lst[idx]
IndexError: list index out of range

 

 

재실행 횟수 설정하기

 

tenacity를 사용하기 위해서는 함수 위에 retry 데코레이터를 적용하면 됩니다. 에러가 발생했을 때 재실행을 적용하기 위해 stop 파라미터에 stop_after_attempt 함수를 넣고 횟수를 설정했습니다. 이제 에러가 발생할 것을 대비해 최대 3번 함수를 실행하게 됩니다.

 

import tenacity as t
import numpy as np
from datetime import datetime
import traceback

@t.retry(
    stop=t.stop_after_attempt(3)
)
def get_value(lst):
    idx = np.random.randint(len(lst), len(lst)+5)
    print(idx, datetime.now())
    value = lst[idx]
    return value

if __name__ == '__main__':
    try:
        lst = ['a', 'b', 'c', 'd']
        get_value(lst)
    except:
        print(traceback.format_exc())

 

 

대기 시간 설정하기 1: 정해진 시간만큼

 

에러가 발생했을 때 정해진 시간(초) 만큼 대기한 후에 함수를 재실행하도록 설정할 수 있습니다. wait 파라미터에 wait_fixed 함수에 초 단위 값을 입력하여 간단하게 설정할 수 있습니다.

 

@t.retry(
    stop=t.stop_after_attempt(3),
    wait=t.wait_fixed(5),
)
def get_value(lst):
    idx = np.random.randint(len(lst), len(lst)+5)
    print(idx, datetime.now())
    value = lst[idx]
    return value

 

시간을 보시면 함수 실행 실패 후 5초 이후에 코드가 재수행된 것을 확인할 수 있습니다.

 

 

대기 시간 설정하기 2: 점점 대기 시간이 증가하도록

 

고정된 시간만큼 대기하는 것이 아니라, 대기 시간이 점점 증가하도록 설정할 수도 있습니다. 통신에 어떤 문제가 생겨서 오랫동안 대기해야 하는 경우에는 대기 시간을 고정시켜 놓기 보다는 점차 늘어나도록 설정하는 것이 합리적일 수도 있습니다. 아래는 대기 시간이 10, 20, 40, 80초로 늘어나며 최대 대기 시간은 100초로 설정하여 계속 재수행을 시도하는 예시입니다.

 

@t.retry(
    wait=t.wait_exponential(multiplier=10, max=100),
)
def get_value(lst):
    idx = np.random.randint(len(lst), len(lst)+5)
    print(idx, datetime.now())
    value = lst[idx]
    return value

 

 

특정 에러가 발생하는 경우 재실행 시키기

 

함수를 동작시키던 중 여러 에러가 발생할 수 있죠. 특정 에러가 발생했을 때에만 재실행을 유도할 수도 있습니다. 아래는 retry 파라미터에 에러 조건을 설정하여 해당 에러가 발생한 경우에 재실행을 하도록 설정하는 예시입니다. 첫 시도에서 IndexError가 발생하여 재실행을 하였고, 두 번째 시도에서 성공하였습니다.

 

@t.retry(
    stop=t.stop_after_attempt(3),
    retry=t.retry_if_exception_type(IndexError)
)
def get_value(lst):
    idx = np.random.randint(len(lst)+5)
    print(idx, datetime.now())
    value = lst[idx]
    return value

 

 

 

마치며

 

개발 시 여러 상황을 대비하여 try, if 와 같은 코드를 작성하지만 항상 예상하지 못한 상황은 발생할 수 있죠. 특히 통신 작업은 한번 에러가 발생하면 정성껏 예측했던 결과가 사라져버릴 수 있기 때문에 여러 차례 시도를 해야 합니다. 라이브러리 이름처럼 성공값을 반환받을 수 있도록, 집념 있는 재수행을 하는 옵션이 필요하다면 tenacity 를 적극 활용해보길 바랍니다.

 

 

 

참고

 

https://tenacity.readthedocs.io/en/latest/index.html

Tenacity — Tenacity documentation

 

https://dojang.io/mod/page/view.php?id=2427)

파이썬 코딩 도장: 42.1 데코레이터 만들기