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`에 그대로 전달하지 않고, 백그라운드 작업 내부에서 독립적인 세션 생명주기를 관리하도록 구현하는 가장 권장되는 방법은 무엇입니까?
댓글4
-
빗방울
2026.10.30 04:23
-
빗방울
2026.10.30 10:10
맞음ㅋㅋ contextmanager 써서 묶어주면
코드 더 깔끔해질듯요
-
빗방울
2026.10.30 10:36
아예 sessionmaker를 contextmanager로 감싼 헬퍼 함수 하나 만들어두면 편함.
매번 세션 생성하고 닫는 거 귀찮으니까요ㅋㅋ
그렇게 하면 코드 중복도 줄고 관리하기 좋을듯요.
-
빗방울
2026.10.30 11:14
아니 근데 토이 프로젝트에 셀러리는 진짜 오버헤드 심함ㅋㅋ
그냥 세션 생성하는 함수 하나 따로 빼서
거기서 호출하는 게 제일 속 편할듯요.
모든 이야기1455
가벼운 안부부터 깊은 생각까지, 함께 나눠요.
태스크 내부에서 SessionLocal()로 새로 세션 만들어서 쓰고 finally에서 닫아주면 해결될듯함ㅋㅋ 원래 의존성 주입받은 세션은 응답 나가면 바로 닫히니까요.