Backend Live Coding

# ======================================================================
# Q: Write a state machine for an order's status with allowed transitions.
#
# The transitions are data (a dict), not if-else scattered across views, so they
# are visible at a glance and can be tested exhaustively (for every pair of
# statuses I check allowed/forbidden). select_for_update prevents two concurrent
# transitions (e.g. a payment webhook and a cancellation by the customer). The
# status changes only from this function: no order.status = ... anywhere else.
# There is django-fsm as an alternative, but for six states I prefer explicit
# code.
# ======================================================================

from enum import StrEnum
from django.db import transaction
from django.utils import timezone

class Status(StrEnum):
    DRAFT = 'draft'
    PLACED = 'placed'
    PAID = 'paid'
    SHIPPED = 'shipped'
    DELIVERED = 'delivered'
    CANCELLED = 'cancelled'

TRANSITIONS: dict[Status, set[Status]] = {
    Status.DRAFT:     {Status.PLACED, Status.CANCELLED},
    Status.PLACED:    {Status.PAID, Status.CANCELLED},
    Status.PAID:      {Status.SHIPPED, Status.CANCELLED},    # cancel => refund
    Status.SHIPPED:   {Status.DELIVERED},
    Status.DELIVERED: set(),
    Status.CANCELLED: set(),
}

class InvalidTransition(Exception):
    pass

@transaction.atomic
def transition(order_id: int, target: Status, actor: User) -> Order:
    order = Order.objects.select_for_update().get(pk=order_id)   # serialization
    current = Status(order.status)
    if target not in TRANSITIONS[current]:
        raise InvalidTransition(f'{current} -> {target}')
    order.status = target
    order.save(update_fields=['status', 'updated_at'])
    OrderEvent.objects.create(order=order, from_status=current, to_status=target,
                              actor=actor, at=timezone.now())       # audit trail
    if target is Status.CANCELLED and current is Status.PAID:
        transaction.on_commit(lambda: refund_payment.delay(order.id))
    if target is Status.SHIPPED:
        transaction.on_commit(lambda: send_tracking_email.delay(order.id))
    return order

# ======================================================================
# Q: How do you test such a rule so that the business trusts it too?
#
# The table of cases is essentially the table I agreed with the sales
# department. I also put it in the PR. For properties that must always hold
# (e.g. gross == net + vat, price never negative, a larger quantity never gives
# a higher unit price) I add property-based tests with hypothesis. In the ASP
# migration I also ran a parallel run: the same orders through the old and the
# new system, comparing the results before the cut-over.
# ======================================================================

D = Decimal
BASE = dict(list_price=D('10.00'), customer_price=None,
            volume_tiers=((50, D('0.05')), (100, D('0.08'))),
            promo_pct=None, vat_rate=D('0.24'))

@pytest.mark.parametrize('overrides, qty, unit, gross, rule', [
    ({},                                10,  D('10.0000'), D('124.00'),  'list'),
    ({},                                50,  D('9.5000'),  D('589.00'),  'volume_0.05'),
    ({},                                120, D('9.2000'),  D('1368.96'), 'volume_0.08'),
    ({'promo_pct': D('0.15')},          120, D('8.5000'),  D('1264.80'), 'promo'),
    ({'customer_price': D('7.3333')},   3,   D('7.3333'),  D('27.28'),   'customer_price'),
])
def test_price_line(overrides, qty, unit, gross, rule):
    p = price_line(PriceContext(**{**BASE, **overrides}), qty)
    assert (p.unit_net, p.gross, p.rule) == (unit, gross, rule)

# ======================================================================
# Q: Implement B2B pricing: customer price > volume tier discount > promo, then
# VAT and rounding.
#
# Points I explain:
# - Decimal everywhere and explicit ROUND_HALF_UP (Python's default is
#   ROUND_HALF_EVEN, which does not agree with the ERP/accounting). Unit price
#   with 4 decimals, amounts with 2.
# - The order of precedence and whether discounts stack is a business decision,
#   not mine. I write it down explicitly and confirm it in writing.
# - Pure function: no database access. The caller loads the PriceContext (one
#   query for all lines).
# - The rule is stored on the order line: when the customer asks "why did I pay
#   this?", there is an answer.
# - VAT per line or on the total gives a one-cent difference: I follow whatever
#   the ERP does, so the invoices agree.
# ======================================================================

from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP

CENT = Decimal('0.01')

@dataclass(frozen=True)
class PriceContext:
    list_price: Decimal                    # list price, excluding VAT
    customer_price: Decimal | None         # special price from the customer's price list
    volume_tiers: tuple[tuple[int, Decimal], ...]   # ((50, 0.05), (100, 0.08))
    promo_pct: Decimal | None              # active promo, if any
    vat_rate: Decimal                      # e.g. Decimal('0.24')

@dataclass(frozen=True)
class LinePrice:
    unit_net: Decimal
    net: Decimal
    vat: Decimal
    gross: Decimal
    rule: str                              # which rule was applied (audit)

def price_line(ctx: PriceContext, qty: int) -> LinePrice:
    if qty <= 0:
        raise ValueError('qty must be positive')
    if ctx.customer_price is not None:     # 1. agreed price: does not stack
        unit, rule = ctx.customer_price, 'customer_price'
    else:
        unit, rule = ctx.list_price, 'list'
        tier = max((pct for min_q, pct in ctx.volume_tiers if qty >= min_q),
                   default=Decimal('0'))
        if tier:                           # 2. volume tier discount
            unit, rule = unit * (1 - tier), f'volume_{tier}'
        if ctx.promo_pct and ctx.promo_pct > tier:   # 3. promo only if it is better
            unit, rule = ctx.list_price * (1 - ctx.promo_pct), 'promo'
    unit = unit.quantize(Decimal('0.0001'), ROUND_HALF_UP)
    net = (unit * qty).quantize(CENT, ROUND_HALF_UP)
    vat = (net * ctx.vat_rate).quantize(CENT, ROUND_HALF_UP)   # VAT per line
    return LinePrice(unit, net, vat, net + vat, rule)

# ======================================================================
# Q: The memory of a worker keeps growing. How do you find the memory leak?
#
# I compare two snapshots and see which line makes allocations that are not
# freed. Usual culprits in Django: a module-level dict used as an unbounded
# cache, DEBUG=True which keeps all queries in connection.queries, iterating a
# huge QuerySet without .iterator() which fills the result cache. Mitigation in
# Celery: worker_max_tasks_per_child or worker_max_memory_per_child, but that is
# a band-aid, not a solution.
# ======================================================================

import tracemalloc

tracemalloc.start(25)                      # keeps 25 frames per allocation
snap1 = tracemalloc.take_snapshot()

for _ in range(1000):
    process_batch()                        # suspect code

snap2 = tracemalloc.take_snapshot()
for stat in snap2.compare_to(snap1, 'lineno')[:10]:
    print(stat)
# pricing/cache.py:31: size=48.2 MiB (+48.1 MiB), count=410233 (+410000)

-- ======================================================================
-- Q: A query or a migration is stuck. How do you find who is blocking whom?
--
-- The classic scenario: a connection that is idle in transaction (someone ran
-- BEGIN and forgot, or code that calls an external API inside atomic) holds a
-- lock, an ALTER TABLE waits for ACCESS EXCLUSIVE, and all the SELECTs queue up
-- behind it. Prevention: lock_timeout on migrations (SET lock_timeout = '5s'),
-- idle_in_transaction_session_timeout and statement_timeout on the application's
-- connection.
-- ======================================================================

-- what is running now and for how long
SELECT pid, state, wait_event_type, wait_event,
       now() - xact_start AS xact_age, left(query, 80) AS query
FROM pg_stat_activity
WHERE datname = current_database() AND state <> 'idle'
ORDER BY xact_start;

-- who blocks whom
SELECT blocked.pid  AS blocked_pid,  left(blocked.query, 60)  AS blocked_query,
       blocking.pid AS blocking_pid, left(blocking.query, 60) AS blocking_query,
       blocking.state
FROM pg_stat_activity blocked
JOIN LATERAL unnest(pg_blocking_pids(blocked.pid)) AS b(pid) ON true
JOIN pg_stat_activity blocking ON blocking.pid = b.pid;

-- lock details on a specific table
SELECT l.pid, l.mode, l.granted
FROM pg_locks l JOIN pg_class c ON c.oid = l.relation
WHERE c.relname = 'orders';

SELECT pg_cancel_backend(4242);      -- cancels the query
SELECT pg_terminate_backend(4242);   -- closes the connection