Hexagonal architecture and Python - Part III: Persistence, Transactions, Exceptions and The Final Assembly

From domain models to database entities

The VotingUserRepository has to map the VotingUser domain model to data model(s) when saving to - and vice versa, when reading from the database. The VotingUser model holds three pieces of information, which eventually influence the database design [source]:

1
2
3
4
5
6

# /src/myapp/application/domain/model/voting_user.py

class VotingUser:
    id: UserId
    karma: Karma
    votes_for_articles: list[ArticleVote]

A modern relational database with native JSON support can store such an object in a single table row. Django natively supports JSON fields and provides unbelievable JSON querying capabilities.

A more traditional way is to store ArticleVotes and VotingUsers in separate tables with an explicit relationship between them:

I bet you already picture how to express these tables as Django models. Here is a tricky question: where should we put them? It's not hard to answer if you remember the structure of HTTP API adapters. Let's do something similar for our persistence SPI adapter:

1
2
3
4
5
6
7
8
9

myapp/application/adapter/spi/
    persistence/
        entity/
            article_vote_entity.py     # Django model here
            voting_user_entity.py      # and here
        repository/
            voting_user_repository.py
        exceptions/
            ...

Here is another unorthodox move: I called Django models "Entities". That is - to avoid confusion between domain models and Django models. There is no other hidden meaning of "Entity" in this case.

Here is the ArticleVoteEntity [source]:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# /src/myapp/application/adapter/spi/persistence/entity/article_vote_entity.py

class ArticleVoteEntity(models.Model):
    VOTE_UP = 'up'
    VOTE_DOWN = 'down'

    VOTES_CHOICES = [
        (VOTE_UP, 'UP'),
        (VOTE_DOWN, 'DOWN')
    ]

    id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
    user_id = models.UUIDField()
    article_id = models.UUIDField()
    vote = models.CharField(max_length=4, choices=VOTES_CHOICES)

    class Meta:
        unique_together = [['user_id', 'article_id']]
        db_table = 'article_vote'

The VotingUserEntity is simpler [source] :

1
2
3
4
5
6
7
8

# /src/myapp/application/adapter/spi/persistence/entity/voting_user_entity.py

class VotingUserEntity(models.Model):
    user_id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
    karma = models.IntegerField()

    class Meta:
        db_table = 'voting_user'

The entity definitions are lean and have no methods in them. All that is required - is to map data from Python to database and back. And Django flawlessly does that.

Inside the repository

Once you've seen the entities (i.e. Django models), things get very predictable. The repository implements two SPI ports - one to get a VotingUser from the database, another to persist a VotingUser to the database. The repository takes care of transforming the domain model to Django entities and calling the required Django ORM methods. The repository then transforms the result back to the domain model:

The implementation fits just in few lines after hiding Django ORM boilerplate in private methods [source]:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# /src/myapp/application/adapter/spi/persistence/repository/voting_user_repository.py

def find_voting_user(self, article_id: ArticleId, user_id: UserId) -> VotingUser:
    voting_user_entity = self._get_voting_user_entity(user_id)
    votes_for_articles = self._get_votes_for_articles(article_id, user_id)

    return VotingUser(
        user_id,
        Karma(voting_user_entity.karma),
        votes_for_articles
    )

Remember that I put no explicit ForeignKey relationships here, so we have to load the votes manually. A user is saved in the same fashion [source]:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# /src/myapp/application/adapter/spi/persistence/repository/voting_user_repository.py

def save_voting_user(self, voting_user: VotingUser) -> VotingUser:
    saved_voting_user_entity = self._save_voting_user(voting_user)
    saved_article_vote_entities = self._save_votes_for_articles(
        voting_user.votes_for_articles
    )

    return self._voting_user_entity_to_domain_model(
        saved_voting_user_entity,
        saved_article_vote_entities
    )

Perhaps you are wondering: why map a VotingUserEntity back to a VotingUser domain model if we already pass it as an argument? The data passed in an INSERT INTO statement may run through numerous transformations before it reaches the data storage. The well-known case is an "auto-incremental" integer primary key generated via BEFORE INSERT triggers. So, it is paramount to return a domain model that resembles the updated database state, not what we pass to the routine.

Transactions in Django

We know that an application service vows to be agnostic of SPI adapters' implementation details. But wrapping SPI calls in transaction breaks this vow. Correction: wrapping SPI calls in a transaction using Django routines breaks this vow. Django's transaction.atomic is the unambiguous way to wrap a code block in a transaction. It is used either as a decorator or a context manager. But the problem is that Python decorators and context managers are runtime constructs. Internally transaction.atomic operates on database connection object. While we certainly need a database connection at a run time, we don't need one for unit testing. This means that the following perfectly valid production code

1
2
3
4
5
6

from django.db import transaction

class ArticleRatingService(VoteForArticleUseCase):
    @transaction.atomic
    def vote_for_article(self, command):
        ...

fails in a unit test (pytest) with RuntimeError("Database access not allowed...") exception, unless a database connection is established. The exception suggest marking the test with @pytest.mark.django_db - but this turns a unit test into an integration test!

With all the "batteries included", Django doesn't include one for this case. My version of such a battery is @transactional decorator, which wraps a callable in a transaction for production, but skips it when running tests [source].

Exception handler

Now that we know which exception to catch in the HTTP API adapter, let's discuss ways to handle it. The most obvious way is to wrap the body of ArticleVoteView.post() in try..except statement. That works fine for a small number of exceptions, but in a larger application with lots of exceptions of a similar origin, a common mechanism might be preferable. For example, an "Entity Not Found" exception can be expressed as the HTTP 404 response for any kind of entity. Luckily for us, Django Rest Framework provides a common exception-handling mechanism. You can define a custom exception handler function and set REST_FRAMEWORK.EXCEPTION_HANDLER settings value.

If a simple form, the handler catches SPI exceptions and converts them to an HTTP response [source]:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# /src/myapp/application/adapter/api/http/exceptions_handler.py

def exceptions_handler(exc, context):
    ...
    if isinstance(exc, VotingUserNotFound):
        return Response({
            'title': "Error",
            'detail': str(exc),
            'status': HTTPStatus.BAD_REQUEST
        })

Configuration

Where and when should we build the container? Consider this: the container creates and wires the components together. We can use different components for different environments - e.g. have a VotingUserExcelRepository. So, the container configures the application. And Django applications have a well-known place for configurations - the AppConfig class and the AppConfig.ready() method:

1
2
3
4
5
6
7
8
9

# /src/myapp/apps.py

class MyAppConfig(AppConfig):
    name = 'myapp'
    container: Dict[str, Any]

    def ready(self) -> None:
        from myapp.dependencies_container import build_production_dependencies_container
        self.container = build_production_dependencies_container()

URLs

The application configuration registry is populated before Django sets up the URLs. It allows to get the application configuration from the registry and bind the view to a URL [source]:

1
2
3
4
5
6
7
8

# /src/myapp/urls.py

app_config = django_apps.get_containing_app_config('myapp')
article_vote_django_view = app_config.container['article_vote_django_view']

urlpatterns = [
    path('article_vote', article_vote_django_view)
]

Django models

Django expects to find models in application's models.py file. Since we declared the entities deep down in SPI adapters, they have to be imported manually here [source]:

1
2
3
4
5
6
7
8

# /src/myapp/models.py

from myapp.application.adapter.spi.persistence.entity.article_vote_entity import (
    ArticleVoteEntity
)
from myapp.application.adapter.spi.persistence.entity.voting_user_entity import (
    VotingUserEntity
)

Testing

The Final Assembly is a slightly misleading term. There is no need to wait for all the components to be available to assemble a production-level application container. On the surface, a container exposes an API entry point, backed either by a production-grade component or a dummy.

So, the dependencies container should not be the last but one of the first items on the list when building an application following Hexagonal architecture principles. You can add the components to the container as they get ready and test every time how they fit together [source]:

1
2
3
4

# tests/test_myapp/test_dependencies_container.py

def test_build_production_ioc_container():
    build_production_dependencies_container()

At the same time, a simple smoke test can make a vote by a non-existing user and expect a response with HTTP 404 - Not Found status back:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

curl http://localhost:8000/api/article_vote \
--header "Content-Type: application/json" \
--request POST \
--data '{
   "user_id": "efed6f83-49ee-4cbc-bdbd-2b92bf428f2b",
   "article_id": "60ccea0c-0bf2-4726-8ac7-324fa03a74cd",
   "vote": "UP"
 }'

>
{
    "title":"Error",
    "detail":"User 'efed6f83-49ee-4cbc-bdbd-2b92bf428f2b' not found",
    "status":404
}

Bear in mind that the job of a smoke test is to verify whether the application works. An ideal smoke test touches all the application layers without affecting its state. Voting by a non-existing user cuts through all the application layers, down to the database and returns the expected error result.

This test can also run against an application-in-development with a dummy HTTP or SPI (database) adapters. Replacing a dummy adapter with a real one should not require any modifications to the test.