9/16 SQLAlchemy

어제 비동기 실행 흐름을 살펴본 데 이어, 오늘은 SQLAlchemy로 데이터베이스를 다뤘다. PostgreSQL에 회원 테이블을 만들고 추가·조회·수정·삭제하는 수업 실습이었다.

같은 작업을 SQL로 직접 작성하는 방식과 SQLAlchemy 구문으로 작성하는 방식으로 비교했다. 조회한 결과를 Pydantic 모델로 바꾸는 과정도 함께 다뤘다.

테이블은 Member, 데이터는 MemberVO

먼저 Member 클래스로 members 테이블 구조를 정의했다. 회원 ID, 이메일, 비밀번호, 이름, 나이, 가입일을 컬럼으로 두었다.

아래는 테이블 정의 코드의 일부다.

class Member(Base):
    __tablename__ = "members"

    id: Mapped[int] = mapped_column(
        BigInteger, Identity(), primary_key=True
    )
    member_email: Mapped[str] = mapped_column(
        String(255), nullable=False, unique=True
    )
    member_age: Mapped[int | None] = mapped_column(Integer)

이메일에는 중복과 NULL을 허용하지 않는 조건을 넣고, 나이는 선택 항목으로 정의했다. 이렇게 클래스에 컬럼과 제약 조건을 적고, Base.metadata.create_all로 테이블을 생성하는 흐름이었다.

별도로 만든 MemberVO는 Pydantic 모델이다. DB 테이블과 연결된 Member와 달리, 함수에 전달하거나 조회 결과를 정리할 데이터 형태를 정의했다.

member_dict = new_member.model_dump(exclude_none=True)
new_member = MemberVO.model_validate(member_dict)

model_dump()는 모델을 딕셔너리로 바꾸고, model_validate()는 전달한 데이터를 검증해 모델로 만든다. 실습에서는 exclude_none=True로 값이 None인 항목을 제외하는 것도 확인했다.

MemberVO에 넣은 ConfigDict(from_attributes=True)는 ORM 객체의 속성을 읽어 모델로 변환할 수 있게 하는 설정이다. 딕셔너리로 내보내는 기능 자체와는 역할이 다르다. Pydantic 모델 문서

SQL을 직접 쓰는 방식과 비교하기

회원 조회에서는 먼저 SQL 문자열을 text()로 작성했다.

FIND_MEMBER_BY_ID_QUERY = text("""
    select
        id, member_email, member_password,
        member_name, member_age, member_created_at
    from members
    where id = :id
""")

같은 조회 조건을 SQLAlchemy 구문으로 표현하면 다음처럼 작성할 수 있었다.

query = select(Member).where(Member.id == id)
result = await session.execute(query)
found_member = result.scalar_one_or_none()

오늘 실습에서는 AsyncSession을 사용해 await session.execute()로 실행했다. 추가·수정·삭제 코드에는 await session.commit()을 넣어 변경 내용을 확정했다.

insert(Member), update(Member), delete(Member)도 같은 방식으로 다뤘다. SQL 문자열을 직접 적을 때와 비교하면서, 조건과 값을 Python 코드로 표현하는 흐름을 볼 수 있었다.

조회 결과를 어떻게 꺼낼까

쿼리를 실행한 뒤에는 결과 형태에 맞춰 값을 꺼내야 했다.

실습에서 여러 컬럼을 직접 조회한 SQL 결과는 mappings()로 키와 값 형태로 받았다. select(Member)로 조회할 때는 scalar_one_or_none()이나 scalars()로 Member 객체를 꺼냈다.

# 여러 컬럼을 조회한 SQL 결과
found_member = result.mappings().one_or_none()

# select(Member)로 조회한 결과
found_member = result.scalar_one_or_none()

위 두 줄은 각각의 조회 방식에서 사용한 코드다. scalar_one_or_none()은 결과가 없으면 None, 한 개면 해당 값을 반환하고, 여러 개면 예외를 발생시킨다.

전체 조회에서는 다음처럼 객체 목록을 꺼낸 뒤 MemberVO로 변환했다.

result = await session.execute(select(Member))
members = result.scalars().all()

return [MemberVO.model_validate(member) for member in members]

all()만 사용하면 행 목록을 받고, 이 예제처럼 scalars().all()을 사용하면 각 행에 담긴 Member 객체의 목록을 받는다. 같은 조회라도 결과를 꺼내는 방식이 다르다는 점을 정리했다. SQLAlchemy 조회 문서

수정·삭제 결과 확인하기

수정과 삭제 함수는 실행 결과의 rowcount가 0보다 큰지 반환하도록 작성되어 있었다.

result = await session.execute(query)
await session.commit()
return result.rowcount > 0

노트북에는 수정 성공, 삭제 실패, 삭제 성공 출력이 남아 있다. 여기서 삭제 실패 문구는 해당 실행에서 rowcount > 0 조건이 거짓이었다는 뜻이다. 이 출력만으로 DB 연결 오류가 있었다고 볼 수는 없다.

수정 코드는 복습할 부분도 있다. 현재는 model_dump(exclude_none=True) 결과를 그대로 .values()에 넣기 때문에 모델에 담긴 id도 수정 대상에 포함될 수 있다. 수정할 회원을 찾는 ID와 변경할 필드가 섞이지 않도록 다시 살펴볼 예정이다.

오늘 정리한 흐름

오늘은 테이블 구조를 정의하고, 세션으로 쿼리를 실행하고, 결과를 필요한 데이터 형태로 변환하는 과정까지 이어서 실습했다. Member와 MemberVO, 그리고 조회 결과인 행 객체를 구분하는 것이 핵심이었다.

다음에는 단건 조회와 전체 조회의 반환값을 다시 비교하고, 수정 함수에서 ID와 변경 데이터를 나누는 부분을 복습하려고 한다.