Backend Live Coding

# ======================================================================
# Q: Celery: chain, group, chord. When do you use them?
#
# A chord needs a result backend. I keep task results small (IDs, not large
# objects). For large workflows I prefer state in the database, so the workflow
# can continue after a failure.
# ======================================================================

from celery import chain, group, chord

chain(fetch_orders.s(day), build_report.s(), email_report.s(to))()   # sequential, passes the result along
group(sync_supplier.s(s.id) for s in suppliers)()                    # parallel
chord((price_for_region.s(r) for r in regions), merge_prices.s())()   # parallel, then a callback

# ======================================================================
# Q: How do you implement a database-backed job queue that works with many
# workers at the same time?
#
# SKIP LOCKED makes each worker pick different rows without waiting. The same
# pattern is used in the outbox relay.
# ======================================================================

with transaction.atomic():
    jobs = list(Job.objects.select_for_update(skip_locked=True)
                .filter(status='queued').order_by('id')[:10])
    for job in jobs:
        job.status = 'running'
    Job.objects.bulk_update(jobs, ['status'])
# processing happens after the commit

# ======================================================================
# Q: Find the bug.
#
# Two problems. 1) The Celery task may run before the commit, fail to find the
# order, or send an email for an order that was eventually rolled back. Fix:
# transaction.on_commit(lambda: send_confirmation_email.delay(order.id)). 2)
# Charging the card inside a transaction keeps the transaction open while we
# wait for an external system, and if it rolls back after the charge, the
# customer was charged without an order. Correct approach: commit a pending
# order, charge outside the transaction with an idempotency key, then update the
# status.
# ======================================================================

@transaction.atomic
def place_order(customer, items):
    order = Order.objects.create(customer=customer, status='pending')
    ...
    send_confirmation_email.delay(order.id)
    charge_card(order)                 # may raise an exception

# ======================================================================
# Q: Find the bug.
#
# Lost update: two concurrent calls read the same value and one of the changes
# is lost. In addition, save() writes all fields. Fix:
# Customer.objects.filter(pk=customer_id).update(points=F('points') + points).
# If a check is needed before the change: select_for_update() inside
# transaction.atomic().
# ======================================================================

def add_loyalty_points(customer_id, points):
    c = Customer.objects.get(pk=customer_id)
    c.points += points
    c.save()

# ======================================================================
# Q: Design the schema for B2B price lists (a different price per customer, with
# validity dates).
#
# A normalized schema, constraints in the database, and an index for the
# frequent lookup "price of product X for customer Y today". To prevent
# overlapping periods: the PostgreSQL exclusion constraint with daterange
# (ExclusionConstraint in django.contrib.postgres). The price charged is stored
# in OrderItem, because the price list changes.
# ======================================================================

class PriceList(models.Model):
    name = models.CharField(max_length=100)

class PriceListEntry(models.Model):
    price_list = models.ForeignKey(PriceList, on_delete=models.CASCADE)
    product = models.ForeignKey(Product, on_delete=models.CASCADE)
    price = models.DecimalField(max_digits=10, decimal_places=2)
    valid_from = models.DateField()
    valid_to = models.DateField(null=True, blank=True)

    class Meta:
        indexes = [models.Index(fields=['price_list', 'product', '-valid_from'])]
        constraints = [models.CheckConstraint(
            condition=Q(valid_to__isnull=True) | Q(valid_to__gte=F('valid_from')),
            name='pricelist_valid_range')]

class Customer(models.Model):
    price_list = models.ForeignKey(PriceList, null=True, on_delete=models.SET_NULL)