Bradley Kirton's Blog

Published on Jan. 18, 2024

Railway oriented programming in Django

As far as I can tell railway oriented programming is not a widely recognised concept. I think the first time I was introduced to the concept was reading through the dry-python returns documentation.

I really like this idea because it treats errors as data. In Python I often find myself using errors as a mechanism for control flow within a try except block. I have not used the returns package in any projects yet but have been experimenting with it a bit.

Below is an example of such an experiment (Note I have not even checked if this works, it is more conceptual at the moment).

Suppose we have a use case to create a user object in our database and on some external CRM.

The happy path, if everything goes well, returns the newly created user
The failure path can raise a ValidationError, DatabaseError or CRMClientError

This specification describes the user case

CreateUser:
    inputs:
        UserData
    outputs:
        Result[User, ValidationError | DatabaseError | CRMClientError]
    dependencies:
        CRMClient

    subprocesses:
        Validate Input:
            raises ValidationError
        Create in DB:
            raises DatabaseError
        Create in CRM:
            raises CRMClientError

from returns import result as r


class UserCreationError(Exception):
    """Base class for user creation errors."""

    def __init__(self, message: str) -> None:
        super().__init__(message)


class ValidationError(UserCreationError):
    """Raised when invalid user data is provided."""

    def __init__(self, message: str, *, form: forms.Form) -> None:
        super().__init__(message)
        self.form = form


class DatabaseError(UserCreationError):
    """Raised when some database error is encountered when creating the user."""


class CRMClientError(UserCreationError):
    """Raised when some error is encountered when creating the user on the external CRM."""


class UserManager(models.Manager):
    """
    Create a user.
    """

    @transaction.atomic
    def create_user(self, data: dict[str, str], crm_client: CRMClient) -> User:
        """
        Create a user.

        :param data: The create user payload.
        :param crm_client: An instance of the CRM client.
        :returns: A user or user creation error.
        :raises ValidationError: If the provided user data is invalid.
        :raises DatabaseError: If a database error is encountered.
        :raises CRMClientError: If a CRM error is encountered.
        """

        form = CreateUserForm(data=data)

        if not form.is_valid():
            raise ValidationError("Invalid create user payload.", form=form)

        first_name = form.cleaned_data["first_name"]
        last_name = form.cleaned_data["last_name"]

        try:
            user = User.objects.create(first_name=first_name, last_name=last_name)
        except Exception as ex:
            raise DatabaseError("Failed to create user record.") from ex

        try:
            crm_client.create(user_id=user.pk, first_name=first_name, last_name=last_name)
        except Exception as ex:
            raise CRMClientError("CRM is currently unavailable.") from ex

        return user


# views.py
@require_POST
def create_user_try_except_view(request: HttpRequest) -> HttpResponse:
    try:
        CustomUser.objects.create_user(data=request.POST, crm_client=CRMClient())
        response = redirect(reverse("login-view"))
    except ValidationError as ex:
        response = render(request, "create-user-form.html", {"form": ex.form})
    except DatabaseError as ex:
        response = redirect(reverse("db-unavailable-view"))
    except CRMClientError as ex:
        response = redirect(reverse("crm-unavailable-view"))
    except Exception as ex:
        raise ex

    return response


@require_POST
def create_user_railway_view(request: HttpRequest) -> HttpResponse:
    safe_create_user = r.safe(CustomUser.objects.create_user)
    result = safe_create_user(data=request.POST, crm_client=CRMClient())

    match result:
        case r.Success(CustomUser()):
            return redirect(reverse("login-view"))
        case r.Failure(ValidationError() as error):
            return render(request, "create-user-form.html", {"form": error.form})
        case r.Failure(DatabaseError() as error):
            return redirect(reverse("db-unavailable-view"))
        case r.Failure(CRMClientError() as error):
            return redirect(reverse("crm-unavailable-view"))
        case r.Failure(Exception() as error):
            error.unwrap()