In-Memory Repository for Unit Testing

Introduction

In our last lesson, we focused on defining the repository interface—the "Port" in our Hexagonal Architecture. We created an abstract contract in the domain layer that specifies what persistence operations our application needs (like add and get) without committing to how they are performed.

Today, we'll build our first concrete implementation of that interface: an in-memory repository. This is a simple yet powerful "Adapter" that lives in the infrastructure layer. Its purpose is not for production use but to serve as a fast, reliable stand-in for a real database during unit testing.

Lesson Goal: By the end of this 60-minute lesson, you will be able to implement an in-memory repository in the infrastructure layer for unit testing purposes.

1. The "Fake" Repository: A Test-Friendly Adapter

The core idea behind the Repository pattern is to decouple our domain logic from persistence details. This decoupling is what allows us to swap different persistence strategies. For production, we might use a PostgreSQL database. For integration tests, maybe SQLite. And for fast unit tests of our domain and application logic, we can use a simple in-memory collection.

This in-memory version is often called a "fake" or "mock" repository. It fulfills the exact same interface contract as the "real" repository, but its implementation is trivial: it just stores objects in a Python list or dict.

Let's start by reading a section from the book "Cosmic Python," which provides a clear and concise implementation of this idea.

As you saw, the FakeRepository is incredibly simple. It's just a thin wrapper around a native Python data structure. This simplicity is its greatest strength. It has no external dependencies, runs instantly, and allows us to test our application's logic in complete isolation.

2. Implementing an In-Memory Repository in Python

Let's make this concrete. In the previous lesson, we defined an IProductRepository interface. Now, we'll create an InMemoryProductRepository that implements it.

Recall our interface from the last lesson:

# Located in: src/domain/product/repository.py

from abc import abstractmethod
from typing import Optional, List
from .model import Product
from ..repository import IRepository # The generic base interface

class IProductRepository(IRepository[Product]):
    """
    A specific repository interface for the Product aggregate.
    """
    @abstractmethod
    def find_by_sku(self, sku: str) -> Optional[Product]:
        raise NotImplementedError

    @abstractmethod
    def list(self) -> List[Product]:
        raise NotImplementedError

(Note: The base IRepository provides add and get methods.)

Now, let's create the in-memory implementation. We will place this file in the infrastructure layer, as it's a concrete implementation detail. A dictionary is a great choice for storage, as it provides fast lookups by ID, which is exactly what our get method needs.

# Located in: src/infrastructure/persistence/in_memory_product_repository.py

from typing import Dict, List, Optional
from uuid import UUID

from src.domain.product.model import Product
from src.domain.product.repository import IProductRepository

class InMemoryProductRepository(IProductRepository):
    """
    An in-memory implementation of the product repository for testing purposes.
    """
    def __init__(self, initial_products: Optional[List[Product]] = None):
        # We use a dictionary for O(1) lookups for the get() method.
        # The key is the product's ID (a UUID), and the value is the Product aggregate.
        self._products: Dict[UUID, Product] = {}
        if initial_products:
            for product in initial_products:
                self.add(product)

    def add(self, product: Product) -> None:
        """Adds a new product to the repository."""
        if product.id in self._products:
            # You might want to raise a custom exception here
            raise ValueError(f"Product with ID {product.id} already exists.")
        self._products[product.id] = product

    def get(self, product_id: UUID) -> Optional[Product]:
        """Retrieves a product by its ID."""
        return self._products.get(product_id)

    def find_by_sku(self, sku: str) -> Optional[Product]:
        """Finds a product by its SKU."""
        for product in self._products.values():
            if product.sku == sku:
                return product
        return None

    def list(self) -> List[Product]:
        """Lists all products."""
        return list(self._products.values())

This class fully implements the IProductRepository contract using a simple dictionary. It's now a fully functional "adapter" that we can plug into our application for testing.

3. The In-Memory Repository in Action

Seeing the code is one thing, but watching it used in a test really clarifies its value. The following video demonstrates a very similar pattern: creating a "mock" repository and then immediately using it to write a simple, effective unit test.

The test shown in the video is a perfect example of why we do this. Here’s how we could write a similar test for our InMemoryProductRepository using pytest:

# Located in: tests/unit/test_in_memory_product_repository.py

import uuid
from src.domain.product.model import Product
from src.infrastructure.persistence.in_memory_product_repository import InMemoryProductRepository

def test_add_and_get_product():
    # Arrange
    repo = InMemoryProductRepository()
    product_id = uuid.uuid4()
    product = Product(id=product_id, sku="TEST-SKU-001", name="Test Product", price=99.99)

    # Act
    repo.add(product)
    retrieved_product = repo.get(product_id)

    # Assert
    assert retrieved_product is not None
    assert retrieved_product.id == product_id
    assert retrieved_product.sku == "TEST-SKU-001"

def test_find_by_sku():
    # Arrange
    product_id = uuid.uuid4()
    product = Product(id=product_id, sku="UNIQUE-SKU-123", name="Another Product", price=10.00)
    repo = InMemoryProductRepository(initial_products=[product])

    # Act
    found_product = repo.find_by_sku("UNIQUE-SKU-123")
    not_found_product = repo.find_by_sku("NON-EXISTENT-SKU")

    # Assert
    assert found_product is not None
    assert found_product.id == product_id
    assert not_found_product is None

These tests are:

• Fast: They execute in milliseconds.
• Isolated: They don't require a database, network, or any other external service.
• Reliable: They won't fail due to network glitches or database configuration issues.

This enables a rapid and robust Test-Driven Development (TDD) workflow.

4. Ports and Adapters: Tying It All Together

Let's briefly revisit the "Ports and Adapters" terminology to solidify our understanding of where the in-memory repository fits.

To summarize in our own context:

• The Port: The IProductRepository abstract base class. It defines the interface from the application's point of view.
• The Adapters: The InMemoryProductRepository is one adapter. A future SqlAlchemyProductRepository will be another.

Our application's core logic will depend only on the port, remaining blissfully unaware of which adapter is currently plugged in.

Conclusion

Today, we built our first, fully functional repository. While it doesn't persist data to a disk, it perfectly fulfills the contract defined by our domain layer, enabling a powerful testing strategy.

Key Takeaways:

• An in-memory repository is a concrete implementation of the repository interface that uses a simple Python collection (like a dict or list) for storage.
• It resides in the infrastructure layer because it's a specific implementation detail, separate from the domain's business rules.
• Its primary purpose is to facilitate fast, isolated unit tests of your domain and application services, forming a key part of a TDD workflow.
• The ability to easily create a "fake" repository is a strong indicator that your repository interface (the "port") is well-designed and properly decoupled.

Preview of the Next Lesson:

We now have a way to test our application logic without a database. The next step is to build an adapter for real-world persistence. In our next lesson, we will begin this process by learning how to set up a basic SQLAlchemy model to map an Aggregate Root to a database schema. This will lay the foundation for building our SqlAlchemyProductRepository.