백그라운드 작업

FastAPI는 BackgroundTasks로 응답 후 작업을 실행하고, Celery + Redis로 대규모 분산 작업을 처리합니다.

---

BackgroundTasks — 경량 백그라운드 작업

from fastapi import FastAPI, BackgroundTasks, Depends
import time
import logging

app = FastAPI()
logger = logging.getLogger(__name__)


# 백그라운드 작업 함수
def send_welcome_email(email: str, username: str) -> None:
    """이메일 발송 시뮬레이션 (실제로는 SMTP/SendGrid 사용)"""
    time.sleep(2)  # 네트워크 지연 시뮬레이션
    logger.info(f"[Email] {email}에 환영 메일 발송 완료")
    print(f"✉️  {email} ({username})에게 환영 메일 발송")


def log_access(user_id: int, action: str) -> None:
    """접근 로그 기록"""
    logger.info(f"[Audit] user={user_id} action={action}")


def generate_report(report_id: str, data: dict) -> None:
    """무거운 보고서 생성"""
    time.sleep(5)
    print(f"[Report] {report_id} 생성 완료: {data}")


# 엔드포인트에서 BackgroundTasks 사용
@app.post("/users/register")
async def register_user(
    username: str,
    email: str,
    background_tasks: BackgroundTasks,
):
    # 즉시 응답 반환
    user = {"id": 1, "username": username, "email": email}

    # 응답 후 백그라운드에서 실행
    background_tasks.add_task(send_welcome_email, email, username)
    background_tasks.add_task(log_access, user["id"], "register")

    return {"message": "회원가입 완료", "user": user}


# 의존성과 함께 사용
async def notify_admin(background_tasks: BackgroundTasks, message: str):
    def _notify():
        print(f"[Admin 알림] {message}")
    background_tasks.add_task(_notify)


@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    background_tasks: BackgroundTasks,
):
    # 삭제 처리
    await notify_admin(background_tasks, f"사용자 {user_id} 탈퇴")
    background_tasks.add_task(log_access, user_id, "delete")
    return {"message": f"사용자 {user_id} 삭제 완료"}

---

Celery + Redis — 분산 작업 큐

pip install celery redis
# Redis 실행: docker run -d -p 6379:6379 redis

# celery_app.py
from celery import Celery
import time

# Celery 설정
celery_app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",   # 작업 큐
    backend="redis://localhost:6379/1",  # 결과 저장
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    timezone="Asia/Seoul",
    task_track_started=True,
)


# 태스크 정의
@celery_app.task(bind=True, max_retries=3)
def send_email_task(self, to: str, subject: str, body: str) -> dict:
    """재시도 가능한 이메일 발송 태스크"""
    try:
        time.sleep(1)  # SMTP 전송 시뮬레이션
        print(f"이메일 전송: {to}")
        return {"status": "sent", "to": to}
    except Exception as exc:
        # 지수 백오프로 재시도
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)


@celery_app.task
def process_image(image_path: str, width: int, height: int) -> str:
    """이미지 리사이즈 (CPU-bound)"""
    time.sleep(3)
    print(f"이미지 처리 완료: {image_path} → {width}x{height}")
    return f"{image_path}_resized"


@celery_app.task
def generate_report(data: dict) -> dict:
    """보고서 생성"""
    time.sleep(5)
    return {"report": "generated", "rows": len(data)}

# main.py — FastAPI와 Celery 연동
from fastapi import FastAPI
from celery_app import send_email_task, process_image, generate_report
from celery.result import AsyncResult

app = FastAPI()


@app.post("/send-email")
def queue_email(to: str, subject: str, body: str):
    """이메일을 Celery 큐에 등록"""
    task = send_email_task.delay(to, subject, body)
    return {
        "task_id": task.id,
        "status": "queued",
        "message": "이메일 발송이 큐에 등록되었습니다",
    }


@app.post("/process-image")
def queue_image(image_path: str, width: int = 800, height: int = 600):
    task = process_image.delay(image_path, width, height)
    return {"task_id": task.id, "status": "queued"}


@app.get("/tasks/{task_id}")
def get_task_status(task_id: str):
    """태스크 상태 조회"""
    result = AsyncResult(task_id)
    return {
        "task_id": task_id,
        "status": result.status,
        "result": result.result if result.ready() else None,
    }

# Celery 워커 실행
celery -A celery_app worker --loglevel=info

# 모니터링 (Flower)
pip install flower
celery -A celery_app flower  # http://localhost:5555

---

비교

BackgroundTasks
  ✅ 설정 불필요, 단순 작업
  ✅ 응답 후 즉시 실행
  ❌ 프로세스 종료 시 작업 소실
  ❌ 재시도, 스케줄링 없음
  → 이메일 발송, 로그 기록, 캐시 업데이트

Celery + Redis
  ✅ 작업 영속성, 재시도, 우선순위
  ✅ 분산 처리, 스케줄링 (celery beat)
  ✅ 모니터링 (Flower)
  ❌ Redis/RabbitMQ 인프라 필요
  → 대용량 이미지 처리, 정기 리포트, 결제 처리

---

정리

도구	적합한 작업
BackgroundTasks	경량·단기 (로깅, 알림)
Celery	대규모·장기·재시도 필요
celery beat	주기적 스케줄 작업
Flower	Celery 실시간 모니터링

API 응답을 빠르게 유지하면서 무거운 작업을 비동기로 처리할 때 백그라운드 작업을 활용하세요.