Bradley Kirton's Blog

Published on Jan. 20, 2024

Representing object life-cycles using proxy models

I have been reading Domain modelling made functional. One of the ideas I find really interesting is using the F# type system to represent the life cycle of an object. The strong type system coupled with the compiler allows the author to enforce business logic by only allowing a specific type to be processed by some function.

Python does not have a compiler like F# but it does have a type system and there are tools for static analysis. This got me thinking, could I apply this idea within a Django application? This is what I have come up with so far.

Suppose we have a ticketing system with the following project structure.

├── config
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── db
│   ├── __init__.py
│   ├── apps.py
│   ├── models.py
│   ├── migrations
│   │   └── __init__.py
├── ticketing
│   ├── __init__.py
│   ├── admin.py
│   ├── apps.py
│   ├── models.py
│   ├── migrations
│   │   └── __init__.py
│   └── views.py
├── db.sqlite3
├── manage.py
├── pyproject.toml
├── requirements.txt

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    # Persistence layer
    "db",
    # Application layer
    "ticketing",
]

Things to note:

The db application contains the entire database definition. All models defined within other apps will be proxies of these models.
The models defined with DB can be proxied from multiple applications i.e. A persistent Person model may be proxied as an Employee in one application and a Client in another.
All business logic should live within models and model managers within their respective applications.
We add both both applications to the INSTALLED_APPS because we want to access the ticketing proxy models via the admin panel.

db/models.py

Things to note:

The entire application database definition is contained within this file.

"""Application persitence layer."""
from django.db import models


class Person(models.Model):
    """
    Models a person.
    """

    name = models.CharField(max_length=255)

    class Meta:
        db_table = "person"

    def __str__(self) -> str:
        return self.name


class Ticket(models.Model):
    """
    Models a ticket.
    """

    class Status(models.TextChoices):
        """
        Models the life-cycle of a ticket.

        OPEN -> IN_PROGRESS -> RESOLVED -> CLOSED
        OPEN -> CLOSED
        IN_PROGRESS -> CLOSED
        """

        OPEN = "open", "Open"
        IN_PROGRESS = "in_progress", "In process"
        RESOLVED = "resolved", "Resolved"
        CLOSED = "closed", "Closed"

    status = models.CharField(max_length=15, choices=Status.choices)
    opened_by = models.ForeignKey("Person", on_delete=models.CASCADE, related_name="+")
    opened_at = models.DateTimeField()
    assigned_to = models.ForeignKey("Person", on_delete=models.CASCADE, null=True, related_name="+")
    assigned_at = models.DateTimeField(null=True)
    resolved_by = models.ForeignKey("Person", on_delete=models.CASCADE, related_name="+")
    resolved_at = models.DateTimeField(null=True)
    closed_by = models.ForeignKey("Person", on_delete=models.CASCADE, related_name="+")
    closed_at = models.DateTimeField(null=True)
    closed_comment = models.TextField()

    class Meta:
        db_table = "ticket"

    def __str__(self) -> str:
        return f"{self.pk}:{self.status}"

ticketing/models.py

Things to note:

The ticketing system does not expose the Ticket model directly. Instead it exposes proxy models which represent a ticket at various stages of the ticketing life-cycle.
All business logic is contained within the model and managers.
The ticketing life-cycle becomes a state machine. For example, we cannot call assign on a closed ticket as this method is not available on the ClosedTicket model.
An attempt retrieve a ticket of a particular type is not currently in that status will return None or raise a DoesNotExist error.

"""Data model for ticketing system."""
import datetime
import typing as t

from django.db import models

from db import models as db_models


class Person(db_models.Person):
    """
    Models a person.
    """

    class Meta:
        proxy = True


class CloseTicketProtocol(t.Protocol):
    """
    Protocol which
    """

    status: t.Union[
        db_models.Ticket.Status.OPEN.value,
        db_models.Ticket.Status.IN_PROGRESS.value,
        db_models.Ticket.Status.RESOLVED.value,
    ]
    closed_by: Person
    closed_at: datetime.datetime | None
    closed_comment: str
    pk: int

    def save(self) -> None:
        ...

    def close(
        self,
        closed_by: Person,
        closed_at: datetime.datetime,
        closed_comment: str,
    ) -> "ClosedTicket":
        ...


class CloseTicketMixin:
    """
    Provides close ticket functionality via mixin.

    Open, pending, and resolved tickets can be closed.

    Ticketing life-cycle:

    OPEN -> IN_PROGRESS -> RESOLVED -> CLOSED
    OPEN -> CLOSED
    IN_PROGRESS -> CLOSED
    """

    def close(
        self: CloseTicketProtocol,
        closed_by: "Person",
        closed_at: datetime.datetime,
        closed_comment: str,
    ) -> "ClosedTicket":
        """
        Close the ticket.

        :param closed_by: The person who closed the ticket.
        :param closed_at: The timestamp at which the ticket was closed.
        :param closed_comment: A comment which descibes the reason for closing.
        """
        self.status = db_models.Ticket.Status.CLOSED
        self.closed_by = closed_by
        self.closed_at = closed_at
        self.closed_comment = closed_comment
        self.save()
        return ClosedTicket.objects.get(pk=self.pk)


class OpenTicketManager(models.Manager):
    """
    Business logic layer for the open ticket model.
    """

    def get_queryset(self) -> models.QuerySet["OpenTicket"]:
        """
        Filters on open tickets only.
        """

        return super().get_queryset().filter(status=db_models.Ticket.Status.OPEN)


class OpenTicket(CloseTicketMixin, db_models.Ticket):
    """
    Models an open ticket.
    """

    objects: models.Manager["OpenTicket"] = OpenTicketManager()

    class Meta:
        proxy = True

    def __str__(self) -> str:
        return f"{self.pk}: {self.opened_by} {self.opened_at}"

    @classmethod
    def new(cls, opened_by: Person, opened_at: datetime.datetime) -> "OpenTicket":
        """
        Assigns a ticket to a person.

        :param opened_by: The person who opened the ticket.
        :param opened_at: The timestamp at which the ticket was opened.
        """
        return cls.objects.create(
            status=db_models.Ticket.Status.OPEN,
            opened_by=opened_by,
            opened_at=opened_at,
            assigned_to=None,
            assigned_at=None,
            resolved_by=None,
            resolved_at=None,
            closed_by=None,
            closed_at=None,
            closed_comment="",
        )

    def assign_ticket(self, assigned_to: Person, assigned_at: datetime.datetime) -> "InProgressTicket":
        """
        Assigns a ticket to a person.

        :param assigned_to: The person to whom the ticket will be assigned.
        :param assigned_at: The timestamp at which the ticket was assigned.
        """

        self.status = db_models.Ticket.Status.IN_PROGRESS
        self.assigned_to = assigned_to
        self.assigned_at = assigned_at
        self.save()

        return InProgressTicket.objects.get(pk=self.pk)


class InProgressManager(models.Manager):
    """
    Business logic layer for the in progress ticket model.
    """

    def get_queryset(self) -> models.QuerySet["InProgress"]:
        """
        Filters on in progress tickets only.
        """

        return super().get_queryset().filter(status=db_models.Ticket.Status.IN_PROGRESS)


class InProgressTicket(CloseTicketMixin, db_models.Ticket):
    """
    Models an in progress ticket.
    """

    objects: models.Manager["InProgressTicket"] = InProgressManager()

    class Meta:
        proxy = True

    def __str__(self) -> str:
        return f"{self.pk}: {self.assigned_to} {self.assigned_at}"

    def resolve_ticket(self, resolved_by: Person, resolved_at: datetime.datetime) -> "ResolvedTicket":
        """
        Resolve the ticket.

        :param resolved_by: The person to whom the ticket will be assigned.
        :param resolved_at: The timestamp at which the ticket was assigned.
        """

        self.status = db_models.Ticket.Status.RESOLVED
        self.resolved_by = resolved_by
        self.resolved_at = resolved_at
        self.save()

        return ResolvedTicket.objects.get(pk=self.pk)


class ResolvedManager(models.Manager):
    """
    Business logic layer for the resolved tickets model.
    """

    def get_queryset(self) -> models.QuerySet["ResolvedTicket"]:
        """
        Filters on resolved tickets only.
        """

        return super().get_queryset().filter(status=db_models.Ticket.Status.RESOLVED)


class ResolvedTicket(CloseTicketMixin, db_models.Ticket):
    """
    Models a resolved ticket.
    """

    objects: models.Manager["ResolvedTicket"] = ResolvedManager()

    class Meta:
        proxy = True

    def __str__(self) -> str:
        return f"{self.pk}: {self.resolved_by} {self.resolved_at}"


class ClosedManager(models.Manager):
    """
    Business logic layer for the closed tickets model.
    """

    def get_queryset(self) -> models.QuerySet["ClosedTicket"]:
        """
        Filters on closed tickets only.
        """

        return super().get_queryset().filter(status=db_models.Ticket.Status.CLOSED)


class ClosedTicket(db_models.Ticket):
    """
    Models a closed ticket.
    """

    objects: models.Manager["ClosedTicket"] = ClosedManager()

    class Meta:
        proxy = True

    def __str__(self) -> str:
        return f"{self.pk}: {self.closed_by} {self.closed_at}"

Once we have hooked up the models to the Admin panel we get an interface into each section of the life-cycle.

Admin panel