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)