문서 수정 기능을 구현했다. 처음에는 간단하게 문서를 업데이트하고, 첨부파일을 추가 및 삭제한 후 결과를 반환하는 방식으로 개발했다. 하지만 API를 실제로 호출했을 때, 문서 객체를 응답하는 과정에서 첨부파일 객체와의 세션이 끊기는 오류가 발생했다.

문제는 트랜잭션 관리가 미흡했기 때문이었다. FastAPI에서 SQLAlchemy 세션 관리를 명시적으로 하지 않으면, 서비스 계층에서 데이터를 수정하거나 조회한 후 스키마로 변환하는 과정에서 DetachedInstanceError와 같은 오류가 발생할 수 있다. 이는 SQLAlchemy가 데이터베이스 세션을 통해 객체를 로드하고, 세션이 닫히면 지연 로딩(Lazy Loading) 같은 동작이 불가능하기 때문이다.

해결 방법은 데코레이터를 사용해 트랜잭션 관리 로직을 캡슐화하고, 문서 조회에서 스키마 반환까지 같은 세션을 유지하는 것이었다. 이를 통해 코드의 간결함을 유지하면서도 안정적으로 트랜잭션을 처리할 수 있었다. 또한 FastAPI의 의존성 주입을 활용한 방식도 대안으로 고려했다. 이 글에서는 두 가지 접근 방식을 비교하고 장단점을 분석해본다.

---

문제 상황

문제는 문서를 수정하고 응답으로 반환하는 과정에서 발생했다. 문서 객체와 첨부파일은 서로 연관되어 있고, 첨부파일은 지연 로딩으로 설정되어 있었다. 그런데 세션이 닫힌 상태에서 첨부파일 데이터를 조회하려고 하면 DetachedInstanceError가 발생한다.

초기 코드에서의 문제점은 다음과 같다:

1. 세션 범위가 명확하지 않음

   문서 조회와 스키마 변환 사이에서 같은 세션을 사용하지 않았기 때문에 세션이 닫혀 첨부파일 객체를 로드하지 못했다.

2. 트랜잭션 관리 부족

   FastAPI는 트랜잭션을 자동으로 관리하지 않는다. 명시적으로 트랜잭션을 정의하지 않으면 데이터 무결성이나 동기화 문제를 야기할 수 있다.

---

초기 코드

초기에는 단순히 문서를 조회, 수정하고 첨부파일을 처리한 뒤 응답을 반환하는 구조였다.

async def update_user_document(
    self,
    user_id: int,
    docs_id: int,
    user_document_update: UserDocumentsUpdate,
    files: List[UploadFile],
    attachment_types: List[str],
    values: List[str],
    file_ids: List[str],
) -> UserDocumentsResponse:
    # 문서 조회
    docs = self.document_repository.get_user_document_by_id(docs_id=docs_id, user_id=user_id)
    if not docs or docs.user_id != user_id:
        raise DocumentNotFoundException("문서 ID를 찾을 수 없거나 수정 권한이 없습니다.")

    # 첨부파일 처리
    await self.docs_attachment_service.update_attachments(
        doc_id=docs_id,
        files=files,
        attachment_types=attachment_types,
        values=values,
        file_ids=file_ids,
    )

    # 문서 수정
    updated_docs = self.document_repository.update_user_documents(
        docs_id=docs_id, user_document_update=user_document_update
    )

    # 스키마로 변환
    return UserDocumentsResponse.from_model(updated_docs)

이 코드는 문서와 첨부파일을 수정하는 역할을 한다. 그러나 문서 조회와 스키마 변환 과정에서 세션이 끊겨 오류가 발생했다. 특히, 첨부파일 객체가 Lazy Loading으로 설정되어 있어 세션이 없으면 데이터를 가져올 수 없었다.

---

해결 방법 1: 데코레이터 방식

데코레이터를 사용해 트랜잭션 관리 로직을 캡슐화했다.

이 방식은 서비스 계층의 함수에 세션을 주입하고, 트랜잭션 범위를 명확히 정의하는 데 유용하다.

데코레이터 구현

from functools import wraps
from sqlalchemy.orm import Session
from app.core.db import SessionLocal

def transactional(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        session: Session = SessionLocal()
        try:
            kwargs['db'] = session  # 세션 주입
            result = await func(*args, **kwargs)
            session.commit()  # 성공 시 커밋
            return result
        except Exception as e:
            session.rollback()  # 예외 발생 시 롤백
            raise e
        finally:
            session.close()  # 세션 닫기
    return wrapper

이 데코레이터는 서비스 계층의 함수가 호출될 때 SQLAlchemy 세션을 생성하고, 트랜잭션을 관리한다.

수정된 코드

@transactional
async def update_user_document(
    self,
    user_id: int,
    docs_id: int,
    user_document_update: UserDocumentsUpdate,
    files: List[UploadFile],
    attachment_types: List[str],
    values: List[str],
    file_ids: List[str],
    db: Session,  # 세션 주입
) -> UserDocumentsResponse:
    # 문서 조회
    docs = self.document_repository.get_user_document_by_id(docs_id=docs_id, user_id=user_id, db=db)
    if not docs:
        raise DocumentNotFoundException("문서를 찾을 수 없습니다.")

    # 첨부파일 처리
    await self.docs_attachment_service.update_attachments(
        doc_id=docs_id, files=files, attachment_types=attachment_types, values=values, file_ids=file_ids, db=db
    )

    # 문서 수정
    updated_docs = self.document_repository.update_user_documents(
        docs_id=docs_id, user_document_update=user_document_update, db=db
    )

    # 스키마 반환
    return UserDocumentsResponse.from_model(updated_docs)

이 방식은 코드가 간결하고, 트랜잭션 관리 로직이 캡슐화되어 서비스 로직에 집중할 수 있는 장점이 있다.

---

해결 방법 2: Depends 방식

Depends를 사용하면 FastAPI의 의존성 주입 시스템을 활용해 트랜잭션을 관리할 수 있다.

세션 의존성 정의

from app.core.db import SessionLocal

def get_db():
    db = SessionLocal()
    try:
        yield db  # 세션 생성
        db.commit()  # 트랜잭션 커밋
    except Exception:
        db.rollback()  # 롤백
        raise
    finally:
        db.close()  # 세션 종료

이 함수는 FastAPI의 의존성 주입으로 세션을 관리한다.

수정된 코드

async def update_user_document(
    self,
    user_id: int,
    docs_id: int,
    user_document_update: UserDocumentsUpdate,
    files: List[UploadFile],
    attachment_types: List[str],
    values: List[str],
    file_ids: List[str],
    db: Session = Depends(get_db),  # 세션 주입
) -> UserDocumentsResponse:
    # 문서 조회
    docs = self.document_repository.get_user_document_by_id(docs_id=docs_id, user_id=user_id, db=db)
    if not docs:
        raise DocumentNotFoundException("문서를 찾을 수 없습니다.")

    # 첨부파일 처리
    await self.docs_attachment_service.update_attachments(
        doc_id=docs_id, files=files, attachment_types=attachment_types, values=values, file_ids=file_ids, db=db
    )

    # 문서 수정
    updated_docs = self.document_repository.update_user_documents(
        docs_id=docs_id, user_document_update=user_document_update, db=db
    )

    # 스키마 반환
    return UserDocumentsResponse.from_model(updated_docs)

---

데코레이터 방식과 Depends 방식 비교

---

결론

두 방법 모두 트랜잭션 관리 문제를 해결할 수 있다. 데코레이터 방식은 코드가 간결하고, 특정 서비스 함수에 초점을 맞출 때 유용하다. 반면, Depends 방식은 FastAPI와 자연스럽게 통합되며 확장성과 테스트 편의성이 더 높다.

이번에는 데코레이터 방식으로 문제를 해결했지만, 다음 프로젝트에서는 Depends 방식을 활용해 FastAPI의 의존성 주입 시스템을 더 깊이 활용해볼 계획이다.

참고 자료

• FastAPI 의존성 주입 공식 문서
• Python 데코레이터 공식 문서

---

---

추가개념

세션이란?

1. 세션 생성

   세션은 데이터베이스 연결을 관리하는 엔진(Engine)을 기반으로 생성된다. SQLAlchemy에서는 SessionLocal 또는 sessionmaker를 사용해 세션을 생성한다.

   from sqlalchemy.orm import sessionmaker
   from sqlalchemy import create_engine
   
   engine = create_engine("sqlite:///example.db")
   SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

2. 세션 사용

   생성된 세션은 데이터베이스와 상호작용하는 데 사용된다. 작업이 끝나면 세션을 닫아야 한다.

   session = SessionLocal()
   try:
       user = session.query(User).filter(User.id == 1).first()
       user.name = "Updated Name"
       session.commit()
   except Exception:
       session.rollback()
   finally:
       session.close()

3. 컨텍스트 관리

   Python의 with 문을 사용해 세션 관리를 간단히 할 수도 있다.

   with SessionLocal() as session:
       user = session.query(User).filter(User.id == 1).first()
       user.name = "Updated Name"
       session.commit()

---

세션이 중요한 이유

1. 데이터 무결성 보장

   세션은 데이터베이스 작업을 하나의 트랜잭션 단위로 묶어 처리하므로, 작업 도중 오류가 발생해도 데이터의 일관성을 유지할 수 있다.

2. 성능 최적화

   세션은 필요한 데이터를 캐싱하여 같은 요청에 대해 중복 쿼리를 방지한다. 이는 데이터베이스의 부하를 줄이고 성능을 향상시킨다.

3. 지연 로딩 지원

   세션을 통해 객체가 필요로 하는 데이터를 동적으로 로드할 수 있다. 이는 메모리 사용량을 줄이고 초기 로드 시간을 단축하는 데 유용하다.

---

DetachedInstanceError의 원인과 해결

세션이 닫히거나 트랜잭션이 끝나면 SQLAlchemy 객체는 데이터베이스 연결과 분리되며, Detached 상태가 된다. Detached 상태에서 지연 로딩이 필요할 경우 오류가 발생한다.

원인

• 세션이 닫힌 후 지연 로딩(Lazy Loading)을 시도할 때 발생한다.
• 트랜잭션 범위가 명확하지 않거나, 객체가 올바르게 관리되지 않을 때 문제가 생긴다.

해결 방법

1. 세션 범위 명시

   세션의 생명 주기를 명확히 정의하고, 서비스 계층 전체에서 같은 세션을 유지하도록 관리한다.

2. 지연 로딩 제거

   지연 로딩 대신, 필요한 데이터를 미리 로드(Eager Loading)한다.

   from sqlalchemy.orm import joinedload
   
   user = session.query(User).options(joinedload(User.addresses)).filter(User.id == 1).first()

3. FastAPI 의존성 사용

   FastAPI의 Depends를 활용해 세션을 주입하고 트랜잭션을 관리한다.

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

---

결론

세션은 애플리케이션이 데이터베이스와 상호작용하는 데 필수적인 역할을 한다. 올바르게 관리된 세션은 데이터 무결성을 보장하고, 오류를 최소화하며, 데이터베이스 작업을 효율적으로 처리할 수 있게 한다.

FastAPI에서 세션 관리를 위해 데코레이터 방식과 Depends 방식을 사용해볼 수 있으며, 각 방식은 프로젝트의 규모와 요구사항에 따라 선택해야 한다.