Repository

Repository define operation to retrieve and store data from a storage backend.

While creating the application, we create definition of those methods in an abstract class. This let the possibility to improve the testability of the application.

Lets define an operation to add a book in the service/repositories.py module.

import abc
import enum
from types import EllipsisType
from uuid import UUID

from reading_club.domain.model import Book
from result import Result

from messagebus import AsyncAbstractRepository


class BookRepositoryError(enum.Enum):
    integrity_error = "integrity_error"
    not_found = "not_found"


BookRepositoryResult = Result[Book, BookRepositoryError]
BookRepositoryOperationResult = Result[EllipsisType, BookRepositoryError]


class AbstractBookRepository(AsyncAbstractRepository[Book]):
    @abc.abstractmethod
    async def add(self, model: Book) -> BookRepositoryOperationResult: ...

    @abc.abstractmethod
    async def by_id(self, id: UUID) -> BookRepositoryResult: ...

Dealing with errors

First, we create types that describe results from the repository. This is a personal choice to avoid exceptions here to deal with error, the messagebus is not enforcing this kind of practice.

The BookRepositoryError describe all kind of errors the repository can encountered, for instance, the add method can raises integrity error in case of duplicate things, and, the get method can raises a not found error. Instead of raising exceptions, the repository wrap the result in an object that must be unwrap to get the response or the error.

The AbstractBookRepository inherits AsyncAbstractRepository and note that it manage Book models. It define two operations on book, an operation to add a book in the repository, and one operation to retrieve a book from the repository.

Note

At this point, we only create definitions, this is why, there is still no tests written.

Implementing a repository

The first implementation of our repository is always the simplest one, we can create on in a tests/conftest.py to get a better view of what it looks like.

from typing import ClassVar
from uuid import UUID

from reading_club.domain.model import Book
from reading_club.service.repositories import (
    AbstractBookRepository,
    BookRepositoryError,
    BookRepositoryOperationResult,
    BookRepositoryResult,
)
from result import Err, Ok


class InMemoryBookRepository(AbstractBookRepository):
    books: ClassVar[dict[UUID, Book]] = {}
    ix_books_isbn: ClassVar[dict[UUID, str]] = {}

    async def add(self, model: Book) -> BookRepositoryOperationResult:
        if model.id in self.books:
            return Err(BookRepositoryError.integrity_error)
        if model.isbn in self.ix_books_isbn:
            return Err(BookRepositoryError.integrity_error)
        self.books[model.id] = model
        self.books[model.isbn] = model.id
        return Ok(...)

    async def by_id(self, id: UUID) -> BookRepositoryResult:
        if id not in self.books:
            return Err(BookRepositoryError.not_found)
        return Ok(self.books[id])

If you plan to store the models in a SQL database, the cool things is that, it can be done later. We can crunch the model, and a complete prototype of the app, without any sql migration noise and stay focus on the model itself.

This pattern comes from the Hexagonal Architecture, the storage backend is an implementation detail.