BioPlayground

🧬
목록으로

견고한 파이프라인 — 예외·재시도·회로 차단으로 500회 호출을 무사히

500번의 NCBI API 호출을 안전하게. try-except, exponential backoff, circuit breaker를 조합한 견고한 파이프라인 패턴.

중급
|
80분
|
검증 완료 (2026-07)
재시도exponential backoffcircuit breakerbatch APINCBIrobust pipeline예외 처리
진행률0/8 (0%)

견고한 파이프라인 — 예외·재시도·회로 차단으로 500회 호출을 무사히

이 토픽을 마치면

교과서에서 배운 예외, try-except, 오류 처리 를 엮어서, 외부 API를 대량으로 호출하는 파이프라인이 몇 개의 실패에 무너지지 않고 완주하도록 만드는 도구를 직접 만들 수 있습니다. 성공률 95%를 99.9%로 끌어올리는 실전 패턴입니다.

이 글은 교육용 일반 예제 입니다. 실전은 tenacity, backoff, resilience4j 같은 라이브러리를 씁니다.

"다음 날 아침, 로그를 열었다" — 순진한 파이프라인의 함정

여러분이 500개 유전자에 대해 NCBI BLAST를 순차 호출하는 스크립트를 밤새 돌렸다고 합시다.

python
def naive_pipeline(gene_ids: list[str]) -> list[dict]:
    results = []
    for gene_id in gene_ids:
        response = call_blast_api(gene_id)
        results.append(response)
    return results

다음 날 아침 로그를 열어보니 이렇습니다.

text
[00:03] gene 1: OK
[00:05] gene 2: OK
...
[01:23] gene 27: TIMEOUT
Traceback (most recent call last):
  ...
requests.exceptions.Timeout

전체 스크립트가 27번째 호출에서 중단됐고, 나머지 473번은 시도조차 못 했습니다. 하나의 실패가 전체 실행을 죽인 것입니다.

이 접근의 문제:

문제 1: 실패의 격리 없음. 한 번의 오류가 전체를 죽입니다.

문제 2: 재시도 없음. 네트워크 결함은 대개 일시적. 잠깐 기다렸다가 다시 시도하면 성공할 확률이 높습니다.

문제 3: 진행 상황 저장 없음. 성공한 26개 결과가 메모리에 있는 채로 크래시. 다시 처음부터.

진짜 접근은 견고한 파이프라인 패턴입니다. 각 호출의 실패를 격리하고, 재시도하며, 지속 실패 시 회로를 차단합니다. 성공한 결과는 즉시 저장합니다.

블랙박스에서 부품으로

부품 1: try-except로 실패 격리

python
def robust_pipeline_basic(gene_ids: list[str]) -> tuple[list[dict], list[dict]]:
    results = []
    errors = []
    
    for gene_id in gene_ids:
        try:
            response = call_blast_api(gene_id)
            results.append({"gene_id": gene_id, "data": response})
        except Exception as e:
            errors.append({"gene_id": gene_id, "error": type(e).__name__, "message": str(e)})
    
    return results, errors

핵심: 실패한 호출은 errors 로 격리하고, 다음 반복이 계속됩니다. 500개 중 몇 개가 실패해도 나머지가 완주합니다.

주의: except Exception 은 모든 예외를 잡습니다. 이것은 파이프라인 최상위에서만 사용. 특정 오류만 잡고 싶으면 명시적으로.

python
try:
    response = call_blast_api(gene_id)
except requests.exceptions.Timeout:
    # 재시도 가능한 오류
    pass
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 429:
        # rate limit → 재시도 가능
        pass
    else:
        # 다른 4xx → 재시도 무의미
        raise

부품 2: 지수 backoff 재시도

일시적 오류는 재시도로 대부분 해결됩니다. 다만 즉시 재시도는 서버에 부담. 각 재시도 사이 간격을 지수적으로 증가시킵니다.

python
import time
import random




def with_retry(func, max_attempts: int = 5, base_delay: float = 1.0):
    """
    func를 호출하되, 실패 시 exponential backoff로 재시도.
    """
    for attempt in range(1, max_attempts + 1):
        try:
            return func()
        except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
            if attempt == max_attempts:
                raise
            
            delay = base_delay * (2 ** (attempt - 1))
            jitter = random.uniform(0, delay * 0.1)
            time.sleep(delay + jitter)

간격: 1초, 2초, 4초, 8초, 16초. 5번 재시도로 최대 31초 대기.

Jitter(무작위 오차) 를 더하는 이유: 여러 클라이언트가 동시에 실패해 정확히 같은 시간에 재시도하면 서버가 다시 죽습니다. 각자 조금씩 다른 시간에 재시도하도록 랜덤 오차를 더합니다.

부품 3: Circuit Breaker (회로 차단기)

서버가 장기 다운이면 재시도가 무의미. 연속 실패가 임계를 넘으면 잠시 호출을 완전히 정지합니다.

python
from dataclasses import dataclass
from enum import Enum




class CircuitState(Enum):
    CLOSED = "closed"      # 정상 - 호출 허용
    OPEN = "open"          # 차단 - 즉시 실패
    HALF_OPEN = "half_open"  # 회복 확인 - 시험 호출




@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_timeout: float = 60.0
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    last_failure_time: float = 0.0
    
    def call(self, func):
        now = time.time()
        
        if self.state == CircuitState.OPEN:
            if now - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker OPEN")
        
        try:
            result = func()
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
            return result
        except Exception:
            self.failure_count += 1
            self.last_failure_time = now
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
            raise

3가지 상태:

  • CLOSED: 정상. 호출 허용.
  • OPEN: 차단. 즉시 실패 반환 (실제 호출 안 함).
  • HALF_OPEN: 회복 시도. 하나의 호출로 서버 상태 시험.

파이프라인 조립

세 부품을 합칩니다.

python
def robust_pipeline_full(
    gene_ids: list[str],
    output_path: str,
    checkpoint_every: int = 50
) -> dict:
    breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60.0)
    results = []
    errors = []
    
    # 기존 체크포인트 로드
    already_done = load_checkpoint(output_path)
    
    for i, gene_id in enumerate(gene_ids):
        if gene_id in already_done:
            continue
        
        try:
            def call():
                return breaker.call(lambda: call_blast_api(gene_id))
            
            response = with_retry(call, max_attempts=3, base_delay=2.0)
            results.append({"gene_id": gene_id, "data": response})
        except Exception as e:
            errors.append({
                "gene_id": gene_id,
                "error": type(e).__name__,
                "message": str(e)
            })
        
        if (i + 1) % checkpoint_every == 0:
            save_checkpoint(output_path, results)
            print(f"Progress: {i+1}/{len(gene_ids)} — checkpoint saved")
    
    save_checkpoint(output_path, results)
    return {
        "total": len(gene_ids),
        "success": len(results),
        "failed": len(errors),
        "errors": errors
    }




def save_checkpoint(path: str, results: list) -> None:
    import json
    with open(path, "w") as f:
        json.dump(results, f, indent=2)




def load_checkpoint(path: str) -> set[str]:
    import json
    from pathlib import Path
    if not Path(path).exists():
        return set()
    with open(path) as f:
        data = json.load(f)
    return {r["gene_id"] for r in data}

이제 크래시가 있어도 다시 실행하면 이미 처리된 gene은 건너뛰고 나머지만 시도합니다.

페이딩 — 여러분이 채워야 할 세 빈칸

빈칸 1: 예외별 재시도 정책

일부 예외만 재시도해야 합니다. 예를 들어 400 Bad Request는 재시도해도 무의미.

python
def with_retry_smart(func, max_attempts: int = 5):
    retryable = (
        requests.exceptions.Timeout,
        requests.exceptions.ConnectionError,
    )
    
    for attempt in range(1, max_attempts + 1):
        try:
            return func()
        except retryable as e:
            # TODO: 재시도 로직 (backoff + jitter)
            pass
        except requests.exceptions.HTTPError as e:
            # TODO: status_code 확인 후 429 or 5xx면 재시도
            # 그 외는 즉시 raise
            pass

힌트: if e.response.status_code == 429 or 500 <= e.response.status_code < 600: sleep(delay); continue; else: raise.

빈칸 2: 실패 리포트

실패한 gene_ids를 분석해 원인별로 분류.

python
def analyze_failures(errors: list[dict]) -> dict:
    """
    error 리스트를 종류별로 그룹핑.
    반환: {"Timeout": 5, "HTTPError_500": 3, ...}
    """
    from collections import Counter
    
    # TODO: 각 error의 "error" 필드로 카운트
    # HTTPError는 status_code로 세분화
    pass

힌트: counter = Counter(); for e in errors: key = e["error"]; if "status" in e["message"]: key = f"{key}_{parse_status}"; counter[key] += 1.

빈칸 3: 재실행 시 실패 부분만 재시도

체크포인트에 성공/실패 모두 저장하고, 재실행 시 실패한 것만 재시도.

python
def retry_failed_only(
    gene_ids: list[str],
    output_path: str,
    error_log_path: str
) -> None:
    # TODO 1: error_log_path 로드
    # TODO 2: 실패한 gene_ids만 추출
    # TODO 3: robust_pipeline_full 재실행 (필터된 리스트로)
    pass

성찰 — 실전 견고성 도구와의 차이

tenacity 라이브러리: 파이썬 표준 재시도 도구. 데코레이터로 훨씬 깔끔한 문법.

python
from tenacity import retry, stop_after_attempt, wait_exponential


@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, max=60))
def call_api(gene_id):
    return requests.get(f"...").json()

Kubernetes readiness probes: 실전 배포는 애플리케이션 자체가 자기 상태를 보고. 문제 감지 시 트래픽 우회.

Netflix Hystrix / resilience4j: 자바 생태계의 정교한 회로 차단기. Bulkhead, TimeLimiter, RateLimiter 등의 패턴 조합.

Dead Letter Queue: 실패한 요청을 별도 큐에 보관해 나중에 수동/자동 재처리. AWS SQS, RabbitMQ의 표준 기능.

분산 트랜잭션 대안: 실전은 실패해도 복구 가능한 방식(saga pattern, event sourcing)으로 설계. 실패를 예외가 아닌 정상 케이스로.

확장 프로젝트

1. tenacity로 재작성: 여러분의 backoff 로직을 tenacity 데코레이터로 대체.

2. 병렬 처리: concurrent.futures.ThreadPoolExecutor 로 여러 호출을 동시에. 회로 차단기가 여러 스레드 공유.

3. 대시보드: 진행 상황·성공률·회로 상태를 실시간 표시. Streamlit이면 30분.

4. 알림: 실패율이 임계 넘으면 Slack/Telegram으로 알림.

이 편의 부품 지도

  • [F] 예외: 파이썬의 계층적 예외 시스템. Exception 상속, 특정 타입만 잡기.
  • [F] try-except: 실패의 격리와 회복. except ... as e 로 에러 정보 접근.
  • [F] 오류 처리: 재시도 · backoff · 회로 차단의 통합 패턴.
  • [W] 파일 I/O: 체크포인트 저장·로드 (완성 스크립트 제공).

[F] = 여러분이 직접 구현 / [W] = 완성 코드로 제공.