Defining Aggregate Repositories

Introduction

In our last lesson, we established the crucial role of the Repository pattern. We learned that it's an architectural pattern designed to abstract persistence logic, creating a clean separation between your business-focused domain model and the technical details of data storage. We saw that the repository's interface belongs in the domain layer, while the implementation lives in the infrastructure layer.

Today, we move from the "why" to the "how." Our goal is to define that repository interface. We will focus on what methods this contract should contain and how to design it to work with our domain aggregates.

Lesson Goal: By the end of this 60-minute lesson, you will be able to define a repository interface in the domain layer, focused on aggregate-level operations (e.g., add, get).

1. The Repository Interface: A Contract for Persistence

In our layered architecture, the domain core needs a way to request data without knowing anything about databases, SQL, or ORMs. It achieves this by defining a contract—an interface that specifies the persistence operations it requires. Any external component that wants to provide persistence for the domain must agree to fulfill this contract.

In Python, this contract is typically defined using an Abstract Base Class (ABC) from the abc module. An ABC can define method signatures without providing any implementation, forcing any concrete subclass to implement those methods.

This concept is a cornerstone of Hexagonal Architecture (also known as Ports and Adapters), where these interfaces are called "Ports."

As the article points out, the port defines what needs to be done (get, insert, update) without specifying how. The how is the job of the adapter (our concrete repository implementation), which we will build in later lessons.

Here is what a basic repository interface looks like in Python, using the abc module.

This code defines a clear contract. Any class that inherits from Repository must provide its own implementation for get, insert, and update. This is how the domain layer enforces its persistence requirements on the infrastructure layer.

2. Designing the Interface: Aggregate-Level Operations

A key principle in DDD is that repositories operate on Aggregates. You don't create a repository for every entity, only for Aggregate Roots. The repository's job is to manage the lifecycle of the entire aggregate as a single unit.

This means when you:

• Add data, you pass the entire, newly-created aggregate root to the repository.
• Retrieve data, you ask the repository for an aggregate by its ID, and the repository is responsible for loading the root along with all its internal entities and value objects, handing you back a fully constituted aggregate object.
• Update data, you first retrieve an aggregate, call business methods on it to change its state, and then the repository ensures those changes are saved.

Let's watch a video that demonstrates how to define a repository interface with these standard operations in mind.

The methods shown in the video form the basis of a typical repository interface:

• add(aggregate): Persists a new aggregate.
• get(id): Retrieves a single aggregate by its unique identifier.
• update(aggregate): Persists changes to an existing aggregate. The business logic that modifies the aggregate's state should have already been executed on the object before it's passed to this method.
• delete(aggregate): Removes an existing aggregate.
• get_all() or list(): Retrieves a collection of aggregates.

The interface should be expressed using the Ubiquitous Language. If your domain experts talk about "finding a customer by their email," your repository interface might have a method called find_by_email(email) instead of a generic get_by_attribute(key, value).

3. Generic vs. Specific Interfaces

The repository in the video was generic. Using Python's typing.Generic, we can define a base interface that can be reused for any aggregate. This is a powerful pattern for reducing boilerplate code.

However, most aggregates will also need specific query methods. For example, a CustomerRepository might need find_by_email(), and an OrderRepository might need find_by_customer_id().

A common and effective approach is to combine both:

1. Define a generic base repository interface with common methods like add and get.
2. For each aggregate, define a specific repository interface that inherits from the generic one and adds any domain-specific methods it needs.

This gives you both reusability and specificity.

Let's translate this into a concrete Python example. Imagine we have a Product aggregate.

First, we define our generic interface:

# In: src/domain/repository.py

from abc import ABC, abstractmethod
from typing import Generic, TypeVar, Optional, List

# A generic type variable for our aggregate roots.
# 'T' can be any type, like a Product, Order, etc.
T = TypeVar('T')

class IRepository(ABC, Generic[T]):
    """
    A generic, abstract repository interface that defines the standard
    contract for persistence operations on an aggregate root.
    """

    @abstractmethod
    def add(self, aggregate: T) -> None:
        """Adds a new aggregate to the repository."""
        raise NotImplementedError

    @abstractmethod
    def get(self, id: any) -> Optional[T]:
        """Retrieves an aggregate by its ID."""
        raise NotImplementedError

Next, we define a specific interface for our Product aggregate. This interface inherits the add and get methods and adds a new, product-specific query method.

# In: src/domain/product/repository.py
# (Assuming a Product class is defined in src/domain/product/model.py)

from .model import Product
from ..repository import IRepository # Import the generic interface
from typing import Optional, List

class IProductRepository(IRepository[Product]):
    """
    A specific repository interface for the Product aggregate.
    It inherits generic methods from IRepository and adds its own.
    """

    @abstractmethod
    def find_by_sku(self, sku: str) -> Optional[Product]:
        """Finds a product by its Stock Keeping Unit (SKU)."""
        raise NotImplementedError

    @abstractmethod
    def list(self) -> List[Product]:
        """Lists all products."""
        raise NotImplementedError

This structure is clean, reusable, and expressive. The domain layer now has a clearly defined contract for how it expects to interact with persistence, all while remaining completely ignorant of the underlying technology.

Conclusion

In this lesson, we transitioned from the theory of repositories to the practical task of defining their contracts. We've designed the "Port" that connects our domain to the outside world of data storage.

Key Takeaways:

• A repository interface is the domain's contract for persistence, typically defined in Python as an Abstract Base Class (ABC).
• The interface lives in the domain layer and defines what operations are needed, not how they are implemented.
• Operations are designed to work with Aggregate Roots as a single unit (e.g., add(order), get(order_id)).
• A powerful pattern is to create a generic base interface for common operations (add, get) and extend it with specific interfaces for each aggregate to add domain-specific queries (e.g., find_by_email).

Preview of the Next Lesson:

We have our contract. The next logical step is to create our first implementation. In the next lesson, we will implement an in-memory repository. This is a simple but incredibly useful implementation that stores data in a Python list or dictionary. It's perfect for unit testing our domain and application logic quickly and reliably, without the overhead of a real database.