Backend Live Coding

# ======================================================================
# Q: How do you handle errors consistently across the whole API?
#
# One error schema for the whole API (ideally RFC 9457 Problem Details).
# Business exceptions become proper status codes. 500s never show a stack trace
# to the client, they go to Sentry.
# ======================================================================

def api_exception_handler(exc, context):
    response = exception_handler(exc, context)      # DRF default
    if isinstance(exc, OutOfStock):
        return Response({'type': 'out_of_stock', 'title': 'Product out of stock',
                         'detail': str(exc)}, status=409)
    if response is not None:
        response.data = {'type': 'validation_error' if response.status_code == 400 else 'error',
                         'errors': response.data}
    return response

REST_FRAMEWORK = {'EXCEPTION_HANDLER': 'api.errors.api_exception_handler'}

# ======================================================================
# Q: Write a custom permission: a user sees only the orders of their own
# company, and only a manager can cancel.
#
# has_object_permission is called only by get_object(), not on list endpoints.
# That is why filtering in get_queryset() is mandatory.
# ======================================================================

class IsCompanyManager(BasePermission):
    def has_permission(self, request, view):
        return request.user.is_authenticated

    def has_object_permission(self, request, view, obj):
        if obj.customer.company_id != request.user.company_id:
            return False
        if view.action == 'cancel':
            return request.user.role == 'manager'
        return True

class OrderViewSet(viewsets.ReadOnlyModelViewSet):
    permission_classes = [IsCompanyManager]
    serializer_class = OrderSerializer

    def get_queryset(self):     # the first and most important filter
        return (Order.objects.filter(customer__company_id=self.request.user.company_id)
                .select_related('customer').prefetch_related('items__product'))

    @action(detail=True, methods=['post'])
    def cancel(self, request, pk=None):
        order = self.get_object()          # calls has_object_permission
        services.cancel_order(order, by=request.user)
        return Response(status=status.HTTP_204_NO_CONTENT)

# ======================================================================
# Q: Write a serializer for an order with nested items (creation and
# validation).
#
# Nested writable serializers need your own create()/update(). The logic (stock,
# prices, transaction) goes in a service, not in the serializer. The price does
# not come from the client, the server calculates it.
# ======================================================================

class OrderItemSerializer(serializers.ModelSerializer):
    class Meta:
        model = OrderItem
        fields = ['product', 'quantity']

class OrderCreateSerializer(serializers.ModelSerializer):
    items = OrderItemSerializer(many=True)

    class Meta:
        model = Order
        fields = ['id', 'items']

    def validate_items(self, items):
        if not items:
            raise serializers.ValidationError('Order must have at least one item.')
        skus = [i['product'].pk for i in items]
        if len(skus) != len(set(skus)):
            raise serializers.ValidationError('Duplicate products.')
        return items

    def create(self, validated_data):
        customer = self.context['request'].user.customer
        return services.place_order(customer, validated_data['items'])  # atomic + stock

# ======================================================================
# Q: How do you write a management command for a large job?
#
# Batches with small transactions, --dry-run, progress in the output, and
# idempotent so that it can be rerun if interrupted.
# ======================================================================

class Command(BaseCommand):
    help = 'Recalculate customer totals'

    def add_arguments(self, parser):
        parser.add_argument('--batch-size', type=int, default=1000)
        parser.add_argument('--dry-run', action='store_true')

    def handle(self, *args, batch_size, dry_run, **options):
        qs = Customer.objects.order_by('pk')
        for i, chunk in enumerate(read_in_chunks(qs, batch_size), 1):
            if not dry_run:
                with transaction.atomic():
                    recalc_totals(chunk)
            self.stdout.write(f'batch {i} done')

# ======================================================================
# Q: Custom QuerySet and Manager: write one.
# ======================================================================

class OrderQuerySet(models.QuerySet):
    def paid(self):
        return self.filter(status='paid')

    def for_company(self, company):
        return self.filter(customer__company=company)

    def with_totals(self):
        return self.annotate(total=Sum(F('items__quantity') * F('items__unit_price')))

class Order(models.Model):
    ...
    objects = OrderQuerySet.as_manager()

Order.objects.for_company(c).paid().with_totals()   # chainable