The Professor
Index Course index

Module 5 · Architecture and maintainable code

Lesson 25 — Service patterns

Repository, service, DTO, domain events and the outbox pattern.

Published
In this lesson
  1. Exercise 1 — Repository
  2. Exercise 2 — Outbox
  3. Exercise 3 — DTO
  4. Exercise 4 — Events
  5. Exercise 5 — Deduplication
  6. Professor's summary

Exercise 1 — Repository

python
class ReservationRepository(Protocol):
    def get(self, ref: str) -> Reservation: ...
    def save(self, r: Reservation) -> None: ...
    def existe_reserva_activa(self, event_id: int, user_id: int) -> bool: ...

class InMemoryReservationRepository:
    def __init__(self): self._items: dict[str, Reservation] = {}
    def get(self, ref): return self._items[ref]          # KeyError if missing: same contract as the ORM's .get()
    def save(self, r): self._items[r.public_ref] = r
    def existe_reserva_activa(self, event_id, user_id):
        return any(r.event_id == event_id and r.user_id == user_id and r.status == "ACTIVE"
                   for r in self._items.values())
  1. For Venue/Seat: direct manager. A repository there is rewriting the ORM without gaining anything; the Reservation aggregate justifies its own because it concentrates the locked query, the invariants and the test fake.

Exercise 2 — Outbox

Migration (excerpt):

python
class OutboxEvent(models.Model):
    aggregate_type = models.CharField(max_length=50)
    aggregate_id = models.CharField(max_length=64)
    event_type = models.CharField(max_length=100)
    payload = models.JSONField()
    created_at = models.DateTimeField(auto_now_add=True)
    publicado_at = models.DateTimeField(null=True, blank=True)

    class Meta:
        indexes = [models.Index(
            fields=["created_at"],
            condition=Q(publicado_at__isnull=True),   # partial index: pending only
            name="outbox_pending_idx",
        )]

Atomicity test: force the failure (seat already reserved) and verify OutboxEvent.objects.count() == 0 after the DomainError — the rollback took the event with it: that is exactly the outbox's point.

Idempotent poller:

python
class Command(BaseCommand):
    def handle(self, *args, **opts):
        while True:
            with transaction.atomic():
                batch = list(OutboxEvent.objects.filter(publicado_at__isnull=True).order_by("id")[:50])
                if not batch:
                    return
                for ev in batch:
                    self.publicar(ev)          # broker/log; at-least-once
                    ev.publicado_at = timezone.now()
                    ev.save(update_fields=["publicado_at"])

Exercise 3 — DTO

  1. The view:
python
@api_view(["GET"])
def reservation_detail(request, ref):
    r = repos.reservas.get(ref)                    # or servicio.leer(ref)
    return Response(ReservationSummarySerializer(resumen(r)).data)
  1. Serializing from Model: renaming seat_refs → seats touches the serializer and potentially every consumer of model_to_dict. With the DTO: it touches resumen() (one place) and the serializer that paints it. One translation point.
  2. expires_in_seconds is business rule (the reservation's TTL, 10) — if it lives in the serializer, every client recomputes it and the CLI doesn't have it. In the DTO with an injected Clock, the test pins it without freezegun.

Exercise 4 — Events

Minimum events table (just-enough payload; the consumer makes no extra queries):

EventEmitterConsumerMinimum payload
ReservationConfirmedreservar serviceemail, analyticsref, user_id, event_id, seats, total, expires_at
ReservationExpiredexpiration job (29)email, release seatsref, user_id
PaymentSucceededpayments serviceconfirm reservation, receiptintent_id, ref, amount
RefundIssuedrefunds serviceemail, accountingintent_id, ref, amount, reason
EventPublishedadmin/organizerpush notificationsevent_id, venue
PaymentFailedpayments service"try again" email, metricsintent_id, ref, reason

The classic minimum-payload mistake: the consumer "only looks at the event_id and then queries" — if the row changed (reservation expired and deleted), the email goes out with wrong data. Self-sufficient payload (idempotency + immutability of the fact).

Exercise 5 — Deduplication

python
def consumir(ev: OutboxEvent):
    key = f"event:{ev.id}"
    if not cache.add(key, "1", 60 * 60 * 24):   # SET NX: atomic
        return                                   # already processed: dedup
    self.publicar(ev)

Test: the "publish" fake is a list; process the same event twice and len(published) == 1. Deduplication lives in the CONSUMER because the poller only guarantees at-least-once from the outbox to the queue; the final effect (email sent, 17's signed webhook) is what needs effective exactly-once, and only the effect's owner knows its state. The poller may dedupe as an optimization, not as a guarantee.

Professor's summary

  • Repository for aggregates with logic; direct manager for trivial CRUD. A layer duplicating the ORM without domain is noise.
  • Outbox = reliable events with a table and a poller; at-least-once + dedup in the consumer. The alternative (distributed transactions) doesn't exist in your stack.
  • DTO outwards when the consumer isn't the HTTP view or the shape differs from the model; Model inside.

^ top