Building and Deploying Annotation and Analytics APIs

1. Modeling the Annotation Workflow

Our previous ShelfImage model tracks the processing status of an image. Now, we need a model to store the results of that processing. In a real system, an ML model would detect products on a shelf and create bounding boxes. Our Celery task will simulate this by creating records in a new Annotation model.

Let's define what an Annotation represents. It's a piece of data that identifies something in an image. It has its own lifecycle, separate from the image processing itself. A typical flow is: the ML model SUGGESTS an annotation, a human operator reviews it and marks it as CONFIRMED or REJECTED.

In your api/models.py, add the following Annotation model:

# api/models.py




# ... (keep the ShelfImage model) ...

class Annotation(models.Model):
    class Status(models.TextChoices):
        SUGGESTED = 'SUGGESTED', 'Suggested'
        CONFIRMED = 'CONFIRMED', 'Confirmed'
        REJECTED = 'REJECTED', 'Rejected'




    # Link back to the image it belongs to
    shelf_image = models.ForeignKey(
        ShelfImage, 
        related_name='annotations', 
        on_delete=models.CASCADE
    )
    



    # What the ML model thinks this object is
    product_name = models.CharField(max_length=100)
    



    # The coordinates of the detected object, stored as JSON
    # e.g., {"x": 100, "y": 150, "width": 50, "height": 80}
    bounding_box = models.JSONField()
    



    # The status of this specific annotation
    status = models.CharField(
        max_length=20,
        choices=Status.choices,
        default=Status.SUGGESTED
    )
    
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    def __str__(self):
        return f"Annotation for '{self.product_name}' in Image {self.shelf_image.id}"

Note the related_name='annotations' on the ForeignKey. This is a powerful Django feature that will allow us to easily access all annotations for a given ShelfImage instance using shelf_image.annotations.all().

After adding the model, create and run the migrations:

python manage.py makemigrations
python manage.py migrate

2. Updating the Celery Task to Generate Annotations

Now, let's make our simulated ML task more realistic. Instead of just sleeping, it will now populate the database with a few dummy Annotation objects. This mimics an object detection model identifying products in the uploaded image.

Update your api/tasks.py file:

# api/tasks.py
import time
import random
from celery import shared_task
from django.utils import timezone
from .models import ShelfImage, Annotation # Import Annotation

@shared_task
def process_shelf_image(image_id):
    """
    Simulates a long-running ML process and creates initial annotations.
    """
    try:
        image_instance = ShelfImage.objects.get(id=image_id)
        image_instance.status = ShelfImage.Status.PROCESSING
        image_instance.save()

        print(f"Processing image {image_id}... This will take a moment.")
        time.sleep(10)




        # *** Simulate creating annotations ***
        products = ["Cola", "Pepsi", "Sprite", "Fanta"]
        for i in range(random.randint(2, 5)): # Create 2 to 5 annotations
            Annotation.objects.create(
                shelf_image=image_instance,
                product_name=random.choice(products),
                bounding_box={
                    "x": random.randint(10, 300),
                    "y": random.randint(10, 400),
                    "width": random.randint(50, 150),
                    "height": random.randint(50, 150)
                },



                # The initial status is 'SUGGESTED' by default
            )
        
        image_instance.status = ShelfImage.Status.COMPLETED
        image_instance.processed_at = timezone.now()
        image_instance.save()
        
        print(f"Image {image_id} processed successfully and annotations created.")

    except ShelfImage.DoesNotExist:



        # ... (error handling remains the same) ...
        print(f"Error: ShelfImage with id={image_id} not found.")
    except Exception as e:
        if 'image_instance' in locals():
            image_instance.status = ShelfImage.Status.FAILED
            image_instance.save()
        print(f"Error processing image {image_id}: {e}")

Now, every time an image is processed, it will have a set of "suggested" annotations ready for review.

3. Building the Annotation Management API

We need endpoints to let a data operator view these suggested annotations and update their status. A ModelViewSet is perfect for this. The following article provides an excellent guide on extending models and viewsets, which is exactly what we're doing.

Let's apply these ideas.

1. Create the AnnotationSerializer

In api/serializers.py, define a serializer for the Annotation model. The user should only be able to update the status.

# api/serializers.py
# ... (ShelfImageSerializer remains) ...
from .models import Annotation # Add this import

class AnnotationSerializer(serializers.ModelSerializer):
    class Meta:
        model = Annotation
        fields = [
            'id', 'shelf_image', 'product_name', 
            'bounding_box', 'status', 'created_at', 'updated_at'
        ]
        read_only_fields = [
            'id', 'shelf_image', 'product_name', 
            'bounding_box', 'created_at', 'updated_at'
        ]

By making most fields read_only, we ensure that an annotator can only change the status via a PATCH request.

2. Create the AnnotationViewSet

We will create a ViewSet that allows list, retrieve, and update operations. We'll nest this ViewSet under the images endpoint, so the URL structure will be /api/images/<image_pk>/annotations/. This is a very common and RESTful pattern for representing nested resources.

Update api/views.py:

# api/views.py
from rest_framework import viewsets
from .models import Annotation
from .serializers import AnnotationSerializer # import new serializer




# ... (ShelfImageUploadView remains for now) ...

class AnnotationViewSet(viewsets.ModelViewSet):
    """
    ViewSet for viewing and editing annotations.
    """
    serializer_class = AnnotationSerializer
    queryset = Annotation.objects.all()
    http_method_names = ['get', 'patch', 'head', 'options'] # Allow list, retrieve, partial_update

    def get_queryset(self):
        """
        This view should return a list of all the annotations
        for the shelf image as determined by the image_pk portion of the URL.
        """
        image_pk = self.kwargs['image_pk']
        return Annotation.objects.filter(shelf_image__id=image_pk)

Notice we override get_queryset to filter annotations based on the image_pk from the URL. We also limit the allowed HTTP methods, as creating new annotations is the job of the Celery task, and deleting them might not be a desired user action.

3. Set up Nested URLs

To create the /images/<image_pk>/annotations/ URL structure, we need a nested router. The drf-nested-routers library is excellent for this, but to keep dependencies minimal, we can configure it manually for now. Let's adjust our api/urls.py and capstone_backend/urls.py to use DRF's DefaultRouter.

First, let's create a ViewSet for ShelfImage so we can have a router for it. Replace ShelfImageUploadView in api/views.py:

# api/views.py
from rest_framework import viewsets, status
from rest_framework.response import Response
from .models import ShelfImage
from .serializers import ShelfImageSerializer
from .tasks import process_shelf_image

class ShelfImageViewSet(viewsets.ModelViewSet):
    """
    ViewSet for listing and creating ShelfImages.
    """
    queryset = ShelfImage.objects.all().order_by('-uploaded_at')
    serializer_class = ShelfImageSerializer
    http_method_names = ['get', 'post', 'head', 'options']

    def perform_create(self, serializer):



        # Save the instance first
        instance = serializer.save()



        # Then, trigger the Celery task
        process_shelf_image.delay(instance.id)




# The old ShelfImageUploadView can now be removed.

This ViewSet handles both listing images and creating them (our upload functionality). The perform_create hook is the standard DRF way to add extra logic after an object is created.

Now, update api/urls.py to use a router and include the nested annotation routes:

# api/urls.py
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import ShelfImageViewSet, AnnotationViewSet




# Create a router and register our viewsets with it.
router = DefaultRouter()
router.register(r'images', ShelfImageViewSet)




# The API URLs are now determined automatically by the router.
# We manually add the nested route for annotations.
urlpatterns = [
    path('', include(router.urls)),
    path('images/<int:image_pk>/annotations/', AnnotationViewSet.as_view({
        'get': 'list'
    }), name='image-annotations-list'),
    path('images/<int:image_pk>/annotations/<int:pk>/', AnnotationViewSet.as_view({
        'get': 'retrieve',
        'patch': 'partial_update'
    }), name='image-annotations-detail'),
]

This setup gives us the desired nested API structure.

4. Serving Aggregated Analytics KPIs

Now for the most exciting part: turning data into insights. We'll create an endpoint at /api/images/<pk>/analytics/ that returns a summary of the annotations for a specific image.

This is not a standard CRUD operation. It requires custom logic and a custom response structure. The best way to add such an endpoint to an existing ViewSet is with a custom @action.

This video from BugBytes is a fantastic guide on using custom actions, as well as filtering, which is highly relevant for analytics.

To structure the response from our analytics endpoint, we'll use a generic serializers.Serializer instead of a ModelSerializer. This is ideal for representing aggregated data that doesn't map to a single model.

Let's combine these two patterns.

1. Create the AnalyticsSerializer

In api/serializers.py, add this generic serializer:

# api/serializers.py
# ...

class ShelfAnalyticsSerializer(serializers.Serializer):
    """
    Serializer for the aggregated annotation analytics.
    """
    total_annotations = serializers.IntegerField()
    suggested_count = serializers.IntegerField()
    confirmed_count = serializers.IntegerField()
    rejected_count = serializers.IntegerField()




    # You could add more complex KPIs here, like:
    # compliance_score = serializers.FloatField()
    # share_of_shelf_by_brand = serializers.DictField()

2. Add the Custom Action to ShelfImageViewSet

Now, add the @action to your ShelfImageViewSet in api/views.py. We'll use Django's ORM aggregation capabilities to calculate the stats.

# api/views.py
from django.db.models import Count, Q
from rest_framework.decorators import action
from .serializers import ShelfAnalyticsSerializer # Add this import

class ShelfImageViewSet(viewsets.ModelViewSet):



    # ... (existing code) ...

    @action(detail=True, methods=['get'])
    def analytics(self, request, pk=None):
        """
        Custom action to return analytics for a single shelf image.
        """
        shelf_image = self.get_object() # Gets the ShelfImage instance for the given pk




        # Use Django's aggregation to count annotations by status
        queryset = shelf_image.annotations.all()
        



        # This is an efficient way to get all counts in one DB query
        stats = queryset.aggregate(
            total_annotations=Count('id'),
            suggested_count=Count('id', filter=Q(status=Annotation.Status.SUGGESTED)),
            confirmed_count=Count('id', filter=Q(status=Annotation.Status.CONFIRMED)),
            rejected_count=Count('id', filter=Q(status=Annotation.Status.REJECTED))
        )

        serializer = ShelfAnalyticsSerializer(data=stats)
        serializer.is_valid(raise_exception=True) # This will always be valid
        
        return Response(serializer.validated_data)

This is a highly efficient way to compute KPIs. The aggregate function performs all the counting in a single database query, which is critical for performance. The Q objects allow for conditional aggregation.

5. Hands-On Task: Implementation and Testing

Your task is to implement the full annotation and analytics workflow.

Part A: Annotation Workflow

1. Add the Annotation model to api/models.py and run migrations.
2. Update the process_shelf_image task in api/tasks.py to create dummy annotations.
3. Add the AnnotationSerializer to api/serializers.py.
4. Replace ShelfImageUploadView with ShelfImageViewSet in api/views.py.
5. Add the AnnotationViewSet to api/views.py.
6. Update api/urls.py to use the DefaultRouter and include the nested annotation routes.

Part B: Analytics Endpoint

1. Add the generic ShelfAnalyticsSerializer to api/serializers.py.
2. Add the analytics custom action, decorated with @action, to your ShelfImageViewSet in api/views.py.

Testing Your Implementation:

1. Start your services: docker-compose up.
2. Upload an image: Send a POST request to http://localhost:8000/api/images/ with an image file. You should get a 201 Created response.
3. Wait for processing: Observe the Celery worker logs. After about 10 seconds, you should see the message "Image X processed successfully and annotations created."
4. View annotations: Send a GET request to http://localhost:8000/api/images/<image_pk>/annotations/ (replace <image_pk> with the ID from step 2). You should see a list of JSON objects representing the suggested annotations.
5. Update an annotation: Send a PATCH request to http://localhost:8000/api/images/<image_pk>/annotations/<annotation_pk>/. The request body should be {"status": "CONFIRMED"}. You should get a 200 OK response.
6. Check analytics: Send a GET request to http://localhost:8000/api/images/<image_pk>/analytics/. You should receive a JSON response with the aggregated counts, which should reflect the change you made in the previous step. For example:

   {
       "total_annotations": 4,
       "suggested_count": 3,
       "confirmed_count": 1,
       "rejected_count": 0
   }
   

Conclusion

Excellent work. You have now built the most complex and business-critical parts of the capstone project. You've created a complete, end-to-end workflow that not only processes data asynchronously but also provides the APIs to manage the resulting annotations and derive valuable business intelligence from them.

Key Takeaways:

• Nested API Design: You structured your API logically using nested routes to represent parent-child relationships between data (ShelfImage -> Annotation).
• Custom ViewSet Actions: You learned to extend the standard functionality of a ViewSet with custom actions (@action) to provide specialized endpoints like /analytics/.
• Efficient Aggregation: You used Django ORM's aggregate(), Count(), and Q() objects to perform complex calculations efficiently at the database level.
• Generic Serializers: You used a generic serializers.Serializer to define a custom, read-only data structure for your analytics response.

You have now directly implemented features that are central to the role at Infilect. In our next lesson, we will focus on ensuring the robustness of this entire system by writing an end-to-end integration test that validates the complete workflow, from image upload to analytics retrieval.