도메인 로직, 엔티티 등

[DDD] 도메인 엔티티, Domain Entity Claude 비즈니스 로직을 도메인에 가까이 두어야하는 이유 (DDD) 나는 지금가지 도메인 로직을 우회하는 개발을 하고 있었다. 도메인 객체가 단순히 데이터 컨테이너의 역할만 하고 있었음. 그러다보니 service 레이어가 비대해지고 중복이 발생. 서비스 계층은 아래와 같은 일을 한다.

1. 트랜잭션 관리

python
Copy
class PostService:
    def create_post_with_places(self, post_data: dict, places: list[Place]):
# 트랜잭션 내에서 여러 작업을 조율
        with self.unit_of_work.begin():
            post = Post(**post_data)
            post.add_places(places)# 도메인 객체의 메서드 호출
            self.post_repository.save(post)
 

2. 여러 도메인 객체/서비스 간의 조율

python
Copy
class PostService:
    def publish_post(self, post_id: int):
        post = self.post_repository.get_by_id(post_id)
        author = self.author_service.get_author(post.author_id)
 
        if author.can_publish():# 작성자의 권한 확인
            post.publish()# 도메인 객체의 메서드 호출
            self.notification_service.notify_followers(author.id)
            self.post_repository.save(post)
 

3. 사용자 입력의 검증과 변환

python
Copy
class PostService:
    def update_post_location(self, post_id: int, location_data: dict):
# 입력 데이터 검증 및 변환
        try:
            latitude = float(location_data['lat'])
            longitude = float(location_data['lng'])
        except (ValueError, KeyError):
            raise InvalidLocationData()
 
        post = self.post_repository.get_by_id(post_id)
        post.update_location(latitude, longitude)# 도메인 객체에 위임
 

4. 외부 시스템과의 통합

python
Copy
class PostService:
    def create_post_with_image(self, content: str, image: File):
# 외부 서비스 통합 조율
        image_url = self.image_service.upload(image)
        post = Post(content=content, image_url=image_url)
        return self.post_repository.save(post)
 

그럼 앞으로 어떻게 해야하는가

post에 place를 추가하는 로직들이 있다고 생각해보자. 지금까지의 구현은 다음과 같다

# services/post.py
def some_method(self):
    if place not in post.places:
        post.places.append(place)
        post.updated_at = datetime.now()

이러면 중복코드 발생, 비즈니스 규칙 일관성을 해칠 위험, 수정 어려움, 테스트 빡셈 등의 단점이 발생한다. 아래와 같이 하자

# models/post.py
class Post(Base):
	id: ~
	asdf: ~
	...
	
	def add_places(self, places: list[Place]):
		for place in places:
			if place not in self.places:  # 중복 검사
				if self.can_add_more_places():  # 비즈니스 규칙 검사
					self.places.append(place)
					self.place_count += 1  # 장소 수 카운트

진작 이렇게 할 걸