Traffic Monitoring REST API

Context

This project was developed as part of a technical challenge for a job interview. Its primary objective was to have a REST API using Django Rest Framework that serves as the backbone for a road traffic monitoring dashboard, allowing for informed decision-making regarding the traffic intensity recorded on local road segments.

In our world of ever-increasing urbanisation and traffic congestion, having real-time insights into road traffic conditions is crucial. This project addresses this need by providing a comprehensive set of data points for assessing traffic conditions and making informed decisions.

Key Features

The API focuses on geographical representations of road segments, offering the following information:

1. Road Segments

  - Overview of all road segments, with the count of traffic readings through that segment.

  - Detailed view on individual road segments, with the all of the data from the traffic readings recorded in that segment.

  - Specific views that only display data according to the last traffic reading's intensity (high, medium or low).

2. Traffic Readings

  - Overview of all traffic readings, their intensity (calculated based on the speed), and the road segment in which it was recorded.

  - Specific views that only display traffic readings according to their intensity (high, medium or low).

3. Sensors

  - Overview of all sensors available, with their id, name and uuid.

  - Overview of all registered sensor readings, indicating the road segment id, the car license plate and the timestamp.

4. Cars

  - Overview of all cars that were registered through a sensor reading.

  - Detailed view on individual cars (through id or license plate), indicating the car details, and its sensor readings from the last 24h, including data from the road segment and sensor.

Prerequisites

To be able to run this API project, I used the following:

  - Python (version 3.11.5)

  - Django (version 4.2.7)

  - Django Rest Framework (version 3.14)

  - PostgreSQL (version 16.0)

  - psycopg (version 2.9.9)

  - drf-yasg (version 1.21.7)

Get the API

To get this project you can head on to my GitHub repository. There you will find the complete project, as well as instructions on how to set it up and get it running.

Code explained

Developing this project was not as straightforward as the following code snippets, and also not in the order I will be presenting. I am following this structure to be easier to grasp how it is working.

Models

Imports necessary for the traffic_api/models.py file:

from django.db.models import Model, FloatField, AutoField, DateTimeField, CharField, UUIDField, ForeignKey, CASCADE
from .traffic_api_helpers import get_intensity

In this file I defined the models for road segments, traffic readings, sensors, sensor readings and cars:

1 - ROAD SEGMENTS --------------------------------------------------------------------
class RoadSegments(Model):
    id = AutoField(primary_key=True)
    long_start = FloatField()
    lat_start = FloatField()
    long_end = FloatField()
    lat_end = FloatField()
    length = FloatField()

def __str__(self):
    return str(self.id)

def save(self, *args, **kwargs):
    if not self.id:
        # Get the latest id from existing recordings and increment by 1
        last_route = RoadSegments.objects.order_by('-id').first()
        if last_route:
            self.id = last_route.id + 1
        else:
            self.id = 1
        super(RoadSegments, self).save(*args, **kwargs)

This model has the fields necessary to identify a road segment, and two methods that override their defaults:

• The __str__ method, to provide a human-readable string representation of the model instance.
• The save method, to check if the road segment is a new instance, and set the correct value for the id. This should be done automatically, but I was getting id conflicts when trying to create new road segments and couldn’t figure out why, so I decided to override it.

# 2 - TRAFFIC READINGS -----------------------------------------------------------------
class TrafficReadings(Model):
    id = AutoField(primary_key=True)
    speed = FloatField(null=True, blank=True)
    road_segment_id = ForeignKey(RoadSegments, on_delete=CASCADE)
    
    def __str__(self):
        return str(self.id)
    
    @property
    def intensity(self):
        return self.get_intensity(self.speed)
    
    def save(self, *args, **kwargs):
        if not self.id:
            # Get the latest id from existing recordings and increment by 1
            last_route = TrafficReadings.objects.order_by('-id').first()
            if last_route:
                self.id = last_route.id + 1
            else:
                self.id = 1
        super(TrafficReadings, self).save(*args, **kwargs)

This model has the fields necessary to identify a traffic reading (including a foreign key to connect to a road segment), the same two overriding methods as the previous model, and the intensity method.

This method is decorated with @property to be accessed like an attribute, and provides a calculated property based on the speed field. get_intensity is a function defined in the file traffic_api_helpers.py:

from traffic_monitoring.settings import LOW_SPEED_THRESHOLD, HIGH_SPEED_THRESHOLD

def get_intensity(speed):
    if speed is not None:
        if speed <= LOW_SPEED_THRESHOLD:
            intensity = "High"
        elif speed > LOW_SPEED_THRESHOLD and speed <= HIGH_SPEED_THRESHOLD:
            intensity = "Medium"
        else:
            intensity = "Low"
        return intensity
    else:
        return 'No Data'

This function uses the global variables for speed thresholds, and assigns the intensity according to the speed registered.

# 3 - SENSORS --------------------------------------------------------------------------
class Sensors(Model):
    id = AutoField(primary_key=True)
    name = CharField(max_length=100)
    uuid = UUIDField()
    
    def __str__(self):
        return str(self.name)
    
    def save(self, *args, **kwargs):
        if not self.id:
            # Get the latest id from existing recordings and increment by 1
            last_route = Sensors.objects.order_by('-id').first()
            if last_route:
                self.id = last_route.id + 1
            else:
                self.id = 1
            super(Sensors, self).save(*args, **kwargs)

This model has the fields necessary to identify a sensor, and the same two overriding methods as the previous models.

# 4 - CARS -----------------------------------------------------------------------------
class Cars(Model):
    id = AutoField(primary_key=True)
    car_license_plate = CharField(max_length=6)
    created_at = DateTimeField()
    
    @property
    def sensor_readings(self):
        return SensorReadings.objects.filter(car_license_plate=self.value)
    
    def __str__(self):
        return str(self.car_license_plate)

This model has the fields necessary to identify a car, the same __str__ overriding method as the previous models, and the sensor_readings method.

This method is decorated with @property to be accessed as an attribute, and retrieves sensor readings associated with the car’s license plate.

# 5 - SENSOR READINGS ------------------------------------------------------------------
class SensorReadings(Model):
    id = AutoField(primary_key=True)
    road_segment_id = ForeignKey(RoadSegments, on_delete=CASCADE)
    car_license_plate = ForeignKey(Cars, on_delete=CASCADE)
    timestamp = DateTimeField()
    sensor_uuid = ForeignKey(Sensors, on_delete=CASCADE)
    
    def __str__(self):
        return str(self.id)
    
    def save(self, *args, **kwargs):
        if not self.id:
            # Get the latest id from existing recordings and increment by 1
            last_route = SensorReadings.objects.order_by('-id').first()
            if last_route:
                self.id = last_route.id + 1
            else:
                self.id = 1
            super(SensorReadings, self).save(*args, **kwargs)

This model has the fields necessary to identify a sensor reading (including a foreign keys to connect to road segments, cars and sensors), and the same two overriding methods as the previous models.

Serializers

Imports necessary for the traffic_api/serializers.py file:

from rest_framework.serializers import ModelSerializer, SerializerMethodField, CharField, PrimaryKeyRelatedField, FloatField
from django.utils import timezone
from .models import TrafficReadings, RoadSegments, Sensors, Cars, SensorReadings  
from .traffic_api_helpers import get_intensity

This file has 8 serializers to get and create data from the models. Starting with traffic readings:

# 1 - GET TRAFFIC READINGS
class TrafficReadingsSerializer(ModelSerializer):
    road_segment_id = PrimaryKeyRelatedField(queryset=RoadSegments.objects.all())
    intensity = SerializerMethodField()
    
    def get_intensity(self, obj):
        return get_intensity(obj.speed)
    
    class Meta:
        model = TrafficReadings
        fields = ['id', 'intensity', 'speed', 'road_segment_id']

This serializer specifies the TrafficReadings model and the fields that should be included in the output. These include a primary key related field that represents a foreign key relationship with road segments, and a serializer method field that does not correspond directly to a model field (it is used to calculate and represent the intensity of traffic based on the speed field).

# 2 - CREATE TRAFFIC READINGS
class CreateTrafficReadingSerializer(ModelSerializer):
    speed = FloatField()
    road_segment_id = PrimaryKeyRelatedField(queryset=RoadSegments.objects.all())
    
    class Meta:
        model = TrafficReadings
        fields = ['id', 'speed', 'road_segment_id']
    
    def create(self, validated_data):
        # Removes the ‘speed’ and ‘road_segment_id’ fields from the validated data
        speed = validated_data.pop('speed')
        road_segment_id = validated_data.pop('road_segment_id')
        
        # Create a new instance just with these speed and road_segment_id
        new_traffic_reading = TrafficReadings.objects.create(speed=speed, road_segment_id=road_segment_id)
        return new_traffic_reading

This serializer specifies the TrafficReadings model and the fields that should be included in the output (including a primary key related field that represents a foreign key relationship with road segments).

It also overrides the create method so that the intensity property is not involved when creating a new instance.

# 3 - GET ROAD SEGMENTS
class RoadSegmentsSerializer(ModelSerializer):
    traffic_readings = TrafficReadingsSerializer(many=True, read_only=True)
    traffic_readings_count = SerializerMethodField()
    
    def get_traffic_readings_count(self, obj):
        count = obj.trafficreadings_set.count()
        return count
    
    class Meta:
        model = RoadSegments
        fields = ['id', 'long_start', 'lat_start', 'long_end', 'lat_end', 'length', 'traffic_readings_count', 'traffic_readings']

This serializer specifies the RoadSegments model and the fields that should be included in the output. These include a nested serializer for the related TrafficReadings instances, and a serializer method field to count all traffic readings associated with the instance of road segment.

# 4 - CREATE ROAD SEGMENTS
class CreateRoadSegmentSerializer(ModelSerializer):
    class Meta:
        model = RoadSegments
        fields = ['id', 'long_start', 'lat_start', 'long_end', 'lat_end', 'length']

This serializer only specifies the RoadSegments model and the fields that should be included in the output.

# 5 - GET SENSORS
class SensorsSerializer(ModelSerializer):
    class Meta:
        model = Sensors
        fields = ['id', 'name', 'uuid']

This serializer only specifies the Sensors model and the fields that should be included in the output.

# 6 - GET SENSOR READINGS
class SensorReadingsSerializer(ModelSerializer):
    car_license_plate = PrimaryKeyRelatedField(queryset=Cars.objects.all())
    road_segment_id = PrimaryKeyRelatedField(queryset=RoadSegments.objects.all())
    sensor_uuid = PrimaryKeyRelatedField(queryset=Sensors.objects.all())
    
    class Meta:
        model = SensorReadings
        fields = ['id', 'road_segment_id', 'car_license_plate', 'timestamp', 'sensor_uuid']

This serializer specifies the SensorReadings model and the fields that should be included in the output, including primary key related fields that represent foreign key relationship with road segments, cars and sensors.

# 7 - CREATE SENSOR READINGS
class CreateSensorReadingSerializer(ModelSerializer):
    car_license_plate = CharField(max_length=6)
    timestamp = DateTimeField()
    
    class Meta:
        model = SensorReadings
        fields = ['car_license_plate', 'timestamp', 'road_segment_id', 'sensor_uuid']
    
    def create(self, validated_data):
        car_license_plate = validated_data.get('car_license_plate')
        timestamp = validated_data.get('timestamp')
        
        cars, created = Cars.objects.get_or_create(car_license_plate=car_license_plate, defaults={'created_at': timestamp})
        
        # Set timestamp to the current time
        validated_data['timestamp'] = timezone.now()
        validated_data['car_license_plate'] = cars
        
        sensor_reading = SensorReadings.objects.create(**validated_data)
        return sensor_reading

This serializer specifies the SensorReadings model and the fields that should be included in the output.

It also overrides the create method to check if the license plate detected by the sensor reading already exists in the Cars model. If it is a new car, it is added to the model.

# 8 - GET CARS
class CarsSerializer(ModelSerializer):
    sensor_readings = SensorReadingsSerializer(many=True, read_only=True)
    road_segments = RoadSegmentsSerializer(many=True, read_only=True)
    sensors = SensorsSerializer(many=True, read_only=True)
    
    class Meta:
        model = Cars
        fields = ['id', 'car_license_plate', 'created_at', 'sensor_readings', 'road_segments', 'sensors']

This serializer specifies the Cars model and the fields that should be included in the output, including nested serializers for the related instances of sensor readings, road segments and sensors.

Views

Imports necessary for the traffic_api/views.py file:

from rest_framework import status
from rest_framework.generics import ListAPIView, RetrieveUpdateDestroyAPIView, CreateAPIView, RetrieveAPIView
from rest_framework.response import Response
from django.utils import timezone
from django.db.models import Q, OuterRef, Subquery, F
from datetime import timedelta
from traffic_monitoring.settings import LOW_SPEED_THRESHOLD, HIGH_SPEED_THRESHOLD
from .permissions import IsAdminOrReadOnly, IsAnonymousReadOnly, HasAPIKey
from .models import TrafficReadings, RoadSegments, Sensors, SensorReadings, Cars
from .serializers import TrafficReadingsSerializer, CreateTrafficReadingSerializer, RoadSegmentsSerializer, CreateRoadSegmentSerializer, SensorsSerializer, SensorReadingsSerializer, CreateSensorReadingSerializer, CarsSerializer

This file has a total of 19 views. Starting with traffic readings:

# 1 - ALL TRAFFIC READINGS
class TrafficReadingsView(ListAPIView):
    queryset = TrafficReadings.objects.all()
    serializer_class = TrafficReadingsSerializer

This read-only view queries a list of all TrafficReadings instances, and specifies that the data should be serialized by TrafficReadingsSerializer.

# 2 - UPDATE OR DELETE INDIVIDUAL TRAFFIC READINGS (only for admin use)
class TrafficReadingsUpdateView(RetrieveUpdateDestroyAPIView):
    def get_permissions(self):
        if self.request.user.is_staff:
            return [IsAdminOrReadOnly()]
        return [IsAnonymousReadOnly()]
    
    queryset = TrafficReadings.objects.all()
    serializer_class = TrafficReadingsSerializer

This view has a function to set permissions because it is capable of updating and deleting instances. It also queries a list of all TrafficReadings instances, and specifies that the data should be serialized by TrafficReadingsSerializer.

The permissions functions are set in traffic_api/permissions.py as follows:

from rest_framework import permissions

# Class to allow admin users to perform any action, while it allows all other users to perform read-only actions
class IsAdminOrReadOnly(permissions.BasePermission):
    def has_permission(self, request, view):
        # Allow read-only access to all users
        if request.method in permissions.SAFE_METHODS:
            return True

        # Allow admin users to perform any action
        return request.user and request.user.is_staff

# Class to allow read-only actions to all users, including anonymous users
class IsAnonymousReadOnly(permissions.BasePermission):
    def has_permission(self, request, view):
        # Allow read-only access to all users
        if request.method in permissions.SAFE_METHODS:
            return True
        return False

Basically, these functions determine if the user is anonymous, registered or the admin, and sets read-only permissions to anyone who is not admin.

# 3 - HIGH INTENSITY TRAFFIC READINGS
class HighIntensityTrafficReadingsView(ListAPIView):
    queryset = TrafficReadings.objects.filter(speed__lte = LOW_SPEED_THRESHOLD)
    serializer_class = TrafficReadingsSerializer

This read-only view queries a list of all TrafficReadings instances that have speed values lower than or equal to the low speed threshold, and specifies that the data should be serialized by TrafficReadingsSerializer.

The same logic applies to views #4 and #5, which are for medium and low intensity traffic readings, respectively.

# 6 - CREATE TRAFFIC READING (only for admin use)
class CreateTrafficReadingView(CreateAPIView):
    queryset = TrafficReadings.objects.all()
    serializer_class = CreateTrafficReadingSerializer
    
    def get_permissions(self):
        if self.request.user.is_staff:
            return [IsAdminOrReadOnly()]
        return [IsAnonymousReadOnly()]

This view has a function to set permissions because it is capable of creating instances. It also queries a list of all TrafficReadings instances, and specifies that the data should be serialized by CreateTrafficReadingSerializer.

# 7 - ALL ROAD SEGMENTS
class RoadSegmentsView(ListAPIView):
    queryset = RoadSegments.objects.all()
    serializer_class = RoadSegmentsSerializer

This read-only view queries a list of all RoadSegments instances, and specifies that the data should be serialized by RoadSegmentsSerializer.

# 8 - UPDATE OR DELETE INDIVIDUAL ROAD SEGMENTS (only for admin use)
class RoadSegmentsUpdateView(RetrieveUpdateDestroyAPIView):
    queryset = RoadSegments.objects.all()
    serializer_class = RoadSegmentsSerializer
    
    def get(self, request, *args, **kwargs):
        instance = self.get_object()
        serializer = self.get_serializer(instance)
        traffic_readings = instance.trafficreadings_set.all()
        traffic_readings_serializer = TrafficReadingsSerializer(traffic_readings, many=True)
        
        return Response({'road_segment': serializer.data, 'traffic_readings': traffic_readings_serializer.data})
    
    def get_permissions(self):
        if self.request.user.is_staff:
            return [IsAdminOrReadOnly()]
        return [IsAnonymousReadOnly()]

This view has a function to set permissions because it is capable of updating and deleting instances. It also queries a list of all RoadSegments instances, and specifies that the data should be serialized by RoadSegmentsSerializer.

It also overrides the get method so that this view displays not only the road segments, but also the traffic readings associated with each road segment.

# 9 - HIGH INTENSITY ROAD SEGMENTS
class HighIntensityRoadSegmentsView(ListAPIView):
    def get_queryset(self):
        # Get the latest TrafficReading for each RoadSegment
        latest_readings = TrafficReadings.objects.filter(id=OuterRef('pk')).order_by('-id')
        
        # Get all road segments where the latest traffic reading has high intensity
        high_intensity_road_segments = RoadSegments.objects.annotate(latest_reading_id=Subquery(latest_readings.values('id')[:1])
        ).filter(Q(trafficreadings__id=F('latest_reading_id')) & Q(trafficreadings__speed__lte=LOW_SPEED_THRESHOLD))
        
        return high_intensity_road_segments

    def list(self, request, *args, **kwargs):
        queryset = self.get_queryset()
        
        # Serialize the road segments
        serializer = RoadSegmentsSerializer(queryset, many=True)
        return Response(serializer.data)

This view overrides the get_queryset method to be able to apply a subquery. It uses two queries to get only road segments where the last traffic reading was recorded with high intensity.

The same logic applies to views #10 and #11, which are for medium and low intensity road segments, respectively.

# 12 - CREATE ROAD SEGMENT (only for admin use)
class CreateRoadSegmentView(CreateAPIView):
    queryset = RoadSegments.objects.all()
    serializer_class = CreateRoadSegmentSerializer
    
    def get_permissions(self):
        if self.request.user.is_staff:
            return [IsAdminOrReadOnly()]
        return [IsAnonymousReadOnly()]

This view has a function to set permissions because it is capable of creating instances. It also queries a list of all RoadSegments instances, and specifies that the data should be serialized by CreateRoadSegmentSerializer.

# 13 - ALL SENSORS
class SensorsView(ListAPIView):
    queryset = Sensors.objects.all()
    serializer_class = SensorsSerializer

This read-only view queries a list of all Sensors instances, and specifies that the data should be serialized by SensorsSerializer.

# 14 - UPDATE OR DELETE INDIVIDUAL SENSORS (only for admin use)
class SensorsUpdateView(RetrieveUpdateDestroyAPIView):
    def get_permissions(self):
        if self.request.user.is_staff:
            return [IsAdminOrReadOnly()]
        return [IsAnonymousReadOnly()]
    
    queryset = Sensors.objects.all()
    serializer_class = SensorsSerializer

This view has a function to set permissions because it is capable of updating and deleting instances. It also queries a list of all Sensors instances, and specifies that the data should be serialized by SensorsSerializer.

# 15 - ALL SENSOR READINGS
class SensorReadingsView(ListAPIView):
    queryset = SensorReadings.objects.all()
    serializer_class = SensorReadingsSerializer

This read-only view queries a list of all SensorReadings instances, and specifies that the data should be serialized by SensorReadingsSerializer.

# 16 - UPDATE OR DELETE INDIVIDUAL SENSOR READINGS (only for admin use)
class SensorReadingsUpdateView(RetrieveUpdateDestroyAPIView):
    def get_permissions(self):
        if self.request.user.is_staff:
            return [IsAdminOrReadOnly()]
        return [IsAnonymousReadOnly()]
    
    queryset = SensorReadings.objects.all()
    serializer_class = SensorReadingsSerializer

This view has a function to set permissions because it is capable of updating and deleting instances. It also queries a list of all SensorReadings instances, and specifies that the data should be serialized by SensorReadingsSerializer.

# 17 - CREATE NEW SENSOR READING (for admin use only)
class CreateSensorReadingView(CreateAPIView):
    query_set = SensorReadings.objects.all()
    serializer_class = CreateSensorReadingSerializer
    permission_classes = [IsAdminOrReadOnly]

This view has a function to set permissions because it is capable of creating instances. It also queries a list of all SensorReadings instances, and specifies that the data should be serialized by CreateSensorReadingSerializer.

# 18 - ALL CARS
class CarsView(ListAPIView):
    queryset = Cars.objects.all()
    serializer_class = CarsSerializer

This read-only view queries a list of all Cars instances, and specifies that the data should be serialized by CarsSerializer.

# 19 - INDIVIDUAL CARS
class CarDetailsView(RetrieveAPIView):
    queryset = Cars.objects.all()
    serializer_class = CarsSerializer
    
    def get(self, request, car_license_plate, *args, **kwargs):
        license_plate = car_license_plate
        cars = Cars.objects.filter(Q(car_license_plate__icontains=license_plate))
        serializer = CarsSerializer(cars, many=True)
        
        car_instance = cars.first()
        
        # Filter sensor readings for the last 24 hours
        cutoff_time = timezone.now() - timedelta(hours=24)
        sensor_readings = car_instance.sensorreadings_set.filter(timestamp__gte=cutoff_time)
        sensor_readings_serializer = SensorReadingsSerializer(sensor_readings, many=True)

        # Extract road segment and sensor IDs from the filtered sensor readings
        road_segment_ids = sensor_readings.values_list('road_segment_id', flat=True).distinct()
        sensor_ids = sensor_readings.values_list('sensor_uuid', flat=True).distinct()

        # Retrieve related road segments and sensors for the last 24 hours
        road_segments_data = RoadSegments.objects.filter(id__in=road_segment_ids)
        road_segments_serializer = RoadSegmentsSerializer(road_segments_data, many=True)

        sensors_data = Sensors.objects.filter(id__in=sensor_ids)
        sensors_serializer = SensorsSerializer(sensors_data, many=True)
        
        return Response({
            'car': serializer.data,
            'sensor_readings': sensor_readings_serializer.data,
            'sensors': sensors_serializer.data,
            'road_segments': road_segments_serializer.data,
        })

This object read-only view queries all Cars instances, and specifies that the data should be serialized by CarsSerializer.

It displays the car’s data and also overrides the get method to get all sensor readings in the last 24 hours associated with the car instance. For those sensor readings, it displays the information about the road segment and sensor.

URL’s

Imports necessary for the traffic_api/urls.py file:

from django.urls import path
from .views import TrafficReadingsView, TrafficReadingsUpdateView, CreateTrafficReadingView, HighIntensityTrafficReadingsView, MediumIntensityTrafficReadingsView, LowIntensityTrafficReadingsView, RoadSegmentsView, RoadSegmentsUpdateView, CreateRoadSegmentView, HighIntensityRoadSegmentsView, MediumIntensityRoadSegmentsView, LowIntensityRoadSegmentsView

This file has a total of 20 URL’s:

urlpatterns = [
    # Home page
    path('', TrafficReadingsView.as_view(), name='home'),
    
    # Traffic Readings
    path('traffic-readings/', TrafficReadingsView.as_view(), name='all-traffic-readings'),
    path('traffic-readings/<int:pk>/', TrafficReadingsUpdateView.as_view(), name='individual-traffic-reading'),
    path('traffic-readings/high-intensity/', HighIntensityTrafficReadingsView.as_view(), name='high-intensity-traffic-readings'),
    path('traffic-readings/medium-intensity/', MediumIntensityTrafficReadingsView.as_view(), name='medium-intensity-traffic-readings'),
    path('traffic-readings/low-intensity/', LowIntensityTrafficReadingsView.as_view(), name='low-intensity-traffic-readings'),
    path('create-traffic-reading/', CreateTrafficReadingView.as_view(), name='create-traffic-reading'),
    
    # Road Segments
    path('road-segments/', RoadSegmentsView.as_view(), name='all-road-segments'),
    path('road-segments/<int:pk>/', RoadSegmentsUpdateView.as_view(), name='individual-road-segment'),
    path('road-segments/high-intensity/', HighIntensityRoadSegmentsView.as_view(), name='high-intensity-road-segments'),
    path('road-segments/medium-intensity/', MediumIntensityRoadSegmentsView.as_view(), name='medium-intensity-road-segments'),
    path('road-segments/low-intensity/', LowIntensityRoadSegmentsView.as_view(), name='low-intensity-road-segments'),
    path('create-road-segment/', CreateRoadSegmentView.as_view(), name='create-road-segment'),

        # Sensors
    path('sensors/', SensorsView.as_view(), name='all-sensors'),
    path('sensors/<int:pk>/', SensorsUpdateView.as_view(), name='individual-sensor'),
    path('sensors-readings/', SensorReadingsView.as_view(), name='sensors-readings'),
    path('sensors-readings/<int:pk>/', SensorReadingsUpdateView.as_view(), name='individual-sensor-readings'),
    path('create-sensor-reading/', CreateSensorReadingView.as_view(), name='create-sensor-reading'),
    
    # Cars
    path('cars/', CarsView.as_view(), name='all-cars'),
    path('cars/<str:car_license_plate>/', CarDetailsView.as_view(), name='individual-car'),
]

Here is a table describing each endpoint:

Testing the API

To test user permissions, in the traffic_api/test.py I created 8 tests:

• For anonymous users:

  1. Can read;
  2. Can not create;
  3. Can not update;
  4. Can not delete.

• For the admin user:

  1. Can read;
  2. Can create;
  3. Can update;
  4. Can delete.

Here is an example to test anonymous permissions: it tries to create a new road segment and then checks if the status code from the HTTP response is “forbidden”.

# Test 2 - Can an anonymous user create data?
class TestAnonymousUserCreate(APITestCase):
    def test_anonymous_user_create_road_segment(self):
        url = reverse('create-road-segment')
        road_segment_data = {"long_start": 100.123, "lat_start": 30.456, "long_end": 105.789, "lat_end": 60.123, "speed": 35.35, "length": 72.55}
        response = self.client.post(url, road_segment_data, format="json")

        # Assert that anonymous users can not create road segments (status code 403 Forbidden)
        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

Here is an example to test admin permission: it authenticates as admin, tries to create a new road segment and then checks if the status code from the HTTP response is “created”.

# Test 6 - Can the admin user create the data?
class TestAdminUserCreate(APITestCase):
    def test_admin_user_create_road_segment(self):
        # Authenticate as the admin
        self.admin_user = User.objects.create_user(username="test_admin_user", password="test_admin_password")
        self.admin_user.is_staff = True   # Assign admin role
        self.admin_user.save()
        self.client.force_authenticate(user=self.admin_user)
        
        # Create a row of data
        url = reverse('create-road-segment')
        road_segment_data = {"long_start": 100.123, "lat_start": 30.456, "long_end": 105.789, "lat_end": 60.123, "speed": 35.35, "length": 72.55}
        response = self.client.post(url, road_segment_data, format="json")

        # Assert that the admin user can create road segments (status code 201 created)
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

The same logic applies for all other tests.

Project Improvements

There are a couple of features that need some improvement. I will write here the ones I am aware of:

• Optimisation for scaling - the methods used to get some properties are not suited for efficient use when considering large amounts of data.
• Tokenize sensor readings - I tried to implement a permission so that only POST requests with a certain token would be able to create new sensor readings, but it was not working with my tests.
• Bulk sensor readings - each sensor should accumulate a records and then send these records in bulk to the platform, but I only implemented single sensor reading creations.

Links to other platforms

LinkdIn; GitHub; Medium; Goodreads.