Repository는 도메인 객체를 주고받지만, 실제로 데이터를 읽고 쓰기 위해서는 도메인 객체를 데이터베이스 스키마에 대응되는 타입으로 변환하는 작업이 필요하다(ref1:DB 레코드와 도메인 객체의 불일치는 실제로 자주 일어나는 일이다).
이런 상황에서 사용되는 것이 Mapper와 DTO라는 개념이다. 하지만 개인적으로 나처럼 JAVA 경험 없이 python에만 익숙한 사람이라면 Mapper과 DTO와 관련 개념이 와닿지 않고 헷갈리기 쉽다고 생각한다. 이 글이 말하고자 하는 바는 ORM을 사용하면 Mapper과 DTO를 작성하는 초기에 조금 줄일 수 있다는 내용이다. 이 말을 이해하기 어렵다고 구현 시 무작정 ORM을 채택하기보다는 등장배경과 이론적 완성형을 이해하려 노력해본 다음에 구현 시 생략해도 괜찮을지 판단하는 편이 낫다.
DTO는 Data Transfer Object의 줄임말로, 말 그대로 추상화된 공간 사이(레이어가 될수도 있고, 같은 레이어의 다른 의미 공간이 될 수도 있다. 그래서 보통 경계boundary라고 부른다.)로 데이터를 담아 넘기기(Transfer) 위해 사용하는 규격화된 객체다.
소프트웨어에는 다양한 추상(경계)이 있고, 그 사이사이 공간마다 DTO를 만들 수 있는데, 그중에서도 DB로부터 읽은 값을 객체화한 결과를 특히 Persistence DTO(또는 Java 진영에서는 JPA(Java Persistence API) 엔티티)라고 한다.
사용자의 id로 사용자를 찾아내는 User Repository의 메서드를 구현한다고 해 보자. 이론상 엄격함을 준수한다면 Repository는 아래 코드와 같이 DB에서 읽은 raw 데이터를 Persistence DTO로 변환한 다음, 그것을 Mapper를 이용해서 도메인 객체로 변환한다. Mapper은 말 그대로 객체를 다른 객체로 옮기는 변환기다.
# 어디엔가 정의된 DTO
class UserDTO(...):
id = ...
class SqlUserRepository(UserRepositoryInterface):
def find_by_id(self, user_id: int) -> Optional[User]:
# 1. 데이터베이스 접근
raw_data = self.db.execute(query, (user_id,)).fetchone()
# 2. Raw 데이터를 DTO로 변환
user_dto = UserDTO(
id=raw_data[0],
username=raw_data[1],
email=raw_data[2],
# ...
)
# 3. Mapper를 통해 DTO를 도메인 객체로 변환
return self.mapper.dto_to_domain(user_dto)
Python
복사
(1.)DB —(2.)→ DTO —(3.)Mapper(DTO)→ Domain Object
하지만 이런 방식은 굉장히 복잡하다. 우선 (1)데이터베이스에 쿼리를 날리고 (2)응답값을 객체에 매핑해주는 과정에 보일러플레이트 코드가 생성되고 개발 생산성이 떨어진다. 만약 ORM을 사용한다면 ORM 자체가 이미 DB 레코드를 객체로 매핑해주는 역할을 하기 때문에, ORM 모델 자체를 DTO로 볼 수 있어 (2)를 없앨 수 있다. 그렇다면 (3)DTO를 도메인 객체로 바꾸는 매핑 과정은 어떨까. sqlalchemy와 같은 ORM 라이브러리는 inspect와 같은 함수를 제공한다. 프로젝트 초기에는 DB 모델과 도메인 모델이 일치하는 경우가 많다. inspect를 이용하면 속성명이 동일한 도메인 모델에 DB 모델을 빠르게 매핑하여 (3)보일러플레이트를 줄일 수 있다.
from sqlalchemy import inspect
# SQLAlchemy ORM 모델
class UserModel(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
username = Column(String)
email = Column(String)
full_name = Column(String)
created_at = Column(DateTime)
class SqlUserRepository(UserRepositoryInterface):
def find_by_id(self, user_id: int) -> Optional[User]:
user_model = self.session.query(UserModel).filter_by(id=user_id).first()
if not user_model:
return None
# SQLAlchemy inspect를 이용한 동적 변환
mapper = inspect(UserModel)
user_data = {}
for column in mapper.columns:
user_data[column.name] = getattr(user_model, column.name)
return User(**user_data)
Python
복사
하지만 inspect()를 사용한 변환은 다음과 같은 제약이 있다:
1.
ORM 모델과 도메인 객체의 속성명이 일치해야 함
2.
DB의 JSON 컬럼을 파이썬 객체로 변환해야 하는 경우 등 타입 변환이 어려움
3.
여러 테이블의 조인 결과를 하나의 도메인 객체로 만들어야 하는 경우 등 복합 객체 구성이 어려움
이런 제약조건이 일부 필드에서만 제한적으로 발생한다면, 아래와 같이 작은 ORM(as DTO) → 도메인 매핑 로직을 작성하는 것도 방법이다.
# SQLAlchemy ORM 모델
class UserModel(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
username = Column(String)
email = Column(String)
full_name = Column(String)
created_at = Column(DateTime)
class SqlUserRepository(UserRepositoryInterface):
def find_by_id(self, user_id: int) -> Optional[User]:
user_model = self.session.query(UserModel).filter_by(id=user_id).first()
if not user_model:
return None
# 직접 변환 (간단한 경우)
return User(
id=user_model.id,
username=user_model.username,
email=user_model.email,
full_name=user_model.full_name,
created_at=user_model.created_at
)
Python
복사
명시적인 Mapper은 여기서 로직이 조금 더 복잡해졌을 때 사용하는 것이 더 적절하다. ORM 모델과 도메인 타입 사이의 변환을 Mapper이 온전히 담당하도록 분리한다.
# 복잡한 변환이 필요한 경우의 Mapper
class UserMapper:
"""ORM 모델을 도메인 객체로 변환"""
@staticmethod
def orm_to_domain(user_model: UserModel) -> User:
return User(
id=user_model.id,
username=user_model.username,
email=user_model.email,
full_name=user_model.full_name,
created_at=user_model.created_at,
# 복잡한 변환 로직
profile=json.loads(user_model.profile_json) if user_model.profile_json else None,
status=UserStatus.from_string(user_model.status_code)
)
@staticmethod
def domain_to_orm(user: User) -> UserModel:
return UserModel(
id=user.id,
username=user.username,
email=user.email,
full_name=user.full_name,
created_at=user.created_at,
profile_json=json.dumps(user.profile) if user.profile else None,
status_code=user.status.to_string()
)
class SqlUserRepository(UserRepositoryInterface):
def find_by_id(self, user_id: int) -> Optional[User]:
user_model = self.session.query(UserModel).filter_by(id=user_id).first()
if not user_model:
return None
return UserMapper.orm_to_domain(user_model)
Python
복사
ORM을 사용하면 DB 쿼리로부터 읽어온 결과물을 DTO에 매핑하는 공수를 줄일 수 있다. ORM을 도메인 모델에 연결할 때 간단한 경우라면 inspect를 사용해 생산성을 높이고, 복잡한 변환이 필요할 때 점진적으로 복잡성을 추가하자.
parse me : 언젠가 이 글에 쓰이면 좋을 것 같은 재료을 보관해 두는 영역입니다.
1.
None
from : 과거의 어떤 원자적 생각이 이 생각을 만들었는지 연결하고 설명합니다.
1.
•
앞의 글은 이 글에서 다루는 인프라 레이어의 Repository에 대해 다룬다.
supplementary : 어떤 새로운 생각이 이 문서에 작성된 생각을 뒷받침하는지 연결합니다.
1.
None
opposite : 어떤 새로운 생각이 이 문서에 작성된 생각과 대조되는지 연결합니다.
1.
None
to : 이 문서에 작성된 생각이 어떤 생각으로 발전되거나 이어지는지를 작성하는 영역입니다.
ref : 생각에 참고한 자료입니다.