FastAPI BackgroundTasks 사용법 | API 응답 후 작업 실행하기

 

FastAPI로 API를 개발하다 보면 이런 상황이 있습니다.

"API 응답은 빨리 보내고, 로그 저장이나 이메일 전송 같은 작업은 나중에 처리하고 싶다."

이럴 때 사용할 수 있는 기능 중 하나가 FastAPI의 BackgroundTasks입니다.

---

1. BackgroundTasks란?

BackgroundTasks는 API 요청에 대한 응답을 반환한 후 추가 작업을 실행할 수 있도록 도와주는 기능입니다.

예를 들어 다음과 같은 상황에서 사용할 수 있습니다.

• 이메일 알림
• 간단한 로그 저장
• 간단한 파일 처리
• 사용자에게 알림을 보낸 후 추가 작업

중요한 점은 무거운 작업을 무조건 BackgroundTasks로 처리하면 안 된다는 것입니다.

---

2. 가장 간단한 예제

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", "a", encoding="utf-8") as file:
        file.write(message + "\n")


@app.post("/log")
async def create_log(
    message: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(write_log, message)

    return {
        "message": "요청이 처리되었습니다."
    }

핵심은 다음 부분입니다.

background_tasks.add_task(write_log, message)

이 코드를 통해 write_log 함수를 백그라운드 작업으로 등록합니다.

---

3. 실행 순서

일반적인 흐름은 다음과 같습니다.

1. 클라이언트가 API 호출
2. FastAPI가 요청 처리
3. 응답 반환
4. 등록된 Background Task 실행

따라서 사용자는 로그 파일에 기록하는 작업이 끝날 때까지 반드시 기다릴 필요가 없습니다.

---

4. 함수에 여러 값을 전달하기

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def send_message(email: str, message: str):
    print(f"받는 사람: {email}")
    print(f"메시지: {message}")


@app.post("/send")
async def send(
    email: str,
    message: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(
        send_message,
        email,
        message
    )

    return {
        "message": "메일 작업이 등록되었습니다."
    }

add_task의 첫 번째 값은 실행할 함수이고, 그 뒤의 값들은 해당 함수에 전달할 값입니다.

---

5. async 함수도 사용할 수 있을까?

가능합니다.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


async def process_data(data: str):
    print(data)


@app.post("/process")
async def process(
    data: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(process_data, data)

    return {
        "message": "처리가 시작되었습니다."
    }

일반 def 함수와 async def 함수 모두 BackgroundTasks에서 사용할 수 있습니다.

---

6. Dependency에서도 사용할 수 있다

from typing import Annotated

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", "a", encoding="utf-8") as file:
        file.write(message + "\n")


def get_query(
    background_tasks: BackgroundTasks,
    q: str | None = None
):
    if q:
        background_tasks.add_task(
            write_log,
            f"검색어: {q}"
        )

    return q


@app.get("/search")
async def search(
    q: Annotated[str | None, Depends(get_query)]
):
    return {
        "query": q
    }

FastAPI 공식 문서에서도 BackgroundTasks를 Dependency와 함께 사용하는 방법을 안내하고 있습니다.

---

7. BackgroundTasks를 사용하면 안 되는 경우

여기서 가장 중요한 부분입니다.

BackgroundTasks는 모든 비동기 작업을 해결해주는 만능 기능이 아닙니다.

예를 들어 다음과 같은 무거운 작업은 별도의 작업 큐를 고려하는 것이 좋습니다.

• 대용량 영상 인코딩
• 대규모 데이터 분석
• 수십 분 이상 걸리는 작업
• 여러 서버에서 작업을 분산해야 하는 경우
• 작업 성공/실패 상태를 별도로 관리해야 하는 경우

이런 경우에는 Celery, RabbitMQ, Redis 기반 작업 큐 등의 별도 구조를 검토하는 것이 좋습니다.

---

8. BackgroundTasks와 Celery의 차이

구분BackgroundTasksCelery

설정 난이도	낮음	높음
간단한 작업	적합	가능
대규모 작업	부적합	적합
별도 메시지 브로커	필요 없음	일반적으로 필요

---

9. 정리

FastAPI의 BackgroundTasks는 API 응답 후 실행해도 되는 간단한 작업에 적합합니다.

특히 이메일 알림이나 간단한 로그 저장처럼 사용자가 결과를 기다릴 필요가 없는 작업에 활용하면 좋습니다.

반대로 무거운 작업이나 여러 서버에 걸쳐 안정적으로 처리해야 하는 작업이라면 Celery 같은 별도의 작업 큐 구조를 고려하는 것이 좋습니다.

관련 검색어

FastAPI BackgroundTasks, FastAPI 백그라운드 작업, FastAPI 비동기, FastAPI Celery, FastAPI 작업 큐