Repository Pattern with Django

I used a plugin called pytest-django to help with test database management.

Rewriting the first repository test was a minimal change, just rewriting some raw SQL with a call to the Django ORM / Queryset language:

from djangoproject.alloc import models as django_models


@pytest.mark.django_db
def test_repository_can_save_a_batch():
    batch = model.Batch("batch1", "RUSTY-SOAPDISH", 100, eta=date(2011, 12, 25))

    repo = repository.DjangoRepository()
    repo.add(batch)

    [saved_batch] = django_models.Batch.objects.all()
    assert saved_batch.reference == batch.reference
    assert saved_batch.sku == batch.sku
    assert saved_batch.qty == batch._purchased_quantity
    assert saved_batch.eta == batch.eta

The second test is a bit more involved since it has allocations, but it is still made up of familiar-looking django code:

@pytest.mark.django_db
def test_repository_can_retrieve_a_batch_with_allocations():
    sku = "PONY-STATUE"
    d_line = django_models.OrderLine.objects.create(orderid="order1", sku=sku, qty=12)
    d_batch1 = django_models.Batch.objects.create(reference="batch1", sku=sku, qty=100, eta=None)
    d_batch2 = django_models.Batch.objects.create(reference="batch2", sku=sku, qty=100, eta=None)
    django_models.Allocation.objects.create(line=d_line, batch=d_batch1)

    repo = repository.DjangoRepository()
    retrieved = repo.get("batch1")

    expected = model.Batch("batch1", sku, 100, eta=None)
    assert retrieved == expected  # Batch.__eq__ only compares reference
    assert retrieved.sku == expected.sku
    assert retrieved._purchased_quantity == expected._purchased_quantity
    assert retrieved._allocations == {
        model.OrderLine("order1", sku, 12),
    }

Here’s how the actual repository ends up looking:

class DjangoRepository(AbstractRepository):

    def add(self, batch):
        super().add(batch)
        self.update(batch)

    def update(self, batch):
        django_models.Batch.update_from_domain(batch)

    def _get(self, reference):
        return django_models.Batch.objects.filter(
            reference=reference
        ).first().to_domain()

    def list(self):
        return [b.to_domain() for b in django_models.Batch.objects.all()]

You can see that the implementation relies on the Django models having some custom methods for translating to and from our domain model.¹

from django.db import models
from allocation.domain import model as domain_model

class Batch(models.Model):
    reference = models.CharField(max_length=255)
    sku = models.CharField(max_length=255)
    qty = models.IntegerField()
    eta = models.DateField(blank=True, null=True)

    @staticmethod
    def update_from_domain(batch: domain_model.Batch):
        try:
            b = Batch.objects.get(reference=batch.reference)  
        except Batch.DoesNotExist:
            b = Batch(reference=batch.reference)  
        b.sku = batch.sku
        b.qty = batch._purchased_quantity
        b.eta = batch.eta  
        b.save()
        b.allocation_set.set(
            Allocation.from_domain(l, b)  
            for l in batch._allocations
        )

    def to_domain(self) -> domain_model.Batch:
        b = domain_model.Batch(
            ref=self.reference, sku=self.sku, qty=self.qty, eta=self.eta
        )
        b._allocations = set(
            a.line.to_domain()
            for a in self.allocation_set.all()
        )
        return b


class OrderLine(models.Model):
    #...

Unit of Work Pattern with Django

The tests don’t change too much

def insert_batch(ref, sku, qty, eta):  
    django_models.Batch.objects.create(reference=ref, sku=sku, qty=qty, eta=eta)

def get_allocated_batch_ref(orderid, sku):  
    return django_models.Allocation.objects.get(
        line__orderid=orderid, line__sku=sku
    ).batch.reference


@pytest.mark.django_db(transaction=True)
def test_uow_can_retrieve_a_batch_and_allocate_to_it():
    insert_batch('batch1', 'HIPSTER-WORKBENCH', 100, None)

    uow = unit_of_work.DjangoUnitOfWork()
    with uow:
        batch = uow.batches.get(reference='batch1')
        line = model.OrderLine('o1', 'HIPSTER-WORKBENCH', 10)
        batch.allocate(line)
        uow.commit()

    batchref = get_allocated_batch_ref('o1', 'HIPSTER-WORKBENCH')
    assert batchref == 'batch1'


@pytest.mark.django_db(transaction=True)  
def test_rolls_back_uncommitted_work_by_default():
    ...

@pytest.mark.django_db(transaction=True)  
def test_rolls_back_on_error():
    ...

Because we had little helper functions in these tests, the actual main body of the tests are pretty much the same as they were with SQLA

the pytest-django mark.django_db(transaction=True) is required to test our custom transaction/rollback behaviors.

And the implementation is quite simple, although it took me a few goes to find what actual invocation of Django’s transaction magic would work:

class DjangoUnitOfWork(AbstractUnitOfWork):

    def __enter__(self):
        self.batches = repository.DjangoRepository()
        transaction.set_autocommit(False)  
        return super().__enter__()

    def __exit__(self, *args):
        super().__exit__(*args)
        transaction.set_autocommit(True)

    def commit(self):
        for batch in self.batches.seen:  
            self.batches.update(batch)  
        transaction.commit()  

    def rollback(self):
        transaction.rollback()  

set_autocommit(False) was the best way to tell Django to stop automatically committing each ORM operation immediately, and begin a transaction.

Then we use the explicit rollback and commits.

One difficulty: because, unlike with SQLAlchemy, we’re not instrumenting the domain model instances themselves, the commit() command needs to explicitly got through all the objects that have been touched by every repository and manually updated them back to the ORM.

API: Django Views Are Adapters

The Django views.py file ends up being almost identical to the old flask_app.py, because our architecture means it’s a very thin wrapper around our service layer (which didn’t change at all btw).

os.environ['DJANGO_SETTINGS_MODULE'] = 'djangoproject.django_project.settings'
django.setup()

@csrf_exempt
def add_batch(request):
    data = json.loads(request.body)
    eta = data['eta']
    if eta is not None:
        eta = datetime.fromisoformat(eta).date()
    services.add_batch(
        data['ref'], data['sku'], data['qty'], eta,
        unit_of_work.DjangoUnitOfWork(),
    )
    return HttpResponse('OK', status=201)

@csrf_exempt
def allocate(request):
    data = json.loads(request.body)
    try:
        batchref = services.allocate(
            data['orderid'],
            data['sku'],
            data['qty'],
            unit_of_work.DjangoUnitOfWork(),
        )
    except (model.OutOfStock, services.InvalidSku) as e:
        return JsonResponse({'message': str(e)}, status=400)

    return JsonResponse({'batchref': batchref}, status=201)

Why was this all so hard?

OK it works but it does feel like more effort than Flask/SQLAlchemy. Why is that?

The main reason at a low level is because Django’s ORM doesn’t work in the same way. We don’t have an equivalent of the SQLAlchemy classical mapper, so our ActiveRecord and our domain model can’t be the same object. Instead we have to build a manual translation layer behind the repository instead. That’s more work (although once it’s done the ongoing maintenance burden shouldn’t be too high).

Because Django is so tightly coupled to the database, you have to use helpers like pytest-django and thinking carefully about test databases, right from the very first line of code, in a way that we didn’t have to when we started out with our pure domain model.

But at a higher level, it’s because the entire reason that Django is so great is that it’s designed around the sweet spot of making it easy to build CRUD apps with minimal boilerplate. But the entire thrust of our book is about what to do when your app is no longer a simple CRUD app.

At that point, Django starts hindering more than it helps. Things like the Django Admin, which are so awesome when you start out, become actively dangerous if the whole point of your app is to build a complex set of rules and modelling around the workflow of state changes. The Django admin bypasses all of that.

What To Do If You Already Have Django

So what should you do if you want to apply some of the patterns in this book to a Django app? We’d say:

• Repository and Unit of Work pattern are going to be quite a lot of work. The main thing they will buy you in the short term is faster unit tests, so evaluate whether that feels worth it in your case. In the longer term, they decouple your app from Django and the database, so if you anticipate wanting to migrate away from either of those, Repository and UoW are a good idea.

• Service Layer might be of interest if you’re seeing a lot of duplication in your views.py. It can be a good way of thinking about your use cases, separately from your web endpoints.

• You can still theoretically do DDD and domain modelling with Django models, tightly coupled as they are to the database; you may be slowed down by migrations, but it shouldn’t be fatal. So, as long as your apps is not too complex and your tests not too slow, you may be able to get something out of the “fat models” approach: push as much logic down to your models as possible, and apply patterns like Entity, Value Object and Aggregate. Although see caveat below

With that said, word in the Django community is that people find that “fat models” runs into scalability problems of its own, particularly around managing interdependencies between apps. In those cases, there’s a lot to be said for extracting out a “business logic” or “domain” layer to sit between your views and forms, and your models.py which you can then keep as minimal as possible.

Steps along the way

Supposing you’re working on a Django project which you’re not sure is going to get complex enough to warrant the patterns we recommend, but you still want to put a few steps in place to make your life easier, both in the medium term, and if you want to migrate to some of our patterns later?

• One piece of advice we’ve heard is to put a logic.py into every Django app, from day one. This gives you a place to put business logic, and to keep your forms, views and models free of business logic. It can become a stepping stone for moving to a fully decoupled domain model and/or service layer later.

• A business logic layer might start out working with Django model objects, and only later become fully decoupled from the framework and work on plain Python data structures.

• For the read-side, you can get some of the benefits of CQS by putting reads into one place, avoiding ORM calls sprinkled all over the place

• When separating out modules for reads and modules for domain logic, it may be worth decoupling yourself from the django apps hierarchy. Business concerns will cut across them.

For more thoughts and actual lived experience dealing with existing applications, read on to Epilogue: What now?.