[FastAPI] Background Tasks

FastAPI Background Tasks, FastAPI BackgroundTasks, FastAPI Celery, FastAPI Arq, FastAPI RQ, FastAPI 백그라운드 작업

정의

FastAPI 에서 응답 반환 후 실행하는 짧은 작업 은 BackgroundTasks, 장기/신뢰성 있는 작업 은 외부 task queue (Celery, Arq, RQ) 로 오프로드합니다. 두 카테고리를 헷갈리면 데이터 유실, 성능 문제가 발생합니다.

BackgroundTasks (in-process)

응답을 즉시 반환하고, 이어서 같은 프로세스에서 백그라운드 작업 실행.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def send_email(email: str, subject: str, body: str):
    smtp.send(email, subject, body)   # 몇 초 걸리는 blocking 호출

@app.post("/subscribe")
def subscribe(email: str, background_tasks: BackgroundTasks):
    save_subscription(email)
    background_tasks.add_task(send_email, email, "Welcome", "Hi!")
    return {"status": "subscribed"}

동작:

1. save_subscription 실행
2. 응답 (200) 즉시 반환
3. 이후 이벤트 루프가 send_email 실행

async / sync 모두 지원

async def async_task(x: int):
    await asyncio.sleep(1)
    print(x)

background_tasks.add_task(async_task, 42)

Sync 함수는 스레드풀에서, async 함수는 이벤트 루프에서 실행.

Dependency 와 조합

def get_notifier() -> Notifier:
    return Notifier()

def send_notification(notifier: Notifier, user_id: int, message: str):
    notifier.send(user_id, message)

@app.post("/notify")
def notify(
    user_id: int,
    message: str,
    background_tasks: BackgroundTasks,
    notifier: Annotated[Notifier, Depends(get_notifier)],
):
    background_tasks.add_task(send_notification, notifier, user_id, message)
    return {"queued": True}

주의: dependency 로 얻은 리소스 (DB 세션 등) 는 응답 시점에 이미 닫힘. Background task 안에서 재사용 불가. 새 세션을 명시적으로 열어야 함.

BackgroundTasks 의 한계

BackgroundTasks 는 간단하고 실패해도 큰 문제 없는 작업 에만 씁니다.

• 동일 프로세스: worker 프로세스가 죽으면 태스크 잃음
• 재시도 없음: 실패해도 그냥 로그
• 큐 상태 관찰 불가: 백로그, 실패 카운트 없음
• 분산 없음: worker 확장 불가
• 응답 시 이벤트 루프 사용: async 태스크는 이벤트 루프 부담

적합: 이메일 발송 시도 (실패해도 재시도 없어도 됨), 캐시 warm, 로그 전송. 부적합: 결제, 회계, 신뢰성 요구 작업. 이 경우 반드시 외부 큐.

외부 Task Queue

Celery (전통적)

가장 널리 쓰이는 Python task queue.

# tasks.py
from celery import Celery

celery_app = Celery("myapp", broker="redis://localhost:6379/0", backend="redis://localhost:6379/1")

@celery_app.task(bind=True, max_retries=3)
def process_payment(self, order_id: int):
    try:
        payment_service.charge(order_id)
    except TemporaryError as exc:
        raise self.retry(exc=exc, countdown=60)

# main.py
from tasks import process_payment

@app.post("/checkout")
def checkout(order: Order):
    save_order(order)
    process_payment.delay(order.id)   # Celery 워커가 처리
    return {"order_id": order.id, "status": "pending"}

실행:

celery -A tasks worker --loglevel=info --concurrency=4
celery -A tasks beat --loglevel=info      # 스케줄러
celery -A tasks flower                    # 모니터링 UI

강점: 성숙, 방대한 기능 (retry, chain, group, chord, periodic tasks, …), 방대한 문서 약점: async 지원 미흡 (sync worker 위주), 무거움, 설정 복잡

Arq (async 지향)

Redis 백엔드, asyncio native.

# worker.py
from arq import create_pool
from arq.connections import RedisSettings

async def send_email(ctx, email: str, subject: str):
    async with aiohttp.ClientSession() as session:
        await session.post("https://mail-api/send", json={"to": email, "subject": subject})

class WorkerSettings:
    functions = [send_email]
    redis_settings = RedisSettings(host="localhost")

# main.py
from arq import create_pool

redis = None

@app.on_event("startup")
async def startup():
    global redis
    redis = await create_pool(RedisSettings())

@app.post("/subscribe")
async def subscribe(email: str):
    await redis.enqueue_job("send_email", email, "Welcome")
    return {"queued": True}

실행:

arq worker.WorkerSettings

강점: async-first, 가볍고 빠름, FastAPI 와 자연스러운 통합 약점: Celery 대비 기능/생태계 좁음

RQ (Redis Queue)

가장 단순.

from rq import Queue
from redis import Redis

redis_conn = Redis()
queue = Queue(connection=redis_conn)

def my_task(x):
    return x * 2

@app.post("/enqueue")
def enqueue(x: int):
    job = queue.enqueue(my_task, x)
    return {"job_id": job.id}

실행: rq worker

강점: 매우 단순, 배우기 쉬움 약점: 기능 제한, sync 위주

Dramatiq

Celery 대안, 더 단순한 API.

큐 선택 가이드

상황	권장
간단, 실패해도 OK, 응답 후 즉시 실행	BackgroundTasks
신뢰성 필요, sync 워크플로	Celery
async 워크플로, FastAPI 호환	Arq
최소 설정, 프로토타입	RQ
대규모, 스케줄, DAG	Celery or Airflow/Prefect
마이크로서비스, 이벤트 소싱	Kafka + consumer

실시간 진행 상황

BackgroundTasks 로 시작한 작업의 진행 상황을 클라이언트에 알려야 한다면:

1. WebSocket (fastapi-websockets): 실시간 push
2. SSE (Server-Sent Events):

from fastapi import Request
from sse_starlette.sse import EventSourceResponse

@app.get("/progress/{task_id}")
async def progress(task_id: str, request: Request):
    async def event_generator():
        while True:
            if await request.is_disconnected():
                break
            status = get_task_status(task_id)
            yield {"data": status}
            if status["done"]:
                break
            await asyncio.sleep(1)
    return EventSourceResponse(event_generator())

3. Polling: 상태를 DB/Redis 저장, 클라이언트가 주기적 GET

Idempotency

Task 는 대개 재시도됩니다. Idempotent (여러 번 실행해도 같은 결과) 하게 작성:

@celery_app.task(bind=True, max_retries=3)
def send_email(self, message_id: str, to: str):
    if EmailLog.exists(message_id=message_id):
        return   # 이미 처리됨
    smtp.send(to, ...)
    EmailLog.create(message_id=message_id)

message_id 를 client 가 생성해 넘겨야 진정한 idempotency 보장.

함정

관련 위키

• FastAPI - 상위 개요
• FastAPI Async - async task
• FastAPI DI - 백그라운드 자원 주입
• FastAPI WebSockets - 진행 상황 push
• asyncio - 이벤트 루프
• concurrent.futures - 스레드/프로세스 풀