40. Background Tasks
1. Background Tasks
- response를 반환한 후 실행할 백그라운드 작업을 정의할 수 있다.
- 이것은 request 후에 발생해야 하는 작업에 유용하지만, 클라이언트는 response를 받기 전 작업이 완료될 때까지 기다릴 필요가 없다.
- 이와 관련하여 다음과 같은 예시가 있다.
1] 작업 수행 후 전송되는 이메일 알림
- 이메일 서버에 연결하여 이메일을 보내는 것은 느린 경향이 있으므로 즉시 response를 반환하고, 백그라운드에서 이메일 알림을 보낼 수 있다.
2] 데이터 처리
- 느린 프로세스를 거쳐야 하는 파일을 받았다고 가정할 때, "Accepted"(HTTP 202) response를 먼저 반환하고 백그라운드에서 처리할 수 있다.
2. Using BackgroundTasks
- 먼저,
BackgroundTasks를 임포트한 후,BackgroundTasks타입 선언을 사용하여 path operation function에서 매개변수를 정의한다.
from fastapi import FastAPI, BackgroundTasks
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 타입의 객체를 만들고 해당 매개변수로 전달한다.
3. Create a task function
- 백그라운드 작업으로 실행할 함수를 만드는데, 이 함수는 매개변수를 받을 수 있는 표준 함수이다.
async def또는 일반적인def함수일 수 있으며, FastAPI는 이를 올바르게 처리한다.- 이 경우 작업 함수는 파일에 기록하게 된다.
- 그리고 쓰기 작업은
async및await을 사용하지 않으므로 일반적인def로 함수를 정의한다.
from fastapi import FastAPI, BackgroundTasks
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"}
4. Add the background task
- path operation function 내에서
.add_task()메소드를 사용하여 작업 함수를 백그라운드 작업 객체에 전달한다.
from fastapi import FastAPI, BackgroundTasks
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()메소드는 다음과 같은 인수를 받는다.
1] 백그라운드에서 실행할 작업 함수(write_notification)
2] 작업 함수에 순서대로 전달되어야 하는 인수의 시퀀스(email)
3] 작업 함수에 전달되어야 하는 모든 키워드 인수(message="some notification")
5. Dependency Injection
BackgroundTasks를 사용하면 dependency injection 시스템과도 작동하므로, path operation function, depedency, sub-dependency 등 여러 수준에서BackgroundTasks타입의 매개변수를 선언할 수 있다.- FastAPI는 각 경우에 수행할 작업과 동일한 객체를 재사용하는 방법을 알고 있으므로 모든 백그라운드 작업이 함께 병합되고 나중에는 백그라운드에서 실행된다.
from typing import Optional
from fastapi import FastAPI, BackgroundTasks, Depends
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: Optional[str] = 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"}
- 위의 예제에서 메시지는 response가 전송된 후
log.txt파일에 기록된다. - request에 query가 있는 경우 백그라운드 작업의 로그에 기록된다.
- 그러면 path operation function에서 생성된 또 다른 백그라운드 작업이 path operation 매개변수
email을 사용하여 메시지를 작성한다.