본문으로 바로가기
COMMUNITY모든 이야기

FastAPI의 BackgroundTasks를 사용하여 이메일 발송 로직을 비동기로 처리하려 합니다. 하지만 백그라운드 작업 내에서 DB 세션에 접근할 때 세션이 이미 닫혀 있다는 오류가 발생합니다.

코드:

from fastapi import FastAPI, BackgroundTasks, Depends
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy import create_engine

engine = create_engine("sqlite:///test.db")
SessionLocal = sessionmaker(bind=engine)
app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def send_email_task(db: Session, user_id: int):
    user = db.query(User).filter(User.id == user_id).first()
    print(f"Sending email to {user.email}")

@app.post("/notify/{user_id}")
async def notify(user_id: int, background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
    background_tasks.add_task(send_email_task, db, user_id)
    return {"message": "Task queued"}

오류: `sqlalchemy.exc.ResourceClosedError: This session is closed.`

시도: 1. `get_db` 의 `finally` 블록을 제거했으나 커넥션 풀 누수가 발생함.

2. `BackgroundTasks` 대신 `Celery` 도입을 검토했으나 토이 프로젝트 규모상 오버헤드가 큼.

질문: `Depends(get_db)`로 주입된 세션을 `BackgroundTasks`에 그대로 전달하지 않고, 백그라운드 작업 내부에서 독립적인 세션 생명주기를 관리하도록 구현하는 가장 권장되는 방법은 무엇입니까?

모든 이야기1464

가벼운 안부부터 깊은 생각까지, 함께 나눠요.

글쓰기
총 1464개의 이야기
댓글