Refactor a fat Django view into a testable service

Backend Live Coding November 13, 2026 python

What I gained: the pricing rule is tested with plain pytest in milliseconds.

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
# ======================================================================
# Q: Take this "fat" Django view and make it testable.
#
# What I gained: the pricing rule is tested with plain pytest in milliseconds.
# The service has a transaction and clean domain exceptions. The ERP and the
# email go out via Celery with on_commit, so we do not send an order to the ERP
# if it was rolled back, and the user does not wait for the ERP. I also fixed
# the N+1 and the float. The same service is called by the view, a management
# command or a Celery task.
# ======================================================================

# BEFORE: everything inside the view, impossible to test without HTTP, DB and ERP
@api_view(['POST'])
def place_order(request):
    cart = Cart.objects.get(user=request.user)
    total = 0
    for item in cart.items.all():                      # N+1
        price = item.product.price
        if request.user.customer.is_b2b:
            price = price * 0.9
        total += price * item.qty                      # float
    if total > request.user.customer.credit_limit:
        return Response({'error': 'credit'}, status=400)
    order = Order.objects.create(customer=request.user.customer, total=total)
    requests.post(ERP_URL, json={'id': order.id, 'total': total})   # inside the request
    send_mail('Order', f'#{order.id}', None, [request.user.email])
    return Response({'id': order.id})



# AFTER: pure rule + service + thin view + async side-effects
# pricing.py (pure, no Django)
def order_total(lines: list[Line], discount: Decimal) -> Decimal:
    gross = sum((l.unit_price * l.qty for l in lines), Decimal('0'))
    return (gross * (1 - discount)).quantize(Decimal('0.01'), ROUND_HALF_UP)

# services.py
class CreditLimitExceeded(Exception): ...

@transaction.atomic
def place_order(customer: Customer, cart: Cart) -> Order:
    lines = [Line(i.product.price, i.qty)
             for i in cart.items.select_related('product')]
    total = order_total(lines, customer.discount_rate)
    if total > customer.credit_limit:
        raise CreditLimitExceeded(total)
    order = Order.objects.create(customer=customer, total=total)
    transaction.on_commit(lambda: push_order_to_erp.delay(order.id))
    transaction.on_commit(lambda: send_order_email.delay(order.id))
    return order

# views.py
@api_view(['POST'])
def place_order_view(request):
    try:
        order = services.place_order(request.user.customer,
                                     Cart.objects.get(user=request.user))
    except CreditLimitExceeded:
        return Response({'code': 'credit_limit_exceeded'}, status=422)
    return Response({'id': order.id}, status=201)