Where to place the business logic in Domain-Driven Design (Python example)
Przemys艂aw G贸recki
Posted on November 6, 2021
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.
-- Martin Fowler
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 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:
- Sellers cannot list items for free
- Only listing drafts can be published - a listing that is already published cannot be published again.
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:
-
User
can change it's username, but no more that once a month (30 days) - Username of a
User
must by unique within a system.
The first rule can be easily check within an aggregate:
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:
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()
Posted on November 6, 2021
Join Our Newsletter. No Spam, Only the good stuff.
Sign up to receive the latest update from our blog.
Related
November 16, 2022