Where to place the business logic in Domain-Driven Design (Python example)
Tags: python, domaindrivendesign, ddd
Published: 2021-11-06 13:49:28

In Domain-Driven Design we have a clear sepatartion between the Domain, Application, and Infrastructure. If we are talking about the business rules, then then the most obvious choice is the Domain layer, but sometimes it could be OK to place the logic in the Application layer. Let's see what options do we have here:

**Option 1 - put your business logic in an application service layer**

This is the simplest scenario. If your domain model is very simple (i.e. CRUD based), then putting the business logic directly in a transaction script or a request handler might be ok. However, I don't like this idea if we have a non-trivial business logic.

**Option 2 - put your business logic in domain models.**

This is a really reasonable approach. To most natural place to put the business logic concerning an aggregate is the aggregate itself.

**Option 3 - put your business logic in a domain services.**

Sometimes, the business logic goes beyond a single aggregate, or spans multiple aggregates. In this case we need to take into consideration both the aggregate and it's surrounding, and this responsibility can be passes to the domain service.

Since option 1 is pretty straightforward, let's get deeper into options 2 and 3.

### Where should I put a businsess logic concerning a single aggregate?

We want to change the state of the system by executing actions (commands) on the Aggregates. Those actions can be exposed via methods of a public API of the Aggregate. Those methods are accessible from an Aggregate Root.

> Any references from outside the aggregate should only go to the aggregate root. The root can thus ensure the integrity of the aggregate as a whole.Any references from outside the aggregate should only go to the aggregate root. The root can thus ensure the integrity of the aggregate as a whole.
-- <cite>Martin Fowler</cite>

So an Aggegate Root acts as a facade of the aggregate, and it is responsible of enforcing the aggregate consistency rules. For each public method, the invariants are checked using `check_rule` method and the business logic is applied to an aggretage. Let's consider an [Auction domain](https://github.com/Ermlab/python-ddd) as an example. Our use case here is to allow users (sellers) to publish their items for sale (listings) on an auction website similar to Ebay. The business rules are as following:
1. Sellers cannot list items for free
2. Only listing drafts can be published - a listing that is already published cannot be published again.

```python
class AggregateRoot:
    def check_rule(...):
	    ...

class Listing(Entity):
	"""A class representing an item for sale"""
    status: ListingStatus = ListingStatus.Draft

    def publish():
		self.status = ListingStatus.Published

	def unpublish():
		self.status = ListingStatus.Draft

class Seller(AggregateRoot):
	"""A class representing a seller willing to list a listing for sale"""
    id: UUID
	published_listings_count: int = 0

    def publish_listing(self, listing: Listing):
		"""This method is a part of a public Seller API"""
		# validate listing
        self.check_rule(ListingPriceMustBeGreaterThanZero(listing))
		self.check_rule(ListingMustBeDraft(listing))
		# do some business logic
        listing.publish()
		self.published_listings_count += 1
		# check aggregate invariants
		self.check_rule(SellerCanHaveUpToThreeListingsInTheCatalog(self.published_listings_count))
```

### Where should I put logic concerning two or more aggregates?

Sometimes the business logic spans multiple aggregates. A common scenario here is to ensure uniqueness. Let's, we have a `User` entity in our system. The business are as following:
1. `User` can change it's username, but no more that once a month (30 days)
2. Username of a `User`must by unique within a system.

The first rule can be easily check within an aggregate:

```python
class User(AggregateRoot):
	id: UUID
	username: str
	username_changed_at: date
	
	def change_username(username: str):
		self.check_rule(UsernameCanBeChangedAtMostOnceAMonth(self.username_changed_at))
		self.username = username
		username_changed_at = date.today()
```

But how we can guarantee the uniqueness of a new username? I think the only plausible choice here is to use the **domain service**. We could create a `UsernameUniquenessChecker` domain service to handle the job:

```python
class UsernameUniquenessChecker:
	def __init__(self, user_repository):
		self.user_repository = user_repository
		
	def is_unique(username: str) -> bool:
		if self.user_repository.find_user_by_username(username):
			# user with this username exists, so it's not unique
			return False
		return True
		
class User(AggregateRoot):
	id: UUID
	username: str
	username_changed_at: date
	
	def change_username(username: str, username_uniqueness_checker: UsernameUniquenessChecker):
		self.check_rule(UsernameCanBeChangedAtMostOnceAMonth(self.username_changed_at))
		if not username_uniqueness_checker.is_unique(username):
			raise BusinessRuleValidationException("Username must be unique")
		self.username = username
		username_changed_at = date.today()
```