Merge pull request #2 from Ke-vin-S/feature/sync-with-order

Add hosted db for production

Author: Ke-vin-S

.env.example

@@ -3,3 +3,10 @@ LOGISTICS_SERVICE_PORT=8002
 KAFKA_BROKER_URL=kafka:9092
 IS_DOCKER=True
 ALLOWED_HOSTS=localhost,logistics_service,api_gateway
+
+ENVIRONMENT=production
+DB_NAME=your_supabase_db_name
+DB_USER=your_supabase_user
+DB_PASSWORD=your_supabase_password
+DB_HOST=db.xxxxxxxxx.supabase.co
+DB_PORT=5432

kafka_client/__init__.py

@@ -0,0 +1,38 @@
+from confluent_kafka import Consumer, KafkaException
+import json
+import os
+import logging
+
+logger = logging.getLogger(__name__)
+
+KAFKA_BROKER_URL = os.getenv("KAFKA_BROKER_URL", "localhost:9092")
+
+def start_consumer(topic: str, group_id: str = "default-group", auto_offset_reset: str = "latest"):
+    consumer = Consumer({
+        'bootstrap.servers': KAFKA_BROKER_URL,
+        'group.id': group_id,
+        'auto.offset.reset': auto_offset_reset,
+    })
+
+    consumer.subscribe([topic])
+    logger.info(f"Listening to topic: {topic}")
+
+    try:
+        while True:
+            msg = consumer.poll(timeout=1.0)
+            if msg is None:
+                continue
+            if msg.error():
+                raise KafkaException(msg.error())
+
+            value = json.loads(msg.value().decode("utf-8"))
+            key = msg.key().decode("utf-8") if msg.key() else None
+            handle_message(msg.topic(), key, value)
+    except KeyboardInterrupt:
+        print("Consumer stopped.")
+    finally:
+        consumer.close()
+
+def handle_message(topic: str, key: str, value: dict):
+    if topic == "shipment.status.updated":
+        print(f"Shipment event for Order {key}: {value}")

kafka_client/consumer.py

@@ -0,0 +1,30 @@
+from confluent_kafka import Producer
+import json
+import os
+import logging
+
+logger = logging.getLogger(__name__)
+
+KAFKA_BROKER_URL = os.getenv("KAFKA_BROKER_URL", "localhost:9092")
+
+producer = Producer({
+    'bootstrap.servers': KAFKA_BROKER_URL
+})
+
+def delivery_report(err, msg):
+    if err is not None:
+        logger.error(f"Delivery failed for record {msg.key()}: {err}")
+    else:
+        logger.info(f"Record {msg.key()} successfully produced to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")
+
+def publish_event(topic: str, key: str, payload: dict):
+    try:
+        producer.produce(
+            topic=topic,
+            key=key,
+            value=json.dumps(payload),
+            callback=delivery_report
+        )
+        producer.flush()  # ensures all messages are delivered before exiting
+    except Exception as e:
+        logger.error(f"[Kafka Error] Failed to publish to topic '{topic}': {e}")

kafka_client/producer.py

@@ -87,14 +87,27 @@
 # Database
 # https://docs.djangoproject.com/en/5.2/ref/settings/#databases
 
-DATABASES = {
-    'default': {
-        'ENGINE': 'django.db.backends.sqlite3',
-        'NAME': BASE_DIR / 'db.sqlite3',
-    }
-}
-
 
+ENV = os.getenv("ENVIRONMENT", "development")
+
+if ENV == "production":
+    DATABASES = {
+        'default': {
+            'ENGINE': 'django.db.backends.postgresql',
+            'NAME': os.getenv('DB_NAME'),
+            'USER': os.getenv('DB_USER'),
+            'PASSWORD': os.getenv('DB_PASSWORD'),
+            'HOST': os.getenv('DB_HOST'),
+            'PORT': os.getenv('DB_PORT'),
+        }
+    }
+else:
+    DATABASES = {
+        'default': {
+            'ENGINE': 'django.db.backends.sqlite3',
+            'NAME': BASE_DIR / "db.sqlite3",
+        }
+    }
 # Password validation
 # https://docs.djangoproject.com/en/5.2/ref/settings/#auth-password-validators
 

logistics_core/settings.py

@@ -25,6 +25,6 @@ sqlparse==0.5.3
 tzdata==2025.2
 uritemplate==4.1.1
 
-kafka~=1.3.5
 requests~=2.32.3
-django-environ~=0.12.0
+django-environ~=0.12.0
+psycopg2-binary==2.9.10

requirements.txt

@@ -0,0 +1,37 @@
+import os
+import json
+import time
+from confluent_kafka import Consumer, KafkaException
+
+def consume_messages(topic, timeout=10, group_id="shipment-integration-tests"):
+    consumer = Consumer({
+        'bootstrap.servers': os.getenv("KAFKA_BROKER_URL", "localhost:9092"),
+        'group.id': group_id,
+        'auto.offset.reset': 'earliest',
+        'enable.auto.commit': True,
+    })
+
+    consumer.subscribe([topic])
+    end_time = time.time() + timeout
+    messages = []
+
+    try:
+        while time.time() < end_time:
+            msg = consumer.poll(timeout=1.0)
+            if msg is None:
+                continue
+            if msg.error():
+                raise KafkaException(msg.error())
+
+            key = msg.key().decode("utf-8") if msg.key() else None
+            value = json.loads(msg.value().decode("utf-8"))
+            print(f"[KAFKA] Got: {value}")
+            messages.append(msg)
+
+            # optional: stop early if messages are found
+            if messages:
+                break
+    finally:
+        consumer.close()
+
+    return messages

shipments/tests/kafka_test_consumer.py

@@ -0,0 +1,74 @@
+import json
+import unittest
+
+from django.urls import reverse
+from rest_framework.test import APITestCase
+from shipments.models import Shipment
+from shipments.tests.kafka_test_consumer import consume_messages
+
+
+class ShipmentKafkaIntegrationTests(APITestCase):
+    def setUp(self):
+        self.shipment = Shipment.objects.create(
+            shipment_id="SHIP-KAFKA-001",
+            order_id="ORD-KAFKA-001",
+            origin={"lat": 7.87, "lng": 80.77},
+            destination={"lat": 6.92, "lng": 79.86},
+            demand=5
+        )
+
+    @unittest.skip("Temporarily skipping this test")
+    def test_dispatch_and_kafka_publish(self):
+        # Schedule the shipment
+        url_schedule = reverse("shipment-mark-scheduled", kwargs={"pk": self.shipment.pk})
+        self.client.post(url_schedule, data={"scheduled_time": "2025-05-17T10:00:00Z"})
+
+        # Dispatch the shipment
+        url_dispatch = reverse("shipment-mark-dispatched", kwargs={"pk": self.shipment.pk})
+        response = self.client.post(url_dispatch, data={"dispatch_time": "2025-05-17T10:15:00Z"})
+
+        self.assertEqual(response.status_code, 200)
+
+        # Consume Kafka messages
+        messages = consume_messages("shipment.status.updated")
+
+        dispatched = []
+        for msg in messages:
+            payload = json.loads(msg.value().decode("utf-8"))
+            print("DEBUG Kafka:", payload)
+            if payload["order_id"] == self.shipment.order_id and payload["status"] == "dispatched":
+                dispatched.append(payload)
+
+        self.assertTrue(dispatched, "Dispatched Kafka message not found.")
+        self.assertEqual(dispatched[0]["shipment_id"], self.shipment.shipment_id)
+
+    @unittest.skip("Temporarily skipping this test")
+    def test_delivered_and_kafka_publish(self):
+        # Full flow: schedule → dispatch → in_transit → deliver
+        self.client.post(reverse("shipment-mark-scheduled", kwargs={"pk": self.shipment.pk}))
+        self.client.post(reverse("shipment-mark-dispatched", kwargs={"pk": self.shipment.pk}))
+        self.client.post(reverse("shipment-mark-in-transit", kwargs={"pk": self.shipment.pk}))
+        response = self.client.post(
+            reverse("shipment-mark-delivered", kwargs={"pk": self.shipment.pk}),
+            data={"delivery_time": "2025-05-17T12:00:00Z"}
+        )
+        self.assertEqual(response.status_code, 200)
+
+        # Consume Kafka messages
+        messages = consume_messages("shipment.status.updated", timeout=15)
+
+        found_dispatched = False
+        found_delivered = False
+
+        for msg in messages:
+            payload = json.loads(msg.value().decode("utf-8"))
+            print("DEBUG Kafka:", payload)
+            if payload["order_id"] != self.shipment.order_id:
+                continue
+            if payload["status"] == "dispatched":
+                found_dispatched = True
+            elif payload["status"] == "delivered":
+                found_delivered = True
+
+        self.assertTrue(found_dispatched, "Dispatched Kafka message not found")
+        self.assertTrue(found_delivered, "Delivered Kafka message not found")

shipments/tests/test_producer_with_kafka.py

@@ -0,0 +1,59 @@
+from django.test import TestCase
+from django.utils import timezone
+from shipments.models import Shipment
+from django.core.exceptions import ValidationError
+
+class ShipmentModelTests(TestCase):
+    def setUp(self):
+        self.shipment = Shipment.objects.create(
+            shipment_id="SHIP100",
+            order_id="ORD100",
+            origin={"lat": 7.8731, "lng": 80.7718},
+            destination={"lat": 6.9271, "lng": 79.8612},
+            demand=10
+        )
+
+    def test_schedule_from_pending(self):
+        self.shipment.mark_scheduled(scheduled_time=timezone.now())
+        self.assertEqual(self.shipment.status, 'scheduled')
+        self.assertIsNotNone(self.shipment.scheduled_dispatch)
+
+    def test_dispatch_from_scheduled(self):
+        self.shipment.mark_scheduled()
+        self.shipment.mark_dispatched()
+        self.assertEqual(self.shipment.status, 'dispatched')
+        self.assertIsNotNone(self.shipment.actual_dispatch)
+
+    def test_invalid_dispatch_from_pending(self):
+        with self.assertRaises(ValidationError):
+            self.shipment.mark_dispatched()
+
+    def test_in_transit_from_dispatched(self):
+        self.shipment.mark_scheduled()
+        self.shipment.mark_dispatched()
+        self.shipment.mark_in_transit()
+        self.assertEqual(self.shipment.status, 'in_transit')
+
+    def test_delivered_from_in_transit(self):
+        self.shipment.mark_scheduled()
+        self.shipment.mark_dispatched()
+        self.shipment.mark_in_transit()
+        self.shipment.mark_delivered()
+        self.assertEqual(self.shipment.status, 'delivered')
+        self.assertIsNotNone(self.shipment.delivery_time)
+
+    def test_invalid_delivered_from_pending(self):
+        with self.assertRaises(ValidationError):
+            self.shipment.mark_delivered()
+
+    def test_failed_from_active_states(self):
+        self.shipment.mark_failed()
+        self.assertEqual(self.shipment.status, 'failed')
+
+    def test_invalid_failed_from_delivered(self):
+        self.shipment.mark_scheduled()
+        self.shipment.mark_dispatched()
+        self.shipment.mark_in_transit()
+        self.shipment.mark_delivered()
+        with self.assertRaises(ValidationError):
+            self.shipment.mark_failed()

shipments/tests/test_status_update.py

@@ -1,10 +1,12 @@
+from django.utils import timezone
 from rest_framework import viewsets, status
 from rest_framework.decorators import action
 from rest_framework.response import Response
 from django.core.exceptions import ValidationError
 from django.utils.dateparse import parse_datetime
 from django.shortcuts import get_object_or_404
 
+from kafka_client.producer import publish_event
 from .models import Shipment
 from .serializers import ShipmentSerializer
 
@@ -48,7 +50,20 @@ def mark_scheduled(self, request, pk=None):
     @action(detail=True, methods=['post'])
     def mark_dispatched(self, request, pk=None):
         shipment = get_object_or_404(Shipment, pk=pk)
-        return self.handle_transition(request, shipment, shipment.mark_dispatched, time_field='dispatch_time')
+        response = self.handle_transition(request, shipment, shipment.mark_dispatched, time_field='dispatch_time')
+
+        if response.status_code == 200:
+            publish_event(
+                topic="shipment.status.updated",
+                key=str(shipment.order_id),
+                payload={
+                    "shipment_id": shipment.shipment_id,
+                    "order_id": shipment.order_id,
+                    "status": "dispatched",
+                    "timestamp": timezone.now().isoformat()
+                }
+            )
+        return response
 
     @action(detail=True, methods=['post'])
     def mark_in_transit(self, request, pk=None):
@@ -58,7 +73,20 @@ def mark_in_transit(self, request, pk=None):
     @action(detail=True, methods=['post'])
     def mark_delivered(self, request, pk=None):
         shipment = get_object_or_404(Shipment, pk=pk)
-        return self.handle_transition(request, shipment, shipment.mark_delivered, time_field='delivery_time')
+        response = self.handle_transition(request, shipment, shipment.mark_delivered, time_field='delivery_time')
+
+        if response.status_code == 200:
+            publish_event(
+                topic="shipment.status.updated",
+                key=str(shipment.order_id),
+                payload={
+                    "shipment_id": shipment.shipment_id,
+                    "order_id": shipment.order_id,
+                    "status": "delivered",
+                    "timestamp": timezone.now().isoformat()
+                }
+            )
+        return response
 
     @action(detail=True, methods=['post'])
     def mark_failed(self, request, pk=None):