원문 캡처 · github.com
원문 캡처 · github.com

BlockGuard는 원격 서버의 요청 제한에 걸렸을 때 재시도를 멈추고, 대기 시간이 지난 뒤 다시 호출하도록 돕는 파이썬 라이브러리다. 자체 호출 속도를 제한하는 도구와 달리, 상대 서버가 429 응답이나 무응답 차단으로 요청을 막는 상황을 다룬다.

pip으로 설치할 수 있다.

bash
pip install blockguard

차단기를 만든 뒤 엔드포인트별 키로 대기 상태를 확인한다. 429 응답을 받으면 trip으로 차단기를 작동시키고, 성공 응답을 받으면 clear로 초기화한다. 기본 대기 시간은 두 배씩 늘어나며 최대 대기 시간으로 제한된다.

python
from blockguard import CircuitBreaker

breaker = CircuitBreaker(base_cooldown_s=60.0, max_cooldown_s=1800.0)

if breaker.in_cooldown("some_api"):
 return cached_result # don't even try -- we already know it'll fail

resp = call_flaky_api()
if resp.status_code == 429:
 breaker.trip("some_api") # next cooldown doubles: 60s, 120s, 240s, ... capped
else:
 breaker.clear("some_api") # back to a clean slate after a real success

컨텍스트 매니저 방식에서는 대기 중인 호출을 건너뛰고 InCooldownError를 발생시킨다. 차단 키는 서로 독립적이므로 앱 전체에 하나를 쓰기보다 논리적 엔드포인트나 요청 제한 단위마다 키를 지정해야 한다.

python
from blockguard import CircuitBreaker, InCooldownError

breaker = CircuitBreaker()

try:
 with breaker.guard("some_api"):
 resp = call_flaky_api()
except InCooldownError as e:
 print(f"skipping call, {e.seconds_remaining:.0f}s left")

테스트에서 모듈이나 앱 범위의 차단기를 사용하면 한 테스트의 대기 상태가 다음 테스트에 남을 수 있다. 저장소는 pytest 플러그인의 track을 사용해 테스트마다 차단기를 초기화하는 방법을 제공한다.

python
from blockguard import CircuitBreaker
from blockguard.pytest_plugin import track

breaker = track(CircuitBreaker()) # now reset between every test, automatically

현재 v0.1은 메모리 기반 단일 프로세스 구현이다. 비동기 API, 호스팅 대시보드, 여러 프로세스가 공유하는 Redis 백엔드는 제공하지 않는다. BlockGuard 저장소에서 자세한 내용을 확인할 수 있다. 라이선스는 MIT다.