Bradley Kirton's Blog

Published on Jan. 20, 2024

Go home

Representing object lifecycles using proxy models

I have been reading Domain modeling 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
│   └── 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:

db/models.py

Things to note:

"""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:

"""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