웹훅과 202 Accepted — 호출 방향을 뒤집고 작업을 분리하기

# Server 쪽 — HMAC signature 검증 가진 webhook 받기
import hashlib, hmac
from fastapi import FastAPI, Request, Header, HTTPException, status

app = FastAPI()
WEBHOOK_SECRET = b'shared-secret-from-env'
_seen_event_ids: set[str] = set()  # production 엔 TTL 가진 Redis

@app.post('/webhooks/stripe')
async def stripe_webhook(
    request: Request,
    stripe_signature: str = Header(None, alias='Stripe-Signature'),
):
    raw_body = await request.body()

    # 1. Signature 검증
    expected = hmac.new(WEBHOOK_SECRET, raw_body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(stripe_signature.split('=')[-1], expected):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail='잘못된 signature')

    payload = await request.json()
    event_id = payload.get('id')

    # 2. event_id 로 dedupe (retry 가 같은 id 전달)
    if event_id in _seen_event_ids:
        return {'received': True, 'duplicate': True}
    _seen_event_ids.add(event_id)

    # 3. 작업 enqueue; 2xx 빨리 돌려줌
    enqueue_payment_processed(payload['data']['object'])
    return {'received': True}

# 202 + Location — long-running operation 위한 async REST 패턴
import asyncio, uuid
from fastapi import FastAPI, BackgroundTasks, Response, status, HTTPException

app = FastAPI()
_JOBS: dict[str, dict] = {}

async def do_long_work(job_id: str, payload: dict):
    _JOBS[job_id]['status'] = 'running'
    await asyncio.sleep(30)  # 실제 작업 상상
    _JOBS[job_id]['status'] = 'done'
    _JOBS[job_id]['result'] = {'processed': True, **payload}

@app.post('/jobs', status_code=status.HTTP_202_ACCEPTED)
async def start_job(payload: dict, response: Response, bg: BackgroundTasks):
    job_id = str(uuid.uuid4())
    _JOBS[job_id] = {'status': 'queued', 'created_at': 'now'}
    bg.add_task(do_long_work, job_id, payload)
    response.headers['Location'] = f'/jobs/{job_id}'
    return {'job_id': job_id, 'status': 'queued'}

@app.get('/jobs/{job_id}')
async def get_job(job_id: str):
    job = _JOBS.get(job_id)
    if not job:
        raise HTTPException(status.HTTP_404_NOT_FOUND)
    return job

# Client — 완료까지 202 status URL poll
import httpx, time

resp = httpx.post('https://api.example.com/jobs', json={'do': 'something'})
assert resp.status_code == 202
job_url = resp.headers['Location']
print(f'job 시작: {job_url}')

# 완료나 실패까지 poll
while True:
    poll = httpx.get(f'https://api.example.com{job_url}')
    status_val = poll.json()['status']
    print(f'status: {status_val}')
    if status_val == 'done':
        print('result:', poll.json()['result'])
        break
    if status_val == 'failed':
        raise RuntimeError(poll.json().get('error'))
    time.sleep(2)  # 정중한 polling interval