The Repository Pattern

Defining objects using data schemas is very useful to do field validations and be consistent along the code. Next step is to define a place where to store collections of objects, and this is done in repositories.

Repository Pattern

The basics for the Repository Pattern are:

• Define an interface for the repository.
• And then you could have as many implementations as you like

The Contract

With this pattern, the important part is not how we implement the repository. The important thing is that all repositories and the implementations respect the defined contract (interface) and all use the same methods. Then we can implement controllers that no longer depend on a specific repository implementation.

Now all we have to do is set up a dependency injection framework to use the desired repository implementation. The dependency injection framework manages the instances of the repository classes and "injects" them into the controller constructor. And simply by modifying these settings we could change our data access technology without touching a single line of code in your controller.

Complete Isolation.

And this is where unit testing comes in. Using this pattern allows us to isolate the different layers. Since the controller code is loosely coupled to the repository, all we have to do in unit tests is provide a mock implementation in the repository that allows you to define its behavior. This gives us the ability to unit test controller actions without relying on a database or whatever.

Advantages

Using the Repository Pattern has many advantages:

• Your business logic can be unit tested without data access logic.
• The database access code can be reused and is managed centrally, making it easy to implement any database access policy.
• It is easy to implement the domain logic.

Example

Let's see it with a simple example:

First of all, let's create a new test file:

touch tests/test_02_repositories.py 

Create the Abstract Class

tests/test_02_repositories.py

@@ -0,0 +1,22 @@

+from abc import ABC, abstractmethod
+from typing import List
+from uuid import UUID
+
+
+class Book(object):
+    pass
+
+
+class BookRepository(ABC):
+
+    @abstractmethod
+    def add(self, book: Book) -> Book:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get(self, book_id: UUID) -> Book:
+        raise NotImplementedError
+
+    @abstractmethod
+    def list(self) -> List[Book]:
+        raise NotImplementedError

For this example, we'll create a very simple repository, that stores all the information in an in-memory dictionary.

Write a Basic Test and Define Empty In-Memory Repository

The first step is to create the initial test and the initial repository implementation, defining all methods:

tests/test_02_repositories.py

@@ -13,10 +13,26 @@

     def add(self, book: Book) -> Book:
         raise NotImplementedError

     @abstractmethod
     def get(self, book_id: UUID) -> Book:
         raise NotImplementedError

     @abstractmethod
     def list(self) -> List[Book]:
         raise NotImplementedError
+
+
+class InMemoryBookRepository(BookRepository):
+    def add(self, new_book: Book) -> Book:
+        pass
+
+    def get(self, book_id: UUID) -> Book:
+        pass
+
+    def list(self) -> List[Book]:
+        pass
+
+
+def test_in_memory_repository_created_successfully():
+    repo = InMemoryBookRepository()
+    assert type(repo) == InMemoryBookRepository

Then, we add the book dictionary variable to store all books:

tests/test_02_repositories.py

@@ -16,23 +16,31 @@

     @abstractmethod
     def get(self, book_id: UUID) -> Book:
         raise NotImplementedError

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


 class InMemoryBookRepository(BookRepository):
+    books = dict()  # this sets a class-level attribute, common to all instances of `InMemoryBooksRepository`
+
     def add(self, new_book: Book) -> Book:
         pass

     def get(self, book_id: UUID) -> Book:
         pass

     def list(self) -> List[Book]:
-        pass
+        return [book for book_id, book in self.books.items()]


 def test_in_memory_repository_created_successfully():
     repo = InMemoryBookRepository()
     assert type(repo) == InMemoryBookRepository
+
+
+def test_in_memory_repository_initially_empty_successfully():
+    repo = InMemoryBookRepository()
+    items = repo.list()
+    assert len(items) == 0

Implement all Methods from the Contract

tests/test_02_repositories.py

@@ -1,46 +1,71 @@

 from abc import ABC, abstractmethod
-from typing import List
-from uuid import UUID
+from typing import List, Optional
+from uuid import UUID, uuid4
+
+import pytest
+from pydantic import BaseModel, Field


-class Book(object):
-    pass
+class Book(BaseModel):
+    title: str
+    author: str
+
+
+class BookInDB(Book):
+    id: UUID = Field(default_factory=uuid4)


 class BookRepository(ABC):

     @abstractmethod
-    def add(self, book: Book) -> Book:
+    def add(self, book: Book) -> BookInDB:
         raise NotImplementedError

     @abstractmethod
-    def get(self, book_id: UUID) -> Book:
+    def get(self, book_id: UUID) -> BookInDB:
         raise NotImplementedError

     @abstractmethod
-    def list(self) -> List[Book]:
+    def list(self) -> List[BookInDB]:
         raise NotImplementedError


 class InMemoryBookRepository(BookRepository):
     books = dict()  # this sets a class-level attribute, common to all instances of `InMemoryBooksRepository`

-    def add(self, new_book: Book) -> Book:
-        pass
+    def add(self, new_book: Book) -> BookInDB:
+        book = BookInDB(**new_book.dict())
+        self.books[book.id] = book
+        return book

-    def get(self, book_id: UUID) -> Book:
-        pass
+    def get(self, book_id: UUID) -> Optional[BookInDB]:
+        return self.books.get(book_id, None)

-    def list(self) -> List[Book]:
+    def list(self) -> List[BookInDB]:
         return [book for book_id, book in self.books.items()]


 def test_in_memory_repository_created_successfully():
     repo = InMemoryBookRepository()
     assert type(repo) == InMemoryBookRepository


 def test_in_memory_repository_initially_empty_successfully():
     repo = InMemoryBookRepository()
     items = repo.list()
     assert len(items) == 0
+
+
+@pytest.fixture()
+def a_book() -> Book:
+    return Book(title="A nice title", author="John Smith")
+
+
+def test_add_valid_data_to_in_memory_repository_successfully(a_book):
+    repo = InMemoryBookRepository()
+    repo.add(a_book)
+    items = repo.list()
+    assert len(items) == 1
+    assert isinstance(items[0], Book)
+    assert items[0].title == a_book.title
+    assert items[0].author == a_book.author

Check that Added Books are Valid

We need to be sure that the implementation is as robust as possible. To do it, we write a test forcing to add a data structure different than a Book, and we see that fails. Then we need to add some extra validation to the add method, just to check that the new_book is a Book type instance.

tests/test_02_repositories.py

@@ -27,20 +27,22 @@


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


 class InMemoryBookRepository(BookRepository):
     books = dict()  # this sets a class-level attribute, common to all instances of `InMemoryBooksRepository`

     def add(self, new_book: Book) -> BookInDB:
+        if not isinstance(new_book, Book):
+            raise ValueError("This repository only accepts Book objects.")
         book = BookInDB(**new_book.dict())
         self.books[book.id] = book
         return book

     def get(self, book_id: UUID) -> Optional[BookInDB]:
         return self.books.get(book_id, None)

     def list(self) -> List[BookInDB]:
         return [book for book_id, book in self.books.items()]

@@ -62,10 +64,20 @@



 def test_add_valid_data_to_in_memory_repository_successfully(a_book):
     repo = InMemoryBookRepository()
     repo.add(a_book)
     items = repo.list()
     assert len(items) == 1
     assert isinstance(items[0], Book)
     assert items[0].title == a_book.title
     assert items[0].author == a_book.author
+
+
+def test_add_invalid_data_to_n_memory_repository_raises_exception():
+    repo = InMemoryBookRepository()
+    a_dictionary = dict(title="Foo", author="Boo")
+    with pytest.raises(ValueError):
+        repo.add(a_dictionary)
+
+    items = repo.list()
+    assert len(items) == 0

At this point, we get a new error, because we are using a class-level variable to store the books. This means that all the of InMemoryBooksRepository shares the same variable. To solve this, we can do an optional initialization of the self.books variable in the __init__ method.

tests/test_02_repositories.py

@@ -1,83 +1,60 @@

-from abc import ABC, abstractmethod
-from typing import List, Optional
-from uuid import UUID, uuid4
+from uuid import UUID

 import pytest
-from pydantic import BaseModel, Field

-
-class Book(BaseModel):
-    title: str
-    author: str
-
-
-class BookInDB(Book):
-    id: UUID = Field(default_factory=uuid4)
-
-
-class BookRepository(ABC):
-
-    @abstractmethod
-    def add(self, book: Book) -> BookInDB:
-        raise NotImplementedError
-
-    @abstractmethod
-    def get(self, book_id: UUID) -> BookInDB:
-        raise NotImplementedError
-
-    @abstractmethod
-    def list(self) -> List[BookInDB]:
-        raise NotImplementedError
-
-
-class InMemoryBookRepository(BookRepository):
-    books = dict()  # this sets a class-level attribute, common to all instances of `InMemoryBooksRepository`
-
-    def add(self, new_book: Book) -> BookInDB:
-        if not isinstance(new_book, Book):
-            raise ValueError("This repository only accepts Book objects.")
-        book = BookInDB(**new_book.dict())
-        self.books[book.id] = book
-        return book
-
-    def get(self, book_id: UUID) -> Optional[BookInDB]:
-        return self.books.get(book_id, None)
-
-    def list(self) -> List[BookInDB]:
-        return [book for book_id, book in self.books.items()]
+from tests.repositories import InMemoryBookRepository
+from tests.schemas import Book


 def test_in_memory_repository_created_successfully():
     repo = InMemoryBookRepository()
     assert type(repo) == InMemoryBookRepository


 def test_in_memory_repository_initially_empty_successfully():
     repo = InMemoryBookRepository()
     items = repo.list()
     assert len(items) == 0


-@pytest.fixture()
-def a_book() -> Book:
-    return Book(title="A nice title", author="John Smith")
-
-
 def test_add_valid_data_to_in_memory_repository_successfully(a_book):
-    repo = InMemoryBookRepository()
+    repo = InMemoryBookRepository(initial_books=dict())
     repo.add(a_book)
     items = repo.list()
     assert len(items) == 1
     assert isinstance(items[0], Book)
     assert items[0].title == a_book.title
     assert items[0].author == a_book.author


 def test_add_invalid_data_to_n_memory_repository_raises_exception():
-    repo = InMemoryBookRepository()
+    repo = InMemoryBookRepository(initial_books=dict())
     a_dictionary = dict(title="Foo", author="Boo")
     with pytest.raises(ValueError):
         repo.add(a_dictionary)

     items = repo.list()
     assert len(items) == 0
+
+
+def test_get_book_by_id_from_in_memory_repository_successfully(a_book, another_book):
+    repo = InMemoryBookRepository(initial_books=dict())
+    book1 = repo.add(a_book)
+    book2 = repo.add(another_book)
+
+    response1 = repo.get(book_id=book1.id)
+    assert response1 == book1
+    assert response1.title == a_book.title
+
+    response2 = repo.get(book_id=book2.id)
+    assert response2 == book2
+    assert response2.title == another_book.title
+
+
+def test_get_book_by_id_from_in_memo_repository_with_invalid_id_returns_none(a_book):
+    repo = InMemoryBookRepository(initial_books=dict())
+    repo.add(a_book)
+
+    invalid_id = UUID('00000000-0000-0000-0000-000000000000')
+    response1 = repo.get(book_id=invalid_id)
+    assert response1 is None

Add some extra tests to check the get function

tests/test_02_repositories.py

@@ -1,60 +1,112 @@

-from uuid import UUID
+from abc import ABC, abstractmethod
+from typing import List, Optional
+from uuid import UUID, uuid4

 import pytest
+from pydantic import BaseModel, Field

-from tests.repositories import InMemoryBookRepository
-from tests.schemas import Book
+
+class Book(BaseModel):
+    title: str
+    author: str
+
+
+class BookInDB(Book):
+    id: UUID = Field(default_factory=uuid4)
+
+
+class BookRepository(ABC):
+
+    @abstractmethod
+    def add(self, book: Book) -> BookInDB:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get(self, book_id: UUID) -> BookInDB:
+        raise NotImplementedError
+
+    @abstractmethod
+    def list(self) -> List[BookInDB]:
+        raise NotImplementedError
+
+
+class InMemoryBookRepository(BookRepository):
+    def __init__(self):
+        self.books = dict()
+
+    def add(self, new_book: Book) -> BookInDB:
+        if not isinstance(new_book, Book):
+            raise ValueError("This repository only accepts Book objects.")
+        book = BookInDB(**new_book.dict())
+        self.books[book.id] = book
+        return book
+
+    def get(self, book_id: UUID) -> Optional[BookInDB]:
+        return self.books.get(book_id, None)
+
+    def list(self) -> List[BookInDB]:
+        return [book for book_id, book in self.books.items()]


 def test_in_memory_repository_created_successfully():
     repo = InMemoryBookRepository()
     assert type(repo) == InMemoryBookRepository


 def test_in_memory_repository_initially_empty_successfully():
     repo = InMemoryBookRepository()
     items = repo.list()
     assert len(items) == 0


+@pytest.fixture()
+def a_book() -> Book:
+    return Book(title="A nice title", author="John Smith")
+
+
+@pytest.fixture()
+def another_book() -> Book:
+    return Book(title="Another nice title", author="Jane Boo")
+
+
 def test_add_valid_data_to_in_memory_repository_successfully(a_book):
-    repo = InMemoryBookRepository(initial_books=dict())
+    repo = InMemoryBookRepository()
     repo.add(a_book)
     items = repo.list()
     assert len(items) == 1
     assert isinstance(items[0], Book)
     assert items[0].title == a_book.title
     assert items[0].author == a_book.author


 def test_add_invalid_data_to_n_memory_repository_raises_exception():
-    repo = InMemoryBookRepository(initial_books=dict())
+    repo = InMemoryBookRepository()
     a_dictionary = dict(title="Foo", author="Boo")
     with pytest.raises(ValueError):
         repo.add(a_dictionary)

     items = repo.list()
     assert len(items) == 0


 def test_get_book_by_id_from_in_memory_repository_successfully(a_book, another_book):
-    repo = InMemoryBookRepository(initial_books=dict())
+    repo = InMemoryBookRepository()
     book1 = repo.add(a_book)
     book2 = repo.add(another_book)

     response1 = repo.get(book_id=book1.id)
     assert response1 == book1
     assert response1.title == a_book.title

     response2 = repo.get(book_id=book2.id)
     assert response2 == book2
     assert response2.title == another_book.title


 def test_get_book_by_id_from_in_memo_repository_with_invalid_id_returns_none(a_book):
-    repo = InMemoryBookRepository(initial_books=dict())
+    repo = InMemoryBookRepository()
     repo.add(a_book)

     invalid_id = UUID('00000000-0000-0000-0000-000000000000')
     response1 = repo.get(book_id=invalid_id)
     assert response1 is None

Improving Tests: Fixtures

For those tests, we have defined a fixture function to create new Book instances for test purposes. Given that we know that we'll need to do this again and again in future tests, what we can do is to move those fixtures from test_02_repositories.py to conftest.py. conftest.py file is loaded everytime we run a test.

tests/test_02_repositories.py

@@ -52,30 +52,20 @@

     repo = InMemoryBookRepository()
     assert type(repo) == InMemoryBookRepository


 def test_in_memory_repository_initially_empty_successfully():
     repo = InMemoryBookRepository()
     items = repo.list()
     assert len(items) == 0


-@pytest.fixture()
-def a_book() -> Book:
-    return Book(title="A nice title", author="John Smith")
-
-
-@pytest.fixture()
-def another_book() -> Book:
-    return Book(title="Another nice title", author="Jane Boo")
-
-
 def test_add_valid_data_to_in_memory_repository_successfully(a_book):
     repo = InMemoryBookRepository()
     repo.add(a_book)
     items = repo.list()
     assert len(items) == 1
     assert isinstance(items[0], Book)
     assert items[0].title == a_book.title
     assert items[0].author == a_book.author

tests/conftest.py

+import pytest
+
+from tests.test_02_repositories import Book
+
+
+@pytest.fixture()
+def a_book() -> Book:
+    return Book(title="A nice title", author="John Smith")
+
+
+@pytest.fixture()
+def another_book() -> Book:
+    return Book(title="Another nice title", author="Jane Boo")

Code Refactor

Data Schemas

And we also know that we'll use the Book data type in future tests, so we can refactor the data schemas from test_02_repositories.py to schemas.py, to clean the code and make more usable.

tests/test_02_repositories.py

@@ -1,25 +1,17 @@

 from abc import ABC, abstractmethod
 from typing import List, Optional
-from uuid import UUID, uuid4
+from uuid import UUID

 import pytest
-from pydantic import BaseModel, Field

-
-class Book(BaseModel):
-    title: str
-    author: str
-
-
-class BookInDB(Book):
-    id: UUID = Field(default_factory=uuid4)
+from tests.schemas import Book, BookInDB


 class BookRepository(ABC):

     @abstractmethod
     def add(self, book: Book) -> BookInDB:
         raise NotImplementedError

     @abstractmethod
     def get(self, book_id: UUID) -> BookInDB:

tests/schemas.py

+from uuid import UUID, uuid4
+
+from pydantic import BaseModel, Field
+
+
+class Book(BaseModel):
+    title: str
+    author: str
+
+
+class BookInDB(Book):
+    id: UUID = Field(default_factory=uuid4)

tests/conftest.py

@@ -1,13 +1,13 @@

 import pytest

-from tests.test_02_repositories import Book
+from tests.schemas import Book


 @pytest.fixture()
 def a_book() -> Book:
     return Book(title="A nice title", author="John Smith")


 @pytest.fixture()
 def another_book() -> Book:
     return Book(title="Another nice title", author="Jane Boo")

Repository

tests/test_02_repositories.py

@@ -1,94 +1,60 @@

-from abc import ABC, abstractmethod
-from typing import List, Optional
 from uuid import UUID

 import pytest

-from tests.schemas import Book, BookInDB
-
-
-class BookRepository(ABC):
-
-    @abstractmethod
-    def add(self, book: Book) -> BookInDB:
-        raise NotImplementedError
-
-    @abstractmethod
-    def get(self, book_id: UUID) -> BookInDB:
-        raise NotImplementedError
-
-    @abstractmethod
-    def list(self) -> List[BookInDB]:
-        raise NotImplementedError
-
-
-class InMemoryBookRepository(BookRepository):
-    def __init__(self):
-        self.books = dict()
-
-    def add(self, new_book: Book) -> BookInDB:
-        if not isinstance(new_book, Book):
-            raise ValueError("This repository only accepts Book objects.")
-        book = BookInDB(**new_book.dict())
-        self.books[book.id] = book
-        return book
-
-    def get(self, book_id: UUID) -> Optional[BookInDB]:
-        return self.books.get(book_id, None)
-
-    def list(self) -> List[BookInDB]:
-        return [book for book_id, book in self.books.items()]
+from tests.repositories import InMemoryBookRepository
+from tests.schemas import Book


 def test_in_memory_repository_created_successfully():
     repo = InMemoryBookRepository()
     assert type(repo) == InMemoryBookRepository


 def test_in_memory_repository_initially_empty_successfully():
     repo = InMemoryBookRepository()
     items = repo.list()
     assert len(items) == 0


 def test_add_valid_data_to_in_memory_repository_successfully(a_book):
-    repo = InMemoryBookRepository()
+    repo = InMemoryBookRepository(initial_books=dict())
     repo.add(a_book)
     items = repo.list()
     assert len(items) == 1
     assert isinstance(items[0], Book)
     assert items[0].title == a_book.title
     assert items[0].author == a_book.author


 def test_add_invalid_data_to_n_memory_repository_raises_exception():
-    repo = InMemoryBookRepository()
+    repo = InMemoryBookRepository(initial_books=dict())
     a_dictionary = dict(title="Foo", author="Boo")
     with pytest.raises(ValueError):
         repo.add(a_dictionary)

     items = repo.list()
     assert len(items) == 0


 def test_get_book_by_id_from_in_memory_repository_successfully(a_book, another_book):
-    repo = InMemoryBookRepository()
+    repo = InMemoryBookRepository(initial_books=dict())
     book1 = repo.add(a_book)
     book2 = repo.add(another_book)

     response1 = repo.get(book_id=book1.id)
     assert response1 == book1
     assert response1.title == a_book.title

     response2 = repo.get(book_id=book2.id)
     assert response2 == book2
     assert response2.title == another_book.title


 def test_get_book_by_id_from_in_memo_repository_with_invalid_id_returns_none(a_book):
-    repo = InMemoryBookRepository()
+    repo = InMemoryBookRepository(initial_books=dict())
     repo.add(a_book)

     invalid_id = UUID('00000000-0000-0000-0000-000000000000')
     response1 = repo.get(book_id=invalid_id)
     assert response1 is None

tests/repositories.py

+from abc import ABC, abstractmethod
+from typing import List, Optional
+from uuid import UUID
+
+from tests.schemas import Book, BookInDB
+
+
+class BookRepository(ABC):
+
+    @abstractmethod
+    def add(self, book: Book) -> BookInDB:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get(self, book_id: UUID) -> BookInDB:
+        raise NotImplementedError
+
+    @abstractmethod
+    def list(self) -> List[BookInDB]:
+        raise NotImplementedError
+
+
+class InMemoryBookRepository(BookRepository):
+    books = dict()
+
+    def __init__(self, initial_books: dict = None):
+        if initial_books is not None:
+            self.books = initial_books
+
+    def add(self, new_book: Book) -> BookInDB:
+        if not isinstance(new_book, Book):
+            raise ValueError("This repository only accepts Book objects.")
+        book = BookInDB(**new_book.dict())
+        self.books[book.id] = book
+        return book
+
+    def get(self, book_id: UUID) -> Optional[BookInDB]:
+        return self.books.get(book_id, None)
+
+    def list(self) -> List[BookInDB]:
+        return [book for book_id, book in self.books.items()]