mariobittencourt

from datetime import datetime
from abc import ABC, ABCMeta, abstractmethod

class DomainEvent(ABC, metaclass=ABCMeta):
    @abstractmethod
    def __init__(self, aggregate_id: str, occurred_at: str):
        self.type = self.__class__.__name__
        self.aggregate_id = aggregate_id
        self.occurred_at = occurred_at

mariobittencourt

from __future__ import annotations
from typing import List

from src.domain.models.domain_event import DomainEvent
from src.domain.models.payment_created import PaymentCreated

class Payment:
    def __init__(self, payment_id: str):
        self._payment_id = payment_id
        self._events = []

mariobittencourt

from __future__ import annotations
from typing import List

from src.domain.models.domain_event import DomainEvent
from src.domain.models.payment_created import PaymentCreated

class Payment:
    def __init__(self, payment_id: str):
        self._payment_id = payment_id
        self._events = []

mariobittencourt

class Payment:
    # ... existing methods
    
    def refund(self, amount_refunded: float, refund_id: str) -> None:
        # business rules
        if self.status != 'SETTLED':
            raise Exception('Only settled payments can be refunded')

        if amount_refunded > self.amount_due:
            raise Exception('You can only refund amounts equal or smaller than the amount due')

mariobittencourt

class PaymentRefunded(DomainEvent):
    def __init__(self, aggregate_id: str, amount_refunded: float, refund_id: str, occurred_at: str = datetime.utcnow().isoformat()):
        super().__init__(aggregate_id, occurred_at)
        self.amount_refunded = amount_refunded
        self.refund_id = refund_id

mariobittencourt

def method_dispatch(func):
    dispatcher = functools.singledispatch(func)

    def wrapper(*args, **kw):
        return dispatcher.dispatch(args[1].__class__)(*args, **kw)

    wrapper.register = dispatcher.register
    functools.update_wrapper(wrapper, func)
    return wrapper

mariobittencourt

class EventStorePaymentRepository(PaymentRepository):
    def __init__(self, username: str, password: str):
        self._loop = asyncio.get_event_loop()
        # photonpump is a client that interfaces with event store
        self._client = photonpump.connect(loop=self._loop, username=username, password=password)
        self._connected = False

    async def save(self, payment: Payment) -> bool:
        await self._check_connection()

mariobittencourt

class EventStorePaymentRepository(PaymentRepository):
    # ... existing code
    
    async def find_by_id(self, payment_id: str) -> Optional[Payment]:
        await self._check_connection()

        stream_name = self._create_stream_name(payment_id)
        payment = Payment(payment_id)
        async for event in self._client.iter(stream=stream_name, from_event=0):
            await self._reconstruct_from_event(event, payment)

mariobittencourt

{
  "customer": {
    "S": "Mr Buy More"
  },
  "deliveryAddress": {
    "S": "My place.com"
  },
  "events": {
    "L": [
      {

mariobittencourt

{
  "Records": [
    {
      "eventID": "54436bc389eb70304e166170e60a3644",
      "eventName": "INSERT",
      "eventVersion": "1.1",
      "eventSource": "aws:dynamodb",
      "awsRegion": "your-region",
      "dynamodb": {