Quite recently, I was working on a backend project written in Django and I had defined the following models:
class Order(BaseModel):
class Status(models.TextChoices):
COMPLETE = "Complete"
PENDING = "Pending"
CANCELLED = "Cancelled"
created_by = models.ForeignKey(
User, on_delete=models.CASCADE, related_name="orders", null=True, blank=True
)
vendor = models.ForeignKey(
Vendor,
on_delete=models.CASCADE,
related_name="vendor_orders",
null=True,
blank=True,
)
order_number = models.CharField(max_length=255, unique=True, blank=True, null=True)
status = models.CharField(
max_length=255, choices=Status.choices, default=Status.PENDING
)
notes = models.TextField(null=True, blank=True)
email_sent = models.BooleanField(default=False)
def save(self, *args, **kwargs):
if not self.order_number or self.order_number == "":
self.order_number = self.generate_unique_order_number()
super().save(*args, **kwargs)
def generate_unique_order_number(self):
prefix = "ORD"
suffix = "".join(random.choices(string.digits, k=5))
return f"{prefix}-{suffix}"
def __str__(self) -> str:
return str(self.order_number)
class OrderItem(BaseModel):
class Status(models.TextChoices):
RECEIVED = "Received"
PENDING = "Pending"
order = models.ForeignKey(Order, on_delete=models.CASCADE, related_name="items")
product = models.ForeignKey(
Product, on_delete=models.CASCADE, related_name="product_items"
)
status = models.CharField(
max_length=255, choices=Status.choices, default=Status.PENDING
)
price = models.DecimalField(max_digits=9, decimal_places=2)
quantity = models.IntegerField(default=1)
total = models.GeneratedField(
expression=F("quantity") * F("price"),
output_field=models.FloatField(),
db_persist=True,
)
def __str__(self) -> str:
return f"{self.order.order_number} - {self.product.name} - {self.total}"
TASK DEFINITION
The task at hand is to automatically update an Order status once all related OrderItems instances have been marked as Received.
To achieve this, we are going to tap into the almighty Django signals that ships out of the box.
So the first step is to create a signals.py file in our app folder:
touch core/signals.py
And register it in the core/apps.py file as so:
from django.apps import AppConfig
class CoreConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "core"
def ready(self):
from .import signals
Once that is out of the way we can start by writing the signal function that will ensure all an order item is marked as complete once all orderitems are updated as received.
The function for that looks like:
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Order, OrderItem
@receiver(post_save, sender=OrderItem)
def update_order_status(sender, instance, **kwargs):
order = instance.order
if order.items.filter(status=OrderItem.Status.PENDING).exists():
if order.status == Order.Status.COMPLETE:
order.status = Order.Status.PENDING
order.save()
return
# If all items are received, update the status of the order to Complete
order.status = Order.Status.COMPLETE
order.save()
So basically when an OrderItem is updated, our signal checks if the Order associated to that OrderItem has any OrderItem with a status of Pending which should mean that not all related OrderItem instances have been received.
If that is the case the signal checks if the Order status is set to Complete. If that is the case, it reverts the status to Pending.
If all OrderItem instances have been received, i.e status is Received, it is time to mark that Order as complete.
And that works okay after I run the tests.
SECOND TASK DEFINITION
The reverse is true, if an Order gets marked as Complete, our backend should mark all related OrderItem instances as Received. This calls for another signal function to do that right?
I have written it and it looks like:
@receiver(post_save, sender=Order)
def mark_order_items_received(sender, instance, **kwargs):
if instance.status == Order.Status.COMPLETE:
order_items = instance.items.all()
for order_item in order_items:
order_item.status = OrderItem.Status.RECEIVED
order_item.save()
So what it does is to check, upon saving, if an Order is set to Complete and if that is the case it loops through all related OrderItem instances, setting them to Received.
However this causes an infinite execution loop since we have another signal that listens on OrderItem upon saving so our signals will recursively execute and Django will show the following screen.
That seems to spoil our fun little party :(
I had to do some research and even ended upon consulting with ChatGPT so that I can get a way out. After a number of iterations, I learnt that we can also connect and disconnect from signal at will!
This looked like it would solve this whole recursive mess. So armed with this new knowledge, I made the following code edits to toggle the signals on and off.
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Order, OrderItem
# Disable signal during certain operations to prevent recursion
def disable_signals():
post_save.disconnect(update_order_status, sender=OrderItem)
post_save.disconnect(mark_order_items_received, sender=Order)
def enable_signals():
post_save.connect(update_order_status, sender=OrderItem)
post_save.connect(mark_order_items_received, sender=Order)
@receiver(post_save, sender=OrderItem)
def update_order_status(sender, instance, **kwargs):
order = instance.order
if order.items.filter(status=OrderItem.Status.PENDING).exists():
if order.status == Order.Status.COMPLETE:
order.status = Order.Status.PENDING
order.save()
return
# If all items are received, update the status of the order to Complete
order.status = Order.Status.COMPLETE
order.save()
@receiver(post_save, sender=Order)
def mark_order_items_received(sender, instance, **kwargs):
if instance.status == Order.Status.COMPLETE:
# Disable signal to prevent recursion
disable_signals()
order_items = instance.items.all()
for order_item in order_items:
order_item.status = OrderItem.Status.RECEIVED
order_item.save()
# Enable signal after performing the operation
enable_signals()
In this updated code,I have defined two methods:
# Disable signal during certain operations to prevent recursion
def disable_signals():
post_save.disconnect(update_order_status, sender=OrderItem)
post_save.disconnect(mark_order_items_received, sender=Order)
Which essentially toggles of the two signals functions.
def enable_signals():
post_save.connect(update_order_status, sender=OrderItem)
post_save.connect(mark_order_items_received, sender=Order)
Which turns them back on.
Since the problem spawned from the second function, that is where I will be placing the toggle functions. Only if an Order status is Complete. The rest of the code remains the same.
And as predicted, this solution fixed the infinite loop of execution that I was facing before :).
Before moving on to the next thing, lest I forget I decided to document my experience for posterity haha😂.
See you on the next one.
Top comments (0)