-
Notifications
You must be signed in to change notification settings - Fork 0
/
domain.py
115 lines (96 loc) · 3.61 KB
/
domain.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
import uuid as _uuid
from depeche_db import event_sourcing as _es
from . import messages as _messages
class Account(_es.EventSourcedAggregateRoot[_uuid.UUID, _messages.AccountEvent]):
id: _uuid.UUID
number: str
balance: int
def get_id(self) -> _uuid.UUID | None:
if hasattr(self, "id"):
return self.id
return None
def _apply(self, event: _messages.AccountEvent) -> None:
if isinstance(event, _messages.AccountCreatedEvent):
self.id = event.account_id
self.number = event.account_number
self.balance = 0
elif isinstance(event, _messages.DepositedEvent):
self.balance += event.amount
elif isinstance(event, _messages.WithdrawnEvent):
self.balance -= event.amount
else:
raise NotImplementedError(f"Event {type(event)} is not supported")
@property
def version(self) -> int:
return self._version
@classmethod
def create(cls, account_id: _uuid.UUID, number: str) -> "Account":
account = cls()
account.apply(
_messages.AccountCreatedEvent(account_id=account_id, account_number=number)
)
return account
def deposit(self, amount: int, transfer_id: _uuid.UUID | None = None) -> None:
self.apply(
_messages.DepositedEvent(
account_id=self.id, amount=amount, transfer_id=transfer_id
)
)
def withdraw(self, amount: int, transfer_id: _uuid.UUID | None = None) -> None:
self.apply(
_messages.WithdrawnEvent(
account_id=self.id, amount=amount, transfer_id=transfer_id
)
)
class Transfer(_es.EventSourcedAggregateRoot[_uuid.UUID, _messages.TransferEvent]):
"""
This is not only a domain object, but also a process manager (in conjunction
with the event handlers defined in `TransferHandler` and `AccountHandler`).
"""
id: _uuid.UUID
from_account_id: _uuid.UUID
to_account_id: _uuid.UUID
amount: int
status: str # initial, withdrawn, finished
def get_id(self) -> _uuid.UUID | None:
if hasattr(self, "id"):
return self.id
return None
def _apply(self, event: _messages.TransferEvent) -> None:
if isinstance(event, _messages.TransferInitiatedEvent):
self.id = event.transfer_id
self.from_account_id = event.from_account_id
self.to_account_id = event.to_account_id
self.amount = event.amount
self.status = "initial"
elif isinstance(event, _messages.TransferWithdrawnEvent):
self.status = "withdrawn"
elif isinstance(event, _messages.TransferFinishedEvent):
self.status = "finished"
else:
raise NotImplementedError(f"Event {type(event)} is not supported")
@property
def version(self) -> int:
return self._version
@classmethod
def initiate(
cls,
transfer_id: _uuid.UUID,
from_account_id: _uuid.UUID,
to_account_id: _uuid.UUID,
amount: int,
) -> "Transfer":
transfer = cls()
transfer.apply(
_messages.TransferInitiatedEvent(
transfer_id=transfer_id,
from_account_id=from_account_id,
to_account_id=to_account_id,
amount=amount,
)
)
return transfer
def track_withdrawn(self) -> None:
self.apply(_messages.TransferWithdrawnEvent(transfer_id=self.id))
def track_deposited(self) -> None:
self.apply(_messages.TransferFinishedEvent(transfer_id=self.id))