문서가격블로그
문의하기
로그인무료로 가입하기
Back to Tech Articles

FastAPI에서 백그라운드 작업 및 장기 실행 작업 관리

9월 09, 2025
# Misc

Grace Collins

Solutions Engineer · Leapcell

이 게시물은 English, 日本語 및 Deutsch에서도 볼 수 있습니다.
Cover of "FastAPI에서 백그라운드 작업 및 장기 실행 작업 관리"

소개

현대 웹 애플리케이션의 세계에서 사용자 경험은 무엇보다 중요합니다. 느리거나 응답이 없는 인터페이스는 사용자에게 좌절감을 주고 결국 이탈하게 만들 수 있습니다. 종종 애플리케이션은 계산 집약적이거나, 외부 서비스를 포함하거나, 단순히 완료하는 데 상당한 시간이 걸리는 작업을 수행해야 합니다. 이러한 작업을 메인 요청-응답 주기 내에서 동기적으로 실행하면 서버가 차단되어 사용자에게 지연 및 잠재적인 시간 초래가 발생합니다.

이것이 바로 백그라운드 작업이라는 개념이 중요한 이유입니다. 이러한 장기 실행 작업을 비동기적으로 처리하도록 하여 API를 빠르고 응답성이 뛰어나며 즉각적인 사용자 상호 작용에 계속 사용할 수 있도록 보장할 수 있습니다. 이 문서는 Python 3.7+를 사용하여 API를 구축하기 위한 현대적이고 빠른(고성능) 웹 프레임워크인 FastAPI가 간단한 즉시 실행 작업부터 전용 작업 큐가 필요한 복잡한 장기 실행 프로세스까지 이러한 작업을 효과적으로 관리할 수 있도록 개발자에게 어떻게 권한을 부여하는지 살펴볼 것입니다.

핵심 개념 설명

구현 세부 사항을 자세히 살펴보기 전에 논의의 중심이 될 몇 가지 주요 용어를 정의해 보겠습니다.

  • 동기 작업(Synchronous Operation): 다음 작업을 시작하기 전에 완료해야 하는 작업입니다. 웹 서버 맥락에서 요청 처리기가 동기 장기 실행 작업을 수행하면 해당 작업이 완료될 때까지 서버가 다른 요청을 처리하는 것을 차단합니다.
  • 비동기 작업(Asynchronous Operation): 메인 프로그램이 완료를 기다리지 않고 다른 작업을 계속 실행할 수 있도록 백그라운드에서 독립적으로 실행될 수 있는 작업입니다. 비동기성을 활용하는 웹 서버는 일부 요청이 I/O 작업을 기다리는 경우에도 여러 요청을 동시에 처리할 수 있습니다.
  • 백그라운드 작업(Background Task): 웹 요청의 컨텍스트 내에서 시작되지만 클라이언트에 HTTP 응답이 전송된 후에 처리되는 비동기 작업의 특정 유형입니다. 이러한 작업은 일반적으로 클라이언트가 작업 완료를 기다릴 필요가 없는 '즉시 실행' 작업입니다.
  • 장기 실행 작업(Long-Running Task): 상당한 시간(초, 분 또는 시간)이 걸리는 작업입니다. 이러한 작업에는 종종 복잡한 계산, 데이터 처리, 파일 변환 또는 외부 API와의 통합이 포함됩니다. 백그라운드 작업은 장기 실행이 될 수 있지만 모든 장기 실행 작업이 간단한 즉시 실행 작업은 아니며 일부는 진행 상황 추적, 오류 처리 및 잠재적 재시도가 필요합니다.
  • 작업 큐(Task Queue): 비동기적으로 작업을 관리하고 실행하도록 설계된 시스템입니다. 장기 실행 작업이 시작되면 큐에 푸시됩니다. 그런 다음 별도의 워커 프로세스가 이 큐에서 작업을 가져와 실행하며, 종종 재시도, 예약 및 결과 저장과 같은 기능을 제공합니다. 인기 있는 예로는 Celery 및 RQ(Redis Queue)가 있습니다.
  • 워커(Worker): 작업 큐에서 작업을 가져와 실행하는 전용 프로세스 또는 워커 세트입니다. 워커는 메인 API 서버와 독립적으로 작동합니다.

FastAPI에서 작업 처리

FastAPI는 BackgroundTasks 종속성을 사용하여 간단한 백그라운드 작업을 기본 지원합니다. 더 복잡하고 진정한 장기 실행 작업을 위해서는 외부 비동기 작업 큐와 통합하는 것이 업계 표준입니다.

간단한 오프로딩을 위한 FastAPI의 BackgroundTasks

FastAPI의 BackgroundTasks 기능은 HTTP 응답이 전송된 후에 실행되어야 하는 가볍고 비정상적인 작업에 이상적입니다. 여기에는 이메일 알림 보내기, 활동 로깅 또는 캐시 업데이트가 포함될 수 있습니다. 여기서 핵심 특징은 클라이언트가 작업 완료를 기다리지 않으며 실패가 즉각적인 사용자 경험에 직접적인 영향을 미치지 않는다는 것입니다.

사용자가 리소스에 성공적으로 액세스한 후 사용자 활동을 기록하는 예를 통해 이를 설명해 보겠습니다.

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

app = FastAPI()

def write_log(message: str):
    with open("app.log", mode="a") as log:
        log.write(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {message}\n")

@app.post("/send_email/{email}")
async def send_email_background(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"Sending email to {email}")
    # 이메일 보내기 프로세스 시뮬레이션 (다른 백그라운드 작업 또는 실제 비동기 작업일 수 있음)
    # email_service.send_async(email, "Welcome!", "Thanks for signing up!")
    return {"message": "Email sending initiated. Check logs for details."}

@app.get("/items/{item_id}")
async def read_item(item_id: int, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"Accessed item {item_id}")
    return {"item_id": item_id, "message": "Item accessed successfully."}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

이 예에서:

  1. 작업을 시뮬레이션하는 간단한 write_log 함수를 정의합니다.
  2. API 엔드포인트(/send_email/{email} 및 /items/{item_id})에서 background_tasks: BackgroundTasks를 종속성으로 선언합니다.
  3. 그런 다음 background_tasks.add_task(function_name, *args, **kwargs)를 사용하여 write_log 함수를 예약합니다.
  4. FastAPI는 클라이언트로의 /send_email/{email} 또는 /items/{item_id} HTTP 응답이 전송된 후에 write_log가 실행되도록 보장합니다. 클라이언트는 즉시 응답을 받지만 로깅은 백그라운드에서 수행됩니다.

원칙: BackgroundTasks는 응답이 스트리밍된 후에 동일한 이벤트 루프 내에서 실행될 호출 가능한 항목을 예약하여 작동합니다. 이는 작고 빠른 작업을 처리하는 데 효율적이지만 그 한계를 이해하는 것이 중요합니다. 백그라운드 작업 자체가 너무 오래 걸리면 여전히 메인 이벤트 루프 리소스를 차지하여 다른 요청의 응답성에 영향을 줄 수 있습니다(초기 응답 후에 발생하더라도). 또한 실패 또는 서버가 다시 시작될 경우 지속성 또는 재시도 메커니즘을 제공하지 않습니다.

진정한 장기 실행 작업을 외부 작업 큐로 처리

CPU 집약적이거나, 외부 서비스에 I/O 집약적이거나, 실패하기 쉽거나, 단순히 메인 FastAPI 프로세스(응답 후에도) 내에서 실행하기에 너무 긴 작업의 경우 Celery 또는 RQ와 같은 전용 비동기 작업 큐와 통합하는 것이 필수적입니다. 이러한 시스템은 웹 서버에서 작업 실행을 분리하여 견고성, 확장성 및 관찰 가능성을 제공합니다.

사용자가 대용량 이미지를 업로드하고(크기 조정, 필터 적용, 클라우드 스토리지 업로드) 몇 초 또는 몇 분이 걸릴 수 있는 처리가 필요하다고 가정해 보겠습니다.

아키텍처:

  1. FastAPI 애플리케이션: 사용자 요청을 수신하고 작업 세부 정보를 작업 큐(예: Redis)에 푸시합니다.
  2. Redis(또는 RabbitMQ): 메시지 브로커 역할을 하여 작업을 큐에 저장합니다.
  3. 워커 프로세스(Celery 또는 RQ): 큐를 지속적으로 모니터링하고 작업을 가져와 실행하는 별도의 영구 프로세스입니다. 완료되면 데이터베이스를 업데이트하거나 사용자에게 알릴 수 있습니다.

여기서는 매우 인기 있고 강력한 선택인 Redis를 사용하는 Celery를 사용할 것입니다.

1. Celery 및 Redis 설정:

  • 필요한 패키지를 설치합니다: pip install "celery[redis]"
  • Redis 서버가 실행 중인지 확인합니다. 일반적으로 redis-server로 시작할 수 있습니다.

2. Celery 애플리케이션 생성(worker.py):

# worker.py
from celery import Celery
import time

# Celery 구성
# broker_url은 Celery가 작업을 가져오는 위치입니다.
# result_backend은 Celery가 작업 결과를 저장하는 위치입니다(장기 실행 작업에 유용하지만 선택 사항).
celery_app = Celery(
    'my_app',
    broker='redis://localhost:6379/0',
    result_backend='redis://localhost:6379/0'
)

# Celery 작업 정의
@celery_app.task
def process_image(image_id: str, operations: list):
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Starting image processing for {image_id} with operations: {operations}")
    time.sleep(10) # 장기 실행 이미지 처리 시뮬레이션
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Finished image processing for {image_id}")
    return {"image_id": image_id, "status": "processed", "applied_operations": operations}

@celery_app.task
def send_marketing_email(recipient_email: str, subject: str, body: str):
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Sending email to {recipient_email}...")
    time.sleep(5) # 이메일 보내기 시뮬레이션
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Email sent to {recipient_email}")
    return {"email": recipient_email, "status": "sent", "subject": subject}

3. Celery와 FastAPI 통합(main.py):

# main.py
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
from worker import process_image, send_marketing_email # Celery 작업 가져오기
import uvicorn

app = FastAPI()

class ImageProcessingRequest(BaseModel):
    image_id: str
    operations: list

class EmailRequest(BaseModel):
    recipient_email: str
    subject: str
    body: str

@app.post("/process_image/", status_code=202) # 나중에 완료될 작업에 대해 202 Accepted 사용
async def schedule_image_processing(request: ImageProcessingRequest):
    # Celery에 작업 큐에 넣기
    task = process_image.delay(request.image_id, request.operations)
    return {"message": "Image processing task scheduled", "task_id": task.id}

@app.post("/send_marketing_email/", status_code=202)
async def schedule_marketing_email(request: EmailRequest):
    task = send_marketing_email.apply_async(
        args=[request.recipient_email, request.subject, request.body],
        countdown=60 # 60초 후에 실행되도록 예약
    )
    return {"message": "Marketing email task scheduled", "task_id": task.id}

@app.get("/task_status/{task_id}")
async def get_task_status(task_id: str):
    task = celery_app.AsyncResult(task_id)
    if task.state == 'PENDING':
        response = {"status": task.state, "info": "Task is waiting to be processed or is in progress."}
    elif task.state != 'FAILURE':
        response = {"status": task.state, "result": task.result}
    else:
        # 작업 실패
        response = {"status": task.state, "info": str(task.info)} # 예외 저장
    return response

# FastAPI의 기본 BackgroundTasks를 시연하기 위한 예시 (이전과 동일)
def write_log(message: str):
    with open("app.log", mode="a") as log:
        log.write(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] FastAPI BgTask: {message}\n")

@app.get("/fastapi_bgtask_example")
async def fastapi_bgtask_example(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "This is a simple native FastAPI background task.")
    return {"message": "FastAPI background task initiated."}

if __name__ == "__main__":
    # 실행 방법:
    # 1. Redis 시작: redis-server
    # 2. Celery 워커 시작: celery -A worker worker --loglevel=info
    # 3. FastAPI 앱 시작: python main.py
    uvicorn.run(app, host="0.0.0.0", port=8000)

4. 시스템 실행:

  1. Redis 서버 시작: redis-server (별도의 터미널에서)
  2. Celery 워커 시작: celery -A worker worker --loglevel=info (다른 별도의 터미널에서, worker는 worker.py를 나타냅니다)
  3. FastAPI 애플리케이션 시작: python main.py (세 번째 터미널에서)

이제 FastAPI 앱에서 /process_image/ 또는 /send_marketing_email/로 요청을 보내면 FastAPI는 즉시 task_id와 함께 202 Accepted 상태를 반환합니다. 실제 이미지 처리 또는 이메일 보내기는 Celery 워커가 백그라운드에서 처리하며 웹 서버의 요청-응답 주기에서 완전히 분리됩니다. /task_status/{task_id}를 쿼리하여 작업 진행 상황이나 결과를 확인할 수 있습니다.

비교 및 사용 사례:

  • BackgroundTasks (FastAPI 네이티브):

    • 장점: 구현이 간단하고, 외부 종속성이 없으며(FastAPI 자체 제외), 빠르고 비정상적인 후속 작업에 좋습니다.
    • 단점: FastAPI 프로세스의 동일한 이벤트 루프 내에서 실행되며, 지속성이 없습니다(서버가 다시 시작되면 진행 중인 작업이 손실됨), 재시도 메커니즘이 없으며, 진행 상황 추적이나 결과 저장이 없습니다. 무거운 작업에 대한 확장성이 제한적입니다.
    • 사용 사례: 사용자 등록 후 빠른 환영 이메일 보내기(서버 부하가 낮은 경우), 카운터 업데이트, API 액세스 로깅, 작은 캐시 지우기.

  • 외부 작업 큐 (예: Celery, RQ):

    • 장점: 작업 실행의 진정한 분리, 장기 실행 및 실패 가능성이 있는 작업의 강력한 처리, 지속성(작업은 서버 재시작 시에도 생존 가능), 자동 재시도, 동시성 제어, 예약 기능, 진행 상황 추적, 결과 저장, 확장성(필요에 따라 워커 더 늘리기).
    • 단점: 외부 종속성(메시지 브로커, 워커 프로세스)으로 복잡성이 증가하며, 처음에는 학습 곡선이 가파릅니다.
    • 사용 사례: 이미지/비디오 처리, 대규모 데이터 내보내기, 보고서 생성, 대량 이메일/알림 보내기, 외부 서비스에 대한 장기 실행 API 호출, 복잡한 금융 계산, 처리 시간이 길거나 오류 복구가 필요한 모든 작업.

결론

백그라운드 및 장기 실행 작업을 효과적으로 관리하는 것은 고성능 및 확장 가능한 웹 애플리케이션을 구축하는 초석입니다. FastAPI의 BackgroundTasks는 간단한 즉시 실행 작업에 대해 즉각적인 응답을 차단하지 않는 간편한 솔루션을 제공합니다. 더 복잡하고, 강력하고, 지속성 및 오류 처리가 필요한 모든 것에 대해서는 Celery 또는 RQ와 같은 전용 비동기 작업 큐와 통합하는 것이 권장되는 접근 방식입니다. 이러한 도구를 전략적으로 활용함으로써 개발자는 FastAPI 애플리케이션을 민첩하고 응답성이 뛰어나며 사용자 경험을 저하시키지 않고 까다로운 워크로드를 처리할 수 있도록 보장할 수 있습니다.

Share this article

HackerNews

More Posts from Leapcell

cover

Flask-SQLAlchemy 모델, 관계 및 트랜잭션 관리를 사용한 견고한 애플리케이션 구축

cover

Gin 라우트 그룹 및 버전 관리를 통한 API 관리 간소화

Popular Posts