Domain driven design CQRS with multiple aggregates and bounded context

22 Views Asked by At

I am trying to apply commands and events along with domain driven desing and I'm not sure how to handle commands that involve multiple aggregates.

Suppose I have a use case where a new customer selects a plan for which an Account needs to be created with the customer added as a Member of that account. Also, the account needs to be loaded with Package details available by purchasing the selected Plan like user_limit, total_seats

For this I have defined an Account aggregate with Member as its local entity in Account context and Plan aggregate with Package value object in Subscription context. I trigger the CreateAccountCommand with plan_id,customer_name, account_name for which I need to fetch the package details for the plan_id from Plan aggregate and then use its details to create new Account with the Member.

what is the valid approach to handle such cases? As I've read and understood I believe fetching of plan details should be separated to Query and CreateUserCommand should receive the query result to perform action. But I'm not sure how can I do this to create loosly coupled and flexible design.

#Account aggregate with Members - Account Context
class Account(AggregateRoot):
    plan_id: uuid
    name: str
    credit: Money
    total_seats = AllocatedTotal
    user_limit: AllocatedUsers
    members: list[Member]

    def __init__(self, name: str, plan_id: uuid):
        self.name = name
        self.plan_id = plan_id
        self.members = []
        self.user_limit = AllocatedUsers(0)
        self.total_seats = AllocatedTotal(0)
        self.credit = Money(0)

    def add_member(self, member_name: str) 
        pass

class Member(Entity):
    name: str
    role: Role

    #other member related stuffs

# Plan aggregate with Package - Subscription Context
class Plan(AggregateRoot):
    name: str
    type: str
    package: Package

class Package(ValueObject):
    total_seats: AllocatedTotal
    user_limit: AllocatedUsers

#command
@dataclass
class CreateAccount(Command):
    account_name: str
    user_name: str
    plan_id: uuid


#command_handler
class CreateAccountHandler(CommandHandler):
    def __init__(self, uow: AbstractUnitOfWork,
                 account_service: 'AccountService'):
        self.uow = uow
        self.account_service = account_service

    def handle(self, command: SubscribeUser) -> None:
        with self.uow:
            self.account_service.create_account(plan_id=command.plan_id, account_name=command.account_name,
                                                member_name=command.user_name)
            self.uow.commit()

    def __enter__(self):
        return self
    
#domain_service
class AccountService(AccountServicePort):
    def __init__(self, account_repository: AccountRepository,
                 plan_service: PlanService):
        self.account_repository = account_repository
        self.plan_service = plan_service

    def create_account(self, plan_id: UUID, account_name: str, member_name: str) -> None:
        plan = self.plan_service.get_plan_by_id(plan_id)
        if not plan:
            raise ValueError("Plan not found")

        account = Account(name=account_name, plan_id=plan_id)
        member = Member(name=member_name, role=Role.ADMIN)
        account.add_member(member.id, member)
        account.load_total_users(plan.package.user_limit.count)
        account.total_seats(plan.package.total_seats.total)
        self.account_repository.add(entity=account)
0

There are 0 best solutions below