FastAPI, SQLAlchemy에서 조인된 테이블 쿼리 시 발생하는 TypeError 이해 및 해결 방법

2024-07-27

FastAPI와 SQLAlchemy를 사용하여 조인된 테이블을 쿼리할 때 다음과 같은 오류가 발생할 수 있습니다.

TypeError: vars() argument must have __dict__ attribute when querying joined tables

오류 원인:

이 오류는 SQLAlchemy의 jsonable_encoder 함수에서 발생하며, 조인된 테이블에서 반환된 객체에 __dict__ 속성이 없기 때문입니다. jsonable_encoder 함수는 응답 객체를 JSON 형식으로 변환하는 데 사용되며, 이 과정에서 객체의 __dict__ 속성에 접근하여 속성 이름과 값을 쌍으로 변환합니다. 하지만 조인된 테이블의 경우, 반환된 객체는 일반적인 SQLAlchemy 모델 객체가 아니며 __dict__ 속성이 없을 수 있습니다.

해결 방법:

이 문제를 해결하려면 다음과 같은 방법을 사용할 수 있습니다.

Pydantic 모델 사용:

Pydantic 모델을 사용하여 조인된 테이블의 결과를 표현하면 jsonable_encoder 함수가 모델의 __fields__ 속성을 사용하여 객체를 JSON으로 변환하므로 오류가 발생하지 않습니다.

from pydantic import BaseModel

class JoinedResult(BaseModel):
    # 조인된 테이블의 모든 컬럼을 포함

# ...

# 쿼리 실행 및 결과 가져오기
results = session.query(User, Address).join(Address, User.address_id).all()

# Pydantic 모델로 변환
joined_results = [JoinedResult(**result) for result in results]

# JSON으로 응답
return json.dumps(joined_results)

직접 딕셔너리 변환:

각 조인된 결과를 직접 딕셔너리로 변환하여 jsonable_encoder 함수에 전달할 수도 있습니다.

# ...

# 쿼리 실행 및 결과 가져오기
results = session.query(User, Address).join(Address, User.address_id).all()

# 딕셔너리로 변환
joined_results = [dict(row) for row in results]

# JSON으로 응답
return json.dumps(joined_results)

SQLAlchemy의 to_dict() 메서드 사용:

SQLAlchemy의 to_dict() 메서드를 사용하여 조인된 결과를 딕셔너리로 변환할 수 있습니다.

# ...

# 쿼리 실행 및 결과 가져오기
results = session.query(User, Address).join(Address, User.address_id).all()

# to_dict() 메서드 사용하여 딕셔너리 변환
joined_results = [row.to_dict() for row in results]

# JSON으로 응답
return json.dumps(joined_results)

주의 사항:

Pydantic 모델을 사용하는 경우, 모델에 정의된 필드만 JSON 응답에 포함됩니다. 필요한 모든 컬럼을 포함하도록 모델을 정의해야 합니다.
직접 딕셔너리로 변환하는 경우, 결과 딕셔너리에 포함될 키 이름을 명확하게 지정해야 합니다.
to_dict() 메서드를 사용하는 경우, SQLAlchemy 버전에 따라 동작 방식이 다를 수 있습니다. 사용 중인 SQLAlchemy 버전의 문서를 참고하십시오.

이 외에도, orm_mode 옵션을 사용하거나 QueryProperty 클래스를 사용하는 등 다양한 해결 방법이 있습니다.

예제 코드: FastAPI, SQLAlchemy에서 조인된 테이블 쿼리하기

from fastapi import FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from pydantic import BaseModel

# 데이터베이스 엔진 생성
engine = create_engine("postgresql://user:password@host:port/database")

# 세션 메이커 생성
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# FastAPI 애플리케이션 생성
app = FastAPI()


# Pydantic 모델 정의
class User(BaseModel):
    id: int
    name: str
    email: str


class Address(BaseModel):
    id: int
    user_id: int
    street: str
    city: str
    state: str
    zip_code: str


class JoinedResult(BaseModel):
    user: User
    address: Address


# 조인된 테이블 쿼리 함수
def get_joined_results():
    with Session() as session:
        results = session.query(User, Address).join(Address, User.address_id).all()

        # Pydantic 모델로 변환
        joined_results = [JoinedResult(user=User(**user_data), address=Address(**address_data))
                         for user_data, address_data in results]

        return joined_results


# 엔드포인트 정의
@app.get("/joined-results")
def get_results():
    results = get_joined_results()
    return results

설명:

User 및 Address Pydantic 모델은 각 테이블의 컬럼을 정의합니다.
JoinedResult Pydantic 모델은 조인된 결과를 나타내는 데 사용됩니다.
get_joined_results() 함수는 조인된 테이블을 쿼리하고 결과를 JoinedResult 모델 목록으로 변환합니다.
/joined-results 엔드포인트는 get_joined_results() 함수를 호출하고 결과를 JSON 형식으로 반환합니다.

이 예제는 기본적인 구조를 보여주는 것으로, 실제 상황에서는 테이블 이름, 컬럼 이름, 데이터베이스 연결 정보 등을 변경해야 할 수 있습니다. 또한, 오류 처리, 인증 및 권한 부여 등의 기능을 추가해야 할 수도 있습니다.

참고:

이 코드는 SQLAlchemy 1.4 및 Python 3.8을 기준으로 작성되었습니다.

FastAPI, SQLAlchemy에서 조인된 테이블 쿼리 시 발생하는 TypeError 해결 방법: 대체 방법

query.statement 사용:

SQLAlchemy의 query.statement 속성을 사용하여 쿼리 객체의 SQL 문을 직접 가져와 실행할 수 있습니다. 이렇게 하면 원하는 형식으로 결과를 변환하는 데 더 많은 제어권을 가질 수 있습니다.

def get_joined_results():
    with Session() as session:
        # 쿼리 객체 생성
        query = session.query(User, Address).join(Address, User.address_id)

        # SQL 문 가져오기
        sql = query.statement()

        # 결과 실행 및 변환
        results = session.execute(sql).fetchall()
        joined_results = []
        for row in results:
            user_data, address_data = row
            joined_results.append(JoinedResult(user=User(**user_data), address=Address(**address_data)))

        return joined_results

@sqlalchemy.ext.declarative.as_declarative 데코레이터 사용:

@sqlalchemy.ext.declarative.as_declarative 데코레이터를 사용하여 조인된 결과를 나타내는 새 모델 클래스를 정의할 수 있습니다. 이 모델 클래스는 Pydantic 모델과 유사하지만, SQLAlchemy의 __dict__ 속성을 가지고 있어 jsonable_encoder 함수에서 문제 없이 변환될 수 있습니다.

from sqlalchemy.ext.declarative import as_declarative
from sqlalchemy.orm import joinedload

@as_declarative()
class JoinedResultBase:
    pass


class User(JoinedResultBase, Base):
    # ...

class Address(JoinedResultBase, Base):
    # ...


def get_joined_results():
    with Session() as session:
        # 조인된 결과를 포함하는 JoinedResult 모델 생성
        JoinedResult = joinedload(User, Address).aliased('joined_result')

        # 쿼리 실행 및 결과 가져오기
        results = session.query(JoinedResult).all()

        return results

SQLAlchemy-Utils 라이브러리 사용:

SQLAlchemy-Utils 라이브러리는 다양한 유용한 도구를 제공하며, 그 중 하나는 조인된 테이블의 결과를 쉽게 변환하는 데 사용할 수 있는 CommaJoined 클래스입니다.

from sqlalchemy_utils import CommaJoined

def get_joined_results():
    with Session() as session:
        # CommaJoined 클래스 사용하여 조인된 결과 변환
        joined_results = session.query(User, Address).join(Address, User.address_id).all()
        results = [CommaJoined(user=user, address=address) for user, address in joined_results]

        return results

각 방법마다 장단점이 있으므로, 상황에 맞는 방법을 선택해야 합니다.
query.statement 사용 방법은 더 많은 제어권을 제공하지만, SQL 문을 직접 다루어야 하기 때문에 복잡할 수 있습니다.
@sqlalchemy.ext.declarative.as_declarative 데코레이터 사용 방법은 Pydantic 모델과 유사한 모델 클래스를 제공하지만, SQLAlchemy 버전에 따라 호환되지 않을 수 있습니다.
SQLAlchemy-Utils 라이브러리는 유용한 도구이지만, 별도로 설치해야 합니다.

python postgresql sqlalchemy