백그라운드 작업¶
🌐 AI와 사람이 함께한 번역
이 번역은 사람의 안내를 받아 AI가 만들었습니다. 🤝
원문의 의미를 오해하거나 부자연스러워 보이는 등 오류가 있을 수 있습니다. 🤖
FastAPI에서는 응답을 반환한 후에 실행할 백그라운드 작업을 정의할 수 있습니다.
백그라운드 작업은 요청 후에 발생해야 하지만, 클라이언트가 응답을 받기 전에 작업이 완료될 때까지 기다릴 필요가 없는 작업에 유용합니다.
예를 들면 다음과 같습니다.
- 작업을 수행한 후 전송되는 이메일 알림:
- 이메일 서버에 연결하고 이메일을 전송하는 것은 (몇 초 정도) "느린" 경향이 있으므로, 응답은 즉시 반환하고 이메일 알림은 백그라운드에서 전송할 수 있습니다.
- 데이터 처리:
- 예를 들어 처리에 오랜 시간이 걸리는 프로세스를 거쳐야 하는 파일을 받았다면, "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가 이를 올바르게 처리하는 방법을 알고 있습니다.
이 경우 작업 함수는 파일에 쓰기를 수행합니다(이메일 전송을 시뮬레이션).
그리고 쓰기 작업은 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). - 작업 함수에 순서대로 전달되어야 하는 인자 시퀀스(
email). - 작업 함수에 전달되어야 하는 키워드 인자(
message="some notification").
의존성 주입¶
BackgroundTasks는 의존성 주입 시스템에서도 동작하며, 경로 처리 함수, 의존성(dependable), 하위 의존성 등 여러 수준에서 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"}
🤓 Other versions and variants
Tip
Prefer to use the Annotated version if possible.
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: str = Depends(get_query)
):
message = f"message to {email}\n"
background_tasks.add_task(write_log, message)
return {"message": "Message sent"}
이 예제에서는 응답이 전송된 후에 메시지가 log.txt 파일에 작성됩니다.
요청에 쿼리가 있었다면, 백그라운드 작업으로 로그에 작성됩니다.
그 다음 경로 처리 함수에서 생성된 또 다른 백그라운드 작업이 email 경로 매개변수를 사용해 메시지를 작성합니다.
기술적 세부사항¶
BackgroundTasks 클래스는 starlette.background에서 직접 가져옵니다.
FastAPI에 직접 임포트/포함되어 있으므로 fastapi에서 임포트할 수 있고, 실수로 starlette.background에서 대안인 BackgroundTask(끝에 s가 없음)를 임포트하는 것을 피할 수 있습니다.
BackgroundTask가 아닌 BackgroundTasks만 사용하면, 이를 경로 처리 함수의 매개변수로 사용할 수 있고 나머지는 FastAPI가 Request 객체를 직접 사용할 때처럼 대신 처리해 줍니다.
FastAPI에서 BackgroundTask만 단독으로 사용하는 것도 가능하지만, 코드에서 객체를 생성하고 이를 포함하는 Starlette Response를 반환해야 합니다.
더 자세한 내용은 Starlette의 Background Tasks 공식 문서에서 확인할 수 있습니다.
주의사항¶
무거운 백그라운드 계산을 수행해야 하고, 반드시 동일한 프로세스에서 실행할 필요가 없다면(예: 메모리, 변수 등을 공유할 필요가 없음) Celery 같은 더 큰 도구를 사용하는 것이 도움이 될 수 있습니다.
이들은 RabbitMQ나 Redis 같은 메시지/작업 큐 관리자 등 더 복잡한 설정을 필요로 하는 경향이 있지만, 여러 프로세스에서, 특히 여러 서버에서 백그라운드 작업을 실행할 수 있습니다.
하지만 동일한 FastAPI 앱의 변수와 객체에 접근해야 하거나, 작은 백그라운드 작업(예: 이메일 알림 전송)을 수행해야 한다면, BackgroundTasks를 간단히 사용하면 됩니다.
요약¶
경로 처리 함수와 의존성에서 매개변수로 BackgroundTasks를 임포트해 사용하여 백그라운드 작업을 추가합니다.