Where to place the business logic in Domain-Driven Design (Python example)

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 Usermust 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()