백그라운드 작업
백그라운드 작업 (Background Tasks)
응답을 반환한 후에 실행할 백그라운드 작업을 정의할 수 있어요.
이건 요청이 끝난 뒤에 일어나야 하지만, 클라이언트가 응답을 받기 전에 작업이 끝나기를 기다릴 필요가 없는 작업에 유용해요.
예를 들면:
- 액션을 수행한 뒤 보내는 이메일 알림:
- 이메일 서버에 연결해서 이메일을 보내는 건 "느린"(몇 초) 경향이 있으므로, 응답을 바로 반환하고 이메일 알림을 백그라운드로 보낼 수 있어요.
- 데이터 처리:
- 예를 들어 느린 과정을 거쳐야 하는 파일을 받았다고 해 볼게요. "Accepted"(HTTP 202) 응답을 반환하고 파일을 백그라운드에서 처리할 수 있어요.
출처: 공식문서
BackgroundTasks 사용하기
먼저 BackgroundTasks를 임포트하고, 경로 연산 함수에 BackgroundTasks 타입 선언으로 파라미터를 정의해요:
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_notification(email: str, message=""):
with open("log.txt", mode="w") as email_file:
content = f"notification for {email}: {message}"
email_file.write(content)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_notification, email, message="some notification")
return {"message": "Notification sent in the background"}
FastAPI가 BackgroundTasks 타입의 객체를 만들어서 그 파라미터로 넘겨줘요.
작업 함수 만들기
백그라운드 작업으로 실행할 함수를 만들어요.
파라미터를 받을 수 있는 평범한 표준 함수예요.
async def 또는 일반 def 함수일 수 있고, FastAPI가 알아서 올바르게 처리해요.
이 경우 작업 함수는 파일에 씁니다(이메일 보내기를 시뮬레이션).
그리고 쓰기(write) 작업은 async와 await를 쓰지 않으므로, 함수를 일반 def로 정의해요:
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_notification(email: str, message=""):
with open("log.txt", mode="w") as email_file:
content = f"notification for {email}: {message}"
email_file.write(content)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_notification, email, message="some notification")
return {"message": "Notification sent in the background"}
백그라운드 작업 추가하기
경로 연산 함수 안에서, 작업 함수를 .add_task() 메서드로 백그라운드 작업 객체에 넘겨요:
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_notification(email: str, message=""):
with open("log.txt", mode="w") as email_file:
content = f"notification for {email}: {message}"
email_file.write(content)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_notification, email, message="some notification")
return {"message": "Notification sent in the background"}
.add_task()는 인자로 이걸 받아요:
- 백그라운드에서 실행할 작업 함수(
write_notification). - 작업 함수에 순서대로 전달할 위치 인자(args) 시퀀스(
email). - 작업 함수에 전달할 키워드 인자(kwargs)(
message="some notification").
의존성 주입 (Dependency Injection)
BackgroundTasks를 쓰는 건 의존성 주입 시스템에서도 동작해요. 여러 단계에서 BackgroundTasks 타입 파라미터를 선언할 수 있어요. 경로 연산 함수에서, 의존성(주입 가능한 것)에서, 하위 의존성에서, 등등.
FastAPI는 각 경우에 무엇을 해야 할지, 그리고 같은 객체를 어떻게 재사용할지 알고 있어서, 모든 백그라운드 작업이 합쳐지고 나중에 백그라운드로 실행돼요:
from typing import Annotated
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: str | None = None):
if q:
message = f"found query: {q}\n"
background_tasks.add_task(write_log, message)
return q
@app.post("/send-notification/{email}")
async def send_notification(
email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
message = f"message to {email}\n"
background_tasks.add_task(write_log, message)
return {"message": "Message sent"}
이 예시에서 메시지는 응답을 보낸 후에 log.txt 파일에 쓰여요.
요청에 쿼리(query)가 있었다면, 그 쿼리가 백그라운드 작업에서 로그에 쓰여요.
그리고 경로 연산 함수에서 생성된 또 다른 백그라운드 작업이 email 경로 파라미터를 써서 메시지를 써요.
팁 — 가능하면
Annotated버전을 쓰는 게 좋아요. (non-Annotated 버전도 있어요.q: str = Depends(get_query)처럼 쓰면 돼요.)
기술적 세부사항 (Technical Details)
BackgroundTasks 클래스는 starlette.background에서 직접 왔어요.
FastAPI에 직접 임포트/포함되어 있어서 fastapi에서 임포트할 수 있고, starlette.background의 대안인 BackgroundTask(끝에 s가 없는 것)를 실수로 임포트하는 걸 피할 수 있어요.
BackgroundTasks만 씀으로써(BackgroundTask가 아니라) 경로 연산 함수 파라미터로 사용할 수 있고 나머지는 FastAPI가 처리해 줘요. Request 객체를 직접 쓸 때와 같죠.
FastAPI에서 BackgroundTask 단독으로도 쓸 수는 있지만, 코드에서 객체를 만들어서 Starlette Response에 포함해 반환해야 해요.
구체적인 내용은 Starlette의 공식 Background Tasks 문서에서 볼 수 있어요.
주의사항 (Caveat)
무거운 백그라운드 계산을 수행해야 하고, 반드시 같은 프로세스에서 실행될 필요가 없다면(예: 메모리, 변수 등을 공유할 필요가 없다면), Celery 같은 더 큰 도구를 쓰는 게 이로울 수 있어요.
그런 도구들은 더 복잡한 설정, RabbitMQ나 Redis 같은 메시지/작업 큐 관리자가 필요하지만, 백그라운드 작업을 여러 프로세스에서, 특히 여러 서버에서 실행할 수 있게 해줘요.
하지만 같은 FastAPI 앱의 변수와 객체에 접근해야 하거나, 작은 백그라운드 작업(이메일 알림 보내기 같은)을 수행해야 한다면 BackgroundTasks를 그냥 쓰면 돼요.
요약 (Recap)
BackgroundTasks를 임포트해서 경로 연산 함수와 의존성에서 파라미터로 쓰면 백그라운드 작업을 추가할 수 있어요.