Django: write something to the database in a catch block, when using an atomic transaction

Question

I have a Django REST Framework serializer which uses select_for_update in combination with atomic transitions, like this: https://docs.djangoproject.com/en/4.2/ref/models/querysets/#select-for-update/. That works fine, except that I want to write something to the database when an error is thrown... and these insert statements are getting rolled back, never making it to the database.

The code is something like this (very much simplified but gets the point across and is reproducible):

models.py

from django.db import models


class LicenseCode(models.Model):
    code = models.CharField(max_length=200, unique=True)


class Activation(models.Model):
    license_code = models.TextField(max_length=200)
    activation_log = models.TextField(blank=True, null=True)
    success = models.BooleanField(default=False)

views.py

from django.http import HttpResponse, Http404
from django.db import transaction
from .models import Activation, LicenseCode

class LicenseRedeemSerializer:
    @transaction.atomic
    def save(self):
        license_codes = LicenseCode.objects.all().select_for_update()

        activations = []

        for license_code in license_codes:
            activations.append(
                Activation(license_code=license_code.code, success=False)
            )

        self.activate_licenses(activations)

    def activate_licenses(self, activations):
        try:
            # In our real app we'd try to activate the licenses with an external SDK. This can fail.
            raise Exception("Something went wrong!")

        except Exception as e:
            for activation in activations:
                activation.activation_log = str(e)
                activation.save()

            # With our real DRF serializer we'd raise ValidationError
            raise Http404("Could not activate the license!")


def view(request):
    # Let's make sure we have a license code to work with
    LicenseCode.objects.get_or_create(code="A")

    serialier = LicenseRedeemSerializer()
    serialier.save()

    html = "Hello there"
    return HttpResponse(html)

The problem I am facing is that when the external SDK triggers an error and I'm trying to write something to the database, that this never ends up in the database, the transaction is just rolled back.

How can I make sure that I can still write something to the database when using atomic transactions, in the except block?

Answer 1

As atomic() documentation mentions, it will roll back the transaction on exception, thus I do not think there is a way to store the information directly in Database during error.

But you can always catch the exception outside of the atomic block, then save the error and re-raise the error like this:

class MySerializer(Serializer):
    # Some serializer fields here..


    def save(self):
       try:
           self.save_data()
       except ValidationError as e:
           Log.objects.create(error = str(e))
           raise e

    @transaction.atomic
    def save_data(self):
        foo = Foo.objects.all().select_for_update()
        self.do_something(foo)
        self.do_something_else(foo)

    def do_something(self, foo):
        try:
           SomeExternalSDK.doStuff(foo)
        except SomeExternalSDK.SomeException as e:
           raise ValidationError(str(e))

    def self.do_something_else(self, foo):
        pass

Alternatively you can create a object variable (like a list) where you can put the exceptions happening in the SDK and then later store those errors in DB.

Usually, logs are not stored in DB, rather stored in the files. Instead of storing the errors in DB, you can consider storing errors in file system or use Django's logging to store the errors.

Update

---

You can remove the nested exception raising which is allowing the rollback happening. Instead, you can use a flag to raise 404 later.

@transaction.atomic
def save(self):
    license_codes = LicenseCode.objects.all().select_for_update()
    activations = []
    for license_code in license_codes:
        activations.append(
            Activation(license_code=license_code.code, success=False)
        )

    error = self.activate_licenses(activations)  # using flag to see if there is any error
    return error

def activate_licenses(self, activations):
    has_error = False
    try:
        raise Exception("Something went wrong!")

    except Exception as e:
        for activation in activations:
            activation.activation_log = str(e)
            activation.save()
        # flag
        has_error = True
    return has_error



 #view
   def view(request):
    # Let's make sure we have a license code to work with
    LicenseCode.objects.get_or_create(code="A")

    serialier = LicenseRedeemSerializer()
    error = serialier.save()
    if error:
       raise Http404('error activation')

Alternatively, you can look into savepoint but I am not sure how it will be applicable in your code.