Capstone API: Image Upload & ML Simulation

1. The Core Architecture

Before we write any code, let's revisit the architecture we're building. The workflow for our image upload and processing directly maps to the producer-consumer pattern you learned about in Module 7.

In this lesson, we will implement this exact flow:

1. Producer (Django Web App): We'll build a DRF API endpoint that accepts an image file.
2. Task Creation: Upon successful upload, the Django app will create a task and send it to the message...
3. Broker (Redis): ...which will hold the task message until a worker is available.
4. Consumer (Celery Worker): Our worker process will pick up the task, simulate ML processing, and update the database with the result.

This architecture is fundamental to building scalable backend systems, especially for ML-heavy applications like Infilect's, as it prevents long-running tasks from blocking the main web server and degrading user experience.

2. Capstone Project and Model Setup

First, let's set up the new environment for our capstone project.

1. Create a new Django project. Let's call it capstone_backend.
2. Inside, create a new app called api.
3. Add rest_framework, storages, and your new api app to INSTALLED_APPS in settings.py.
4. Configure your settings.py to use AWS S3 for media storage, just as you did in Module 10. Remember to set your AWS_STORAGE_BUCKET_NAME, AWS_S3_ACCESS_KEY_ID, etc., in your environment variables.
5. Ensure your CELERY_BROKER_URL is also configured to point to your Redis instance.

With the project set up, let's define the central model for this workflow. In api/models.py, create a ShelfImage model. This model will store the uploaded image and track its processing status.

# api/models.py
from django.db import models

class ShelfImage(models.Model):
    class Status(models.TextChoices):
        PENDING = 'PENDING', 'Pending'
        PROCESSING = 'PROCESSING', 'Processing'
        COMPLETED = 'COMPLETED', 'Completed'
        FAILED = 'FAILED', 'Failed'




    # The uploaded image file, will be stored in S3
    image = models.ImageField(upload_to='shelf_images/')
    



    # The current status of the processing task
    status = models.CharField(
        max_length=20,
        choices=Status.choices,
        default=Status.PENDING
    )
    



    # Timestamps
    uploaded_at = models.DateTimeField(auto_now_add=True)
    processed_at = models.DateTimeField(null=True, blank=True)

    def __str__(self):
        return f"Image {self.id} - {self.status}"

This model includes a status field that acts as a simple state machine. This is crucial for tracking the lifecycle of the image as it moves through our asynchronous pipeline.

After defining the model, create and run the migrations:

python manage.py makemigrations
python manage.py migrate

3. Building the Image Upload API

Next, we'll create the DRF API endpoint to handle the file upload. This requires a serializer to validate the data and a view to orchestrate the logic.

The following video provides an excellent walkthrough of creating file upload endpoints with DRF. While you've done this before, a focused refresher is always valuable.

Now, let's apply those concepts.

1. Create the Serializer

In a new file, api/serializers.py, define a serializer for our ShelfImage model. We'll make most fields read-only since they will be set by the system, not the user.

# api/serializers.py
from rest_framework import serializers
from .models import ShelfImage

class ShelfImageSerializer(serializers.ModelSerializer):
    class Meta:
        model = ShelfImage
        fields = ['id', 'image', 'status', 'uploaded_at', 'processed_at']
        read_only_fields = ['id', 'status', 'uploaded_at', 'processed_at']

2. Create the View

In api/views.py, create the APIView that will use this serializer.

# api/views.py
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from rest_framework.parsers import MultiPartParser, FormParser

from .serializers import ShelfImageSerializer

class ShelfImageUploadView(APIView):
    parser_classes = (MultiPartParser, FormParser)

    def post(self, request, *args, **kwargs):



        # We pass request.data to the serializer
        serializer = ShelfImageSerializer(data=request.data)
        
        if serializer.is_valid():



            # If valid, serializer.save() will create a new ShelfImage instance
            saved_instance = serializer.save()
            



            # TODO: Trigger the async task here!
            



            # Return the data of the created instance
            return Response(serializer.data, status=status.HTTP_201_CREATED)
        else:



            # If invalid, return the errors
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

Notice the parser_classes attribute. As the video explained, this tells DRF to expect multipart/form-data, which is the standard for requests containing files.

3. Wire up the URL

Finally, create api/urls.py and connect it to your project's main urls.py file.

# api/urls.py
from django.urls import path
from .views import ShelfImageUploadView

urlpatterns = [
    path('images/upload/', ShelfImageUploadView.as_view(), name='image-upload'),
]




# In capstone_backend/urls.py
from django.urls import path, include

urlpatterns = [



    # ... other urls
    path('api/', include('api.urls')),
]

At this point, you have a working API endpoint. If you were to test it, it would accept an image, save a ShelfImage record to your database, upload the file to S3, and return a 201 Created response. But we're missing the final piece: the asynchronous task.

4. Implementing the Asynchronous ML Simulation Task

Why can't we just run the ML model inside the post method of our view? The following article explains the core challenges of running ML models in a synchronous web request flow.

Now that you're clear on the "why," let's implement the "how." We'll create a Celery task to simulate the ML processing.

In a new file, api/tasks.py, define your task.

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

@shared_task
def process_shelf_image(image_id):
    """
    Simulates a long-running ML process on a shelf image.
    """
    try:



        # Fetch the image instance from the database
        image_instance = ShelfImage.objects.get(id=image_id)
        



        # 1. Update status to PROCESSING
        image_instance.status = ShelfImage.Status.PROCESSING
        image_instance.save()




        # Simulate the ML model running (e.g., object detection)
        print(f"Processing image {image_id}... This will take a moment.")
        time.sleep(10)  # Simulate a 10-second process




        # 2. Update status to COMPLETED
        image_instance.status = ShelfImage.Status.COMPLETED
        image_instance.processed_at = timezone.now()
        image_instance.save()
        
        print(f"Image {image_id} processed successfully.")

    except ShelfImage.DoesNotExist:
        print(f"Error: ShelfImage with id={image_id} not found.")
    except Exception as e:



        # Handle potential errors during processing
        if 'image_instance' in locals():
            image_instance.status = ShelfImage.Status.FAILED
            image_instance.save()
        print(f"Error processing image {image_id}: {e}")

This task follows a robust pattern:

• It accepts a simple, serializable identifier (image_id).
• It fetches the full object from the database itself.
• It updates the object's status to provide real-time feedback on its progress.
• It includes basic error handling.

5. Triggering the Task

The final step is to connect the API view to the Celery task. We'll use the .delay() method, which you've seen before, to send the task to the Redis broker immediately after the image is saved.

Let's update api/views.py to trigger the task.

# api/views.py (updated)
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from rest_framework.parsers import MultiPartParser, FormParser

from .serializers import ShelfImageSerializer
from .tasks import process_shelf_image  # Import the task

class ShelfImageUploadView(APIView):
    parser_classes = (MultiPartParser, FormParser)

    def post(self, request, *args, **kwargs):
        serializer = ShelfImageSerializer(data=request.data)
        
        if serializer.is_valid():
            saved_instance = serializer.save()
            



            # *** Trigger the async task here! ***
            process_shelf_image.delay(saved_instance.id)
            
            return Response(serializer.data, status=status.HTTP_201_CREATED)
        else:
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

And that's it! With this one line, you've connected the synchronous web world to the asynchronous worker world. The API can now return a response to the user in milliseconds, while the 10-second processing task runs completely in the background.

This pattern—saving the model and passing its ID to a Celery task—is a clean, effective, and widely-used approach for offloading work in a web application.

6. Hands-On Task: Implementation and Testing

You now have all the pieces. Your task is to implement the entire workflow in your capstone_backend project.

Checklist:

1. Set up the Django project, api app, and configure settings.py.
2. Define the ShelfImage model and run migrations.
3. Create ShelfImageSerializer.
4. Create api/tasks.py and define the process_shelf_image task.
5. Create ShelfImageUploadView and wire it up in urls.py. Remember to import and call your task!
6. Ensure your Celery __init__.py and celery.py files are set up correctly in your project directory (as covered in Module 7).

Testing Your Implementation:

1. In one terminal, start your Django development server and Celery worker using docker-compose up. Make sure you have services for backend, worker, db, and cache (or mq).
2. Open a tool like Postman or Insomnia.
3. Create a new POST request to your endpoint (e.g., http://localhost:8000/api/images/upload/).
4. Set the body type to "Multipart Form".
5. Add a field named image and set its type to "File". Select an image from your computer.
6. Send the request.

Expected Outcome:

• You should get an immediate 201 Created response with the JSON data of the new ShelfImage instance, with status: "PENDING".
• In your terminal, the Celery worker logs should show that it received and started processing the task api.tasks.process_shelf_image.
• After 10 seconds, the worker logs should print "Image X processed successfully."
• If you check your S3 bucket, the image file should be there.
• If you inspect the database (e.g., via the Django admin), the status of the ShelfImage object will have changed from PENDING to COMPLETED, and processed_at will be populated.

Conclusion

Congratulations! You have successfully built the foundational feature of your capstone project. You've created a robust API endpoint that ingests data, stores it securely, and kicks off a long-running background process without blocking the user. This single workflow demonstrates your ability to integrate multiple core backend technologies (Django, DRF, Celery, Redis, S3) to solve a real-world problem.

Key Takeaways:

• Producer-Consumer Pattern: You implemented this pattern with Django as the producer and Celery as the consumer.
• Asynchronous Triggering: You learned to use task.delay(object_id) to offload work from the web thread, passing a simple ID instead of a large object.
• State Management: You used a status field in your model to track the progress of a background job, which is essential for providing feedback to users or other systems.

In the next lesson, we will build upon this foundation. You will create the APIs needed to manage the annotation workflow, implementing the state machine you designed and building endpoints for analysts to retrieve and update annotation data.