2024-03-11 17:45

Pydantic 특정 컬럼 DTO 변환하기

Pydantic 은 SQLAlchemy 와 같은 ORM과 연계하여 객체를 DTO로 자동변환해주는 기능을 제공한다.

참고: https://daco2020.tistory.com/794

문제

객체 전체를 조회해오는 것이 아니라, 특정 컬럼만 가져오거나, 계산 함수를 적용한 필드 등을 DTO로 가져오고 싶을 땐 어떻게 해야할까?

예시

from pydantic import BaseModel, ConfigDict

class UserORM(BaseModel):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)

class UserCountDTO(BaseModel):
    count: int

	model_config: ConfigDict = ConfigDict(from_attributes=True)

위 코드처럼 User 테이블에서 총 유저 수를 가져와 UserCountDTO 로 변환해야 한다고 하자. (예제에서는 굳이 변환하지 않아도 되지만 복잡하다고 가정)

SQLALCHEMY_DATABASE_URL = "sqlite:///./myapi.db"

async_engine = create_async_engine(
    SQLALCHEMY_DATABASE_URL
)
AsyncSessionLocal = sessionmaker(async_engine)

async def get_user_count() -> UserCountDTO:
	with AsyncSessionLocal() as session:
	
		query = select(func.count(User.id).label('count'))
				.selece_from(User)
	
		result = await session.execute(query)

위와 같이 총 유저 수를 가져와 result 에 담고 return 할 일만 남았다.

1. 객체처럼 처리하기

결과

return result.scalars().all() 을 통해서 반환하고, Pydantic 을 통해 DTO로 변환시도
model_attributes_type 에러 발생

설명

scalars() 메서드는 단일 컬럼의 값을 가져오는 데 사용한다.
객체를 조회하는 경우 result 에 실제 object 들이 들어있기 때문에 scalars() 로 가져올 경우 잘 가져올 수 있다.
그러나, 객체와 상관없이 여러 열을 가져오고자 하는 경우 scalars 메서드는 유용하지 않다.

2. 튜플로 가져와서 처리하기

결과

return result.tuples().all() 을 통해서 반환하고, Pydantic 을 통해 DTO로 변환시도
정상적으로 변환

설명

전체를 튜플로 조회하여 해당 컬럼명과 DTO의 필드명을 매핑하여 처리해준다.
따라서, 컬럼명과 필드명이 일치하지 않으면 ValidationError 가 발생한다.
가져오는 컬럼이 더 많은 것은 상관없다.

추가. result.all() 만 사용해도 동작한다.

하지만, result.all()은 Row 의 Sequence 를 반환하고 result.tuples().all() 은 TupleResult 를 반환한다.
따라서, result.tuples().all() 을 사용해야 추가적인 페이징이나 필터 기능을 사용할 수 있다.

공식문서: https://docs.sqlalchemy.org/en/20/core/connections.html#sqlalchemy.engine.Result

Pydantic

SQLAlchemy