Python에서 웹훅 처리: FastAPI + Redis

FastAPI와 Redis 큐로 안정적인 웹훅 처리하기

결제 서비스 웹훅은 일반적으로 5초 이내에 응답을 요구합니다. 엔드포인트에서 동기식 처리를 하면 데이터베이스나 외부 서비스의 지연으로 인해 타임아웃이 발생합니다. 단순한 구현 방식은 다음과 같습니다:

@app.post("/webhook")
async def webhook(request: dict):
    update_database(request)
    send_email_to_user(request)
    return {"status": "ok"}

이 접근법은 취약합니다. 처리 중 서버가 다운되면 데이터 손실이 발생하고, 중복 알림은 데이터베이스의 고유 제약 조건 오류를 유발합니다. 해결책은 수신과 처리를 분리하는 작업 대기열 사용입니다.

작업 대기열을 활용한 아키텍처

API 엔드포인트는 웹훅을 수신하고 서명을 검증하며, 중복성 보장을 확인한 후 Redis에 작업을 등록합니다. 별도의 워커가 비동기적으로 비즈니스 로직을 실행합니다. 장점은 다음과 같습니다:

• 즉시 HTTP 응답 (50ms 미만).
• 워커가 다운되더라도 작업이 유지됩니다.
• 수평 확장 가능 (여러 워커가 동시에 실행 가능).

구성 요소:

• API 계층: FastAPI
• 메시지 큐: Redis (LPUSH 및 BRPOP 사용)
• 워커: 격리된 프로세스에서 실행

API 수신기 구현

엔드포인트는 보안과 속도에 집중합니다. HMAC 서명을 검사하고, transaction_id를 통해 중복을 방지하며, 작업을 큐에 추가합니다.

from fastapi import FastAPI, HTTPException, Header
from pydantic import BaseModel
import redis
import json
import hashlib
import os

app = FastAPI()
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
SECRET = os.getenv('WEBHOOK_SECRET', 'test_secret')

def check_signature(body: str, sign: str) -> bool:
    my_sign = hashlib.sha256(f"{body}{SECRET}".encode()).hexdigest()
    return my_sign == sign

@app.post("/api/payment")
async def payment_webhook(request: dict, x_sign: str = Header(None)):
    if not check_signature(json.dumps(request), x_sign):
        raise HTTPException(status_code=403, detail="Bad sign")
    
    task_id = request.get('transaction_id')
    if r.exists(f"processed:{task_id}"):
        return {"status": "duplicate"}
    
    task = {
        "id": task_id,
        "amount": request.get('amount'),
        "user_id": request.get('user_id')
    }
    r.lpush("payment_queue", json.dumps(task))
    
    return {"status": "accepted"}

핵심 실천 사항:

• HMAC 서명으로 위조 요청을 차단합니다.
• Redis 키 processed:{id}를 통해 중복 처리를 방지합니다.
• LPUSH를 사용해 큐에 작업을 효율적으로 추가합니다.

배경 워커를 통한 작업 처리

워커는 루프를 돌며 BRPOP(5초 타임아웃)으로 작업을 가져오고, 로직을 실행한 후 처리 완료 상태를 기록합니다(1일 TTL). 실패 시 지연 후 재시도합니다.

import redis
import json
import time
import logging

logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)

def process_task(data):
    log.info(f"Processing payment {data['id']}")
    time.sleep(1) 
    return True

def main():
    r = redis.Redis(host='localhost', port=6379)
    log.info("Worker started...")
    
    while True:
        task = r.brpop("payment_queue", timeout=5)
        if not task:
            continue
        _, raw_data = task
        data = json.loads(raw_data)
        
        try:
            success = process_task(data)
            if success:
                r.setex(f"processed:{data['id']}", 86400, "1")
                log.info(f"Task {data['id']} completed")
        except Exception as e:
            log.error(f"Error: {e}")
            r.lpush("payment_queue", raw_data)
            time.sleep(5)

if __name__ == "__main__":
    main()

흔한 문제와 해결책

• 서명 검증 누락: 가짜 POST 요청에 취약. 해결: 비밀키를 사용한 HMAC 적용.
• Redis의 불안정성: 재시작 시 데이터 손실. 해결: RDB/AOF 지속성 활성화 또는 PostgreSQL을 큐로 사용.
• 워커 재시도 루프: 작업이 무한 재시도에 갇힘. 해결: 시도 횟수 추적, 딱지 큐(Dead Letter Queue) 사용, 지수 백오프 적용.

기타 리스크:

• 외부 서비스 타임아웃: process_task 내 타임아웃 설정.
• 확장성 문제: 큐 길이(LLEN) 모니터링 및 워커 수 조절.

핵심 교훈

• 중복 방지: processed:{id} 같은 키로 중복 처리를 막습니다.
• 보안: 모든 웹훅 수신 시 HMAC 서명을 반드시 검증하세요.
• 신뢰성: 큐는 실패 상황에서도 작업을 보존합니다.
• 확장성: 독립적인 워커로 병렬 처리가 가능합니다.
• 모니터링: 큐 메트릭과 오류를 로그로 기록해 운영 가능성을 높입니다.

— Editorial Team