python-aggregate-and-repo

Aggregate and repository pattern

The aggregate is the unit of consistency: a chunk of state that's loaded together, mutated together, and saved together as a single transaction. The repository is the only thing that reads or writes aggregates of its type. Together they're the heart of the data layer.

This skill encodes the SA-model-with-behavior style: the SQLAlchemy ORM model carries the business methods (order.place(), customer.deactivate()), and there's no separate pure-Python "domain" class. For most backend services — especially integration- and orchestration-heavy ones — this is the right trade-off. For invariant-dense systems where you want sub-millisecond invariant tests without a DB, see python-pure-domain-layer.

When to use this skill

• Adding a new aggregate (Order, Customer, Subscription, Document, ...)
• Adding a method on an existing aggregate
• Adding a repo query that a service needs
• Refactoring a "fat router" or "anaemic model" into a proper aggregate

The rules

1. SA model carries behavior; no separate domain class

The SA-mapped class holds the columns, the methods, the derived values, and the domain exceptions in one file. The shape:

# app/models/order.py — shape only; see examples/order_model.py for the full version
class Order(Base):
    __tablename__ = "order"

    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    customer_id: Mapped[UUID] = mapped_column(ForeignKey("customer.id"))
    status: Mapped[OrderStatus] = mapped_column(default=OrderStatus.DRAFT)
    lines: Mapped[list["OrderLine"]] = relationship(
        cascade="all, delete-orphan", lazy="selectin",
    )

    @property
    def total(self) -> Decimal: ...                  # derived value

    @classmethod
    def new(cls, *, customer_id: UUID, currency: str) -> Self: ...   # constructor

    def add_line(self, *, product_id: UUID, quantity: int, unit_price: Decimal) -> None:
        if self.status is not OrderStatus.DRAFT:
            raise OrderAlreadyPlacedException(self.id)
        self.lines.append(OrderLine(...))            # mutation guarded by precondition

    def place(self) -> None: ...                     # transition, guarded
    def cancel(self, *, reason: str) -> None: ...    # idempotent guard