SQLAlchemy Select 심화: Result와 Row 객체 파헤치기

오늘은 SQLAlchemy ORM의 select 문을 기본적인 사용법을 넘어 조금 더 깊이 있게 다뤄보겠습니다. 특히 Session.execute() 메서드의 반환값과 그 내부 구조에 초점을 맞추어, 쿼리 결과를 효과적으로 다루는 방법을 탐구할 것입니다.

기본 설정 및 예제 데이터 생성

먼저 예제 실행을 위한 기본적인 SQLAlchemy 설정을 살펴보겠습니다. DeclarativeBase를 상속받는 Base 클래스를 정의하고, User와 Address 모델을 매핑합니다. 데이터베이스는 메모리상의 SQLite를 사용합니다.

from sqlalchemy import create_engine

# 메모리 내 SQLite 데이터베이스 엔진 생성 (echo=True로 SQL 쿼리 로깅 활성화)
engine = create_engine("sqlite+pysqlite:///:memory:", echo=True)

from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session

class Base(DeclarativeBase):
    """
    DeclarativeBase를 상속받아 ORM 매핑의 기본 클래스로 사용합니다.
    이 Base를 상속하는 클래스는 데이터베이스 테이블과 매핑됩니다.
    __tablename__ 속성을 반드시 정의해야 합니다.
    """
    pass

class User(Base):
    __tablename__ = "user_account" # 테이블명 정의

    # 컬럼 정의 (Mapped와 mapped_column 사용)
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]] # Optional 타입 매핑
    # Address 모델과의 관계 설정 (일대다)
    addresses: Mapped[List["Address"]] = relationship(
        back_populates="user", cascade="all, delete-orphan"
    )
    def __repr__(self) -> str:
        # 객체 표현 방식 정의
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

class Address(Base):
    __tablename__ = "address" # 테이블명 정의

    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    # 외래키 설정 (user_account 테이블의 id 참조)
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    # User 모델과의 관계 설정 (다대일)
    user: Mapped["User"] = relationship(back_populates="addresses")
    def __repr__(self) -> str:
        return f"Address(id={self.id!r}, email_address={self.email_address!r})"

# 정의된 모델을 기반으로 데이터베이스 테이블 생성
Base.metadata.create_all(engine)

이제 예제 데이터를 생성하는 함수를 정의하고 실행하여 데이터베이스에 100명의 사용자 정보를 추가합니다.

from sqlalchemy.orm import Session

def generate_100_users():
    """100명의 User와 연관된 Address 데이터를 생성하여 데이터베이스에 추가합니다."""
    with Session(engine) as session:
        for i in range(100):
            session.add(User(name=f"user{i}", fullname=f"User {i}", addresses=[Address(email_address=f"user{i}@example.com")]))
        session.flush() # 변경사항을 데이터베이스 세션에 반영 (커밋 전)
        session.commit() # 트랜잭션 커밋

generate_100_users()

Result 와 Row 객체: 쿼리 결과의 핵심

SQLAlchemy 2.0 버전부터 ORM에서 쿼리를 실행하는 주요 방법은 Session.execute() 메서드를 사용하는 것입니다. 공식 문서에 따르면 이 메서드는 Result 객체를 반환한다고 명시되어 있습니다. 그렇다면 이 Result 객체는 정확히 무엇일까요?

from sqlalchemy import select

with Session(engine) as session:
    stmt = select(User) # User 테이블 전체를 조회하는 select 문 생성
    users_result = session.execute(stmt) # select 문 실행
    print(users_result) # 실행 결과 객체 출력
    print(type(users_result)) # 결과 객체의 타입 출력

위 코드를 실행하면 실제로는 ChunkedIteratorResult 타입의 객체가 반환되는 것을 확인할 수 있습니다. SQLAlchemy의 내부 코드를 살펴보면 Result 클래스 자체는 여러 메서드가 NotImplementedError를 발생시키는, 일종의 인터페이스(Interface) 역할을 하고 있음을 알 수 있습니다.

# SQLAlchemy 내부 코드 예시 (간략화)
class Result(_WithKeys, ResultInternal[Row[Unpack[_Ts]]]):
    """데이터베이스 결과 집합을 나타냅니다.
    ORM 사용 시에는 일반적으로 ChunkedIteratorResult 객체가 사용됩니다.
    """
    @property
    def _soft_closed(self) -> bool:
        raise NotImplementedError()

    @property
    def closed(self) -> bool:
        raise NotImplementedError()
    # ... 다른 메서드들 ...

즉, Session.execute()는 실제 구현체인 ChunkedIteratorResult 객체를 반환하며, 이 객체는 Result 인터페이스를 구현합니다. 따라서 우리는 Result 객체가 제공하는 다양한 메서드를 통해 쿼리 결과를 다룰 수 있습니다.

Result 객체는 다음과 같은 여러 유용한 메서드를 제공합니다.

all(), close(), closed, columns(), fetchall(), fetchmany(), fetchone(), first(), freeze(), keys(), mappings(), merge(), one(), one_or_none(), partitions(), scalar(), scalar_one(), scalar_one_or_none(), scalars(), t, tuples(), unique(), yield_per()

이번 포스트에서는 이 중 몇 가지 주요 메서드, 특히 데이터를 가져오는 메서드들을 중심으로 살펴보겠습니다.

Result.all() 메서드 심층 분석

가장 흔히 사용되는 메서드 중 하나인 all()은 쿼리 결과의 모든 행(row)을 리스트 형태로 반환합니다. 내부적으로 어떻게 동작하는지 SQLAlchemy 코드를 통해 간단히 살펴보겠습니다. (주석은 설명을 위해 추가되었습니다)

# Result 클래스의 _allrows 메서드 (all() 내부에서 호출됨, 간략화)
def _allrows(self) -> List[_R]:
    # 후처리 필터: 결과를 최종 형태로 가공하는 내부 로직
    post_creational_filter = self._post_creational_filter

    # 행(row) 생성 함수: 가져온 데이터를 Row 객체 등으로 변환
    make_row = self._row_getter

    # 데이터베이스로부터 모든 데이터 가져오기
    rows = self._fetchall_impl()
    made_rows: List[_InterimRowType[_R]]

    # 가져온 데이터를 Row 객체 또는 지정된 타입으로 변환
    if make_row:
        made_rows = [make_row(row) for row in rows]
    else:
        made_rows = rows

    interim_rows: List[_R]

    # unique 필터 적용: strategy에 따라 중복 제거 (set 사용)
    if self._unique_filter_state:
        uniques, strategy = self._unique_strategy
        interim_rows = [
            # strategy 함수를 적용하여 고유 식별자를 얻고, uniques set에 없는 경우만 포함
            made_row
            for made_row, sig_row in [
                (made_row, strategy(made_row) if strategy else made_row)
                for made_row in made_rows
            ]
            if sig_row not in uniques and not uniques.add(sig_row)
        ]
    else:
        interim_rows = made_rows

    # 후처리 필터 적용
    if post_creational_filter:
        interim_rows = [post_creational_filter(row) for row in interim_rows]

    return interim_rows # 최종 결과 리스트 반환

코드를 분석해보면 all() 메서드는 단순히 모든 데이터를 가져오는 것 외에도 다음과 같은 과정을 포함합니다.

1. 데이터 인출 (_fetchall_impl): 데이터베이스로부터 결과 행들을 가져옵니다.

2. 행 객체 생성 (make_row): 가져온 데이터를 SQLAlchemy의 Row 객체나 다른 지정된 형태로 변환합니다.

3. 고유값 처리 (_unique_filter_state): unique() 메서드가 호출된 경우, 제공된 strategy 함수를 사용하여 중복된 행을 제거합니다. strategy는 각 행에서 고유성을 판단할 값을 반환하는 함수이며, 파이썬의 set을 이용하여 중복을 검사합니다.

4. 후처리 (post_creational_filter): 내부적인 추가 처리를 통해 최종 반환 형태를 만듭니다.

이렇게 all() 메서드는 내부적으로 여러 단계를 거쳐 최종 결과 리스트를 반환합니다.

from sqlalchemy import select

with Session(engine) as session:
    # User 테이블에서 name과 fullname 컬럼만 선택
    stmt = select(User.name, User.fullname)
    # 쿼리 실행 후 모든 결과를 리스트로 가져옴
    users = session.execute(stmt).all()
    print(type(users)) # <class 'list'>
    # 리스트의 첫 번째 요소 (Row 객체)에 접근하여 'name' 속성 출력
    print(users[0]["name"]) # Row 객체는 키(컬럼명)로 접근 가능
    print(users[0].name)   # Row 객체는 속성처럼 접근 가능

all() 메서드는 Row 객체들의 리스트를 반환합니다. Row 객체는 마치 이름 있는 튜플(named tuple)처럼 동작하여, 컬럼 이름(user.name)이나 인덱스(user[0]) 또는 키(user["name"])로 값에 접근할 수 있습니다.

중복 데이터 처리: unique() 메서드 활용

쿼리 결과에서 중복을 제거해야 하는 경우가 있습니다. 예를 들어 특정 조건을 만족하는 사용자를 조회했는데, 이름이 같은 사용자가 여러 명 포함될 수 있습니다.

먼저, 이름이 'user1'인 사용자를 추가하여 중복 상황을 만들어 보겠습니다.

# 예시: 중복 데이터 확인
with Session(engine) as session:
    # 기존 'user1' 조회
    stmt_before = select(User).where(User.name == "user1")
    users_before = session.execute(stmt_before).all()
    print(f"중복 추가 전 'user1' 수: {len(users_before)}")
    if users_before:
        print(f"기존 'user1' 정보: {users_before[0][0]}") # Row 객체 안의 User 객체 접근

# 이름이 'user1'인 사용자 추가 (ID는 다름)
with Session(engine) as session:
    duplicated_user = User(id=101, name="user1", fullname="user1_duplicated", addresses=[Address(email_address="duplicated_user@gmail.com")])
    session.add(duplicated_user)
    session.commit()

# 중복 추가 후 'user1' 조회
with Session(engine) as session:
    stmt_after = select(User).where(User.name == "user1")
    users_after = session.execute(stmt_after).all()
    print(f"중복 추가 후 'user1' 수: {len(users_after)}")
    print(f"조회된 'user1' 사용자 리스트: {users_after}")

이제 이름('user1')을 기준으로 중복을 제거해 보겠습니다. Result 객체의 unique() 메서드를 사용하고, strategy 인자에 어떤 기준으로 고유성을 판단할지 정의하는 함수(lambda)를 전달합니다.

with Session(engine) as session:
    stmt = select(User).where(User.name == "user1")
    # unique() 메서드로 중복 제거. strategy는 User 객체의 name 속성을 기준으로 함.
    unique_users = session.execute(stmt).unique(strategy=lambda row: row[0].name).all()
    print(f"unique() 적용 후 'user1' 수: {len(unique_users)}") # 중복 제거 확인

unique() 메서드는 all()과 같은 다른 결과 추출 메서드( first(), scalars() 등) 전에 호출되어야 합니다. strategy 함수는 Row 객체를 인자로 받으므로, Row 객체의 구조를 이해하고 올바른 접근 방식을 사용해야 합니다 (이후 Row 섹션에서 자세히 설명).

Row 객체와 접근 방식의 이해

앞선 unique() 예제에서 lambda row: row[0].name 와 같은 접근 방식을 사용했습니다. 어떤 경우에는 row.name처럼 바로 속성에 접근할 수 있는데, 왜 여기서는 row[0]을 거쳐야 했을까요? 이 차이를 이해하려면 Row 객체의 동작 방식과 select() 구문에 따른 변화를 알아야 합니다.

Row 객체는 기본적으로 파이썬의 namedtuple과 유사하게 설계되었습니다. namedtuple은 튜플처럼 불변(immutable)성을 가지면서도 각 요소에 이름을 부여하여 속성처럼 접근할 수 있게 해주는 편리한 자료구조입니다.

from collections import namedtuple

# 클래스를 사용한 데이터 표현
class BookClass:
    def __init__(self, title, code):
        self.title = title
        self.code = code
book_obj = BookClass(title="roach", code="1")
print(f"클래스 객체: {book_obj.title}")

# 튜플을 사용한 데이터 표현 (인덱스로 접근)
book_tuple = ("roach", "1")
print(f"튜플: {book_tuple[0]}")

# namedtuple을 사용한 데이터 표현 (이름 또는 인덱스로 접근 가능)
BookNamedTuple = namedtuple('Book', ['title', 'code'])
book_namedtuple = BookNamedTuple("roach", "1")
print(f"namedtuple (속성 접근): {book_namedtuple.title}")
print(f"namedtuple (인덱스 접근): {book_namedtuple[0]}")

Row 객체도 이와 비슷하게, 인덱스(row[0]), 키/컬럼명(row["name"]), 그리고 속성(row.name) 방식으로 값에 접근할 수 있습니다.

핵심은 select() 구문에 무엇을 전달했는지에 따라 Row 객체의 구조가 달라진다는 점입니다.

1. select(User.name, User.fullname): 특정 컬럼들을 명시적으로 선택한 경우

   • Result 객체의 메타데이터(ResultMetadata)에는 선택된 컬럼 이름들(['name', 'fullname'])이 keys로 저장됩니다.

   • 반환된 Row 객체는 이 keys를 기반으로 구조화됩니다.

   • 따라서 row.name 또는 row[0] (첫 번째 선택 컬럼), row.fullname 또는 row[1] (두 번째 선택 컬럼) 방식으로 직접 접근할 수 있습니다.

    from sqlalchemy import select

    with Session(engine) as session:
        # name, fullname 컬럼만 선택
        stmt = select(User.name, User.fullname).where(User.name == "user1")
        users = session.execute(stmt).all()
        print(f"\nselect(User.name, User.fullname) 결과 타입: {type(users)}")
        for user_row in users:
            print(f"Row 객체 타입: {type(user_row)}")
            # 속성 접근
            print(f"  속성 접근: user_row.name = {user_row.name}, user_row.fullname = {user_row.fullname}")
            # 인덱스 접근
            print(f"  인덱스 접근: user_row[0] = {user_row[0]}, user_row[1] = {user_row[1]}")
            # 키 접근
            print(f"  키 접근: user_row['name'] = {user_row['name']}, user_row['fullname'] = {user_row['fullname']}")

2. select(User): ORM 모델 객체 자체를 선택한 경우

   • Result 객체의 메타데이터에는 모델 클래스 이름(['User'])이 keys로 저장됩니다.

   • 반환된 Row 객체의 첫 번째 요소(row[0])가 바로 User 객체 자체가 됩니다. Row 객체는 이 User 객체를 감싸고 있는 형태입니다.

   • 따라서 User 객체의 속성에 접근하려면 row[0].name, row[0].fullname과 같이 인덱스로 먼저 User 객체에 접근한 후 속성에 접근해야 합니다.

   • 또는 Row 객체가 키로도 접근을 지원하므로, row.User.name (키 User로 User 객체 접근 후 속성 접근) 방식으로도 가능합니다.

    with Session(engine) as session:
        # User 객체 전체를 선택
        stmt = select(User).where(User.name == "user1")
        users_result = session.execute(stmt) # Result 객체
        print(f"\nselect(User) Result keys: {users_result.keys()}") # ['User']

        users = users_result.all()
        print(f"select(User) 결과 타입: {type(users)}")
        for user_row in users:
            print(f"Row 객체 타입: {type(user_row)}")
            # 인덱스 접근 후 속성 접근
            user_obj = user_row[0] # User 객체
            print(f"  인덱스 접근: type(user_row[0]) = {type(user_obj)}")
            print(f"  인덱스 접근 -> 속성: user_row[0].name = {user_obj.name}")
            # 키/속성 접근 후 속성 접근
            print(f"  키/속성 접근 -> 속성: user_row.User.name = {user_row.User.name}") # row.User 로 User 객체 접근 가능

이제 아까 unique() 메서드의 strategy에서 lambda row: row[0].name 또는 lambda row: row.User.name을 사용해야 했던 이유가 명확해졌습니다. select(User)로 조회했기 때문에 Row 객체에서 User 객체(row[0] 또는 row.User)를 먼저 얻어온 다음 name 속성에 접근해야 했던 것입니다.

# select(User) 결과에 대해 .User 속성을 사용한 unique 처리
with Session(engine) as session:
    stmt = select(User).where(User.name == "user1")
    # strategy에서 row.User.name 사용
    users = session.execute(stmt).unique(strategy=lambda row: row.User.name).all()
    print(f"\nselect(User) + unique(strategy=lambda row: row.User.name) 결과 수: {len(users)}")

이처럼 select() 구문에 무엇을 전달하느냐에 따라 Row 객체의 구조와 접근 방식이 달라지므로, 코드를 작성할 때 이를 명확히 인지하는 것이 중요합니다.

결론

이번 포스트에서는 SQLAlchemy ORM의 select 쿼리 실행 결과인 Result 객체와 그 안의 Row 객체에 대해 심도 있게 알아보았습니다.

• Session.execute()는 ChunkedIteratorResult ( Result 인터페이스 구현체) 객체를 반환합니다.

• Result 객체는 all(), unique() 등 다양한 메서드를 통해 쿼리 결과를 가공하고 추출하는 기능을 제공합니다.

• Row 객체는 namedtuple과 유사하게 동작하며, 인덱스, 키(컬럼명), 속성 방식으로 데이터에 접근할 수 있습니다.

• select() 구문에 어떤 요소를 전달하는지에 따라 (select(User.name) vs select(User)) Row 객체의 내부 구조와 접근 방식이 달라진다는 점을 유의해야 합니다.

Result와 Row 객체의 동작 방식을 이해하면 SQLAlchemy를 더욱 효과적으로 활용하고, 다양한 상황에 맞는 최적의 데이터 접근 코드를 작성하는 데 큰 도움이 될 것입니다.