What I gained: the pricing rule is tested with plain pytest in milliseconds.
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) |