Food Delivery (Swiggy)
Order lifecycle via state machine, delivery agent assignment via strategy pattern, and real-time tracking via observer. A three-sided marketplace where coordination timing is everything.
Key Abstractions
Orchestrator that coordinates restaurants, customers, agents, and order lifecycle
Menu provider that confirms or rejects incoming orders
Orderer who places orders and receives delivery status updates
Driver with location and current load who picks up and delivers orders
Core entity with state machine: PLACED -> CONFIRMED -> PREPARING -> OUT_FOR_DELIVERY -> DELIVERED
Individual food item with name, price, and availability flag
Algorithm for picking a delivery agent: nearest, load-balanced, or rating-based
Class Diagram
The Key Insight
Food delivery looks like ride-sharing with an extra participant. It is not. The critical difference is the three-sided coordination problem. In ride-sharing, you match two parties and go. In food delivery, you have a customer, a restaurant, and a delivery agent, and they all operate on different timelines. The customer wants speed. The restaurant needs preparation time. The agent needs to be assigned at the right moment, not too early and not too late.
If you assign the agent when the order is placed, that agent sits idle for 20 minutes while the kitchen works. If you assign too late, the food gets cold waiting for pickup. The state machine is what makes this coordination possible. Each transition represents a real-world handoff between participants, and the system enforces that handoffs happen in order.
Requirements
Functional
- Customer places an order from a restaurant's menu with item quantities
- Restaurant confirms or rejects the order
- Order progresses through states: placed, confirmed, preparing, out for delivery, delivered
- Delivery agent assigned when food is ready for pickup
- Agent assignment uses a configurable strategy (nearest, load-balanced)
- Customer and restaurant receive status updates at each transition
- Orders can be cancelled before dispatch
Non-Functional
- Thread-safe agent assignment to prevent double-booking
- Assignment strategy swappable at runtime for peak vs off-peak hours
- State machine enforces valid transitions only
- Agent capacity tracked to prevent overloading a single agent
Design Decisions
Why does the order go through CONFIRMED before PREPARING?
This is the restaurant acceptance gate. Without it, you charge the customer and then discover the restaurant is closed or out of a key ingredient. The CONFIRMED state means the restaurant has explicitly accepted the order. If they reject it, the order goes to CANCELLED with a refund. Skipping this step is how you get angry customers who paid for food that never gets made.
Why is agent assignment a strategy, not hardcoded logic?
During lunch rush, you want load balancing so no single agent gets buried with five orders while others sit idle. During off-peak hours, nearest-agent minimizes delivery time. Different cities might have different traffic patterns requiring different algorithms entirely. Strategy lets you configure this per-market without touching order logic.
Why track agent load instead of just available/unavailable?
A binary available flag means an agent can only handle one order at a time. In reality, agents batch deliveries. An agent heading to Koramangala can pick up two orders from nearby restaurants heading the same direction. The load counter with a max allows this. The LoadBalancedStrategy then spreads orders evenly across agents who still have capacity.
Why is Order the state machine owner, not DeliveryPlatform?
The platform orchestrates but should not own transition logic. If the platform validates transitions, you need to duplicate that validation everywhere an order status could change. Putting the state machine inside Order means the rules follow the entity. No matter who calls confirm() or dispatch(), the same transition rules apply.
Interview Follow-ups
- "How would you handle order modifications after placement?" Allow modifications only in PLACED state. Once CONFIRMED, the restaurant has started planning. Add a
modifyItems()method on Order that checks the current status before allowing changes. - "How would you implement real-time tracking?" The observer pattern is already in place. Add a
LocationUpdateObserverthat receives agent GPS coordinates. The customer-facing app subscribes to updates for their specific order. - "How would you handle restaurant prep time estimates?" Add a
prepTimeMinutesfield to the CONFIRMED state. The dispatch timer waits untilconfirmTime + prepTimebefore calling the assignment strategy. This prevents early agent assignment. - "How would you add ratings and reviews?" After DELIVERED, prompt the customer to rate both the restaurant and the agent. Store ratings as a rolling average. The assignment strategy can factor in agent ratings for premium customers.
Code Implementation
from __future__ import annotations
from abc import ABC, abstractmethod
from enum import Enum
from dataclasses import dataclass, field
from typing import Protocol
from threading import Lock
import uuid
import math
class OrderStatus(Enum):
PLACED = "placed"
CONFIRMED = "confirmed"
PREPARING = "preparing"
OUT_FOR_DELIVERY = "out_for_delivery"
DELIVERED = "delivered"
CANCELLED = "cancelled"
VALID_TRANSITIONS: dict[OrderStatus, set[OrderStatus]] = {
OrderStatus.PLACED: {OrderStatus.CONFIRMED, OrderStatus.CANCELLED},
OrderStatus.CONFIRMED: {OrderStatus.PREPARING, OrderStatus.CANCELLED},
OrderStatus.PREPARING: {OrderStatus.OUT_FOR_DELIVERY, OrderStatus.CANCELLED},
OrderStatus.OUT_FOR_DELIVERY: {OrderStatus.DELIVERED},
OrderStatus.DELIVERED: set(),
OrderStatus.CANCELLED: set(),
}
@dataclass
class Location:
lat: float
lng: float
def distance_to(self, other: "Location") -> float:
"""Haversine distance in km."""
R = 6371
lat1, lat2 = math.radians(self.lat), math.radians(other.lat)
dlat = math.radians(other.lat - self.lat)
dlng = math.radians(other.lng - self.lng)
a = (math.sin(dlat / 2) ** 2
+ math.cos(lat1) * math.cos(lat2) * math.sin(dlng / 2) ** 2)
return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
@dataclass
class MenuItem:
name: str
price: float
available: bool = True
@dataclass
class OrderItem:
menu_item: MenuItem
quantity: int
@property
def subtotal(self) -> float:
return self.menu_item.price * self.quantity
@dataclass
class Customer:
id: str
name: str
address: Location
class Restaurant:
def __init__(self, restaurant_id: str, name: str, location: Location,
menu: list[MenuItem]):
self.id = restaurant_id
self.name = name
self.location = location
self._menu = {item.name: item for item in menu}
def get_item(self, name: str) -> MenuItem | None:
return self._menu.get(name)
def get_available_items(self) -> list[MenuItem]:
return [item for item in self._menu.values() if item.available]
class DeliveryAgent:
def __init__(self, agent_id: str, name: str, location: Location,
max_load: int = 3):
self.id = agent_id
self.name = name
self.location = location
self.current_load = 0
self.max_load = max_load
self.rating = 4.5
def is_available(self) -> bool:
return self.current_load < self.max_load
def assign_order(self) -> None:
if not self.is_available():
raise RuntimeError(f"Agent {self.name} is at max capacity")
self.current_load += 1
def complete_order(self) -> None:
self.current_load = max(0, self.current_load - 1)
class OrderObserver(ABC):
@abstractmethod
def on_status_change(self, order_id: str, customer_name: str,
restaurant_name: str, agent_name: str | None,
old_status: OrderStatus,
new_status: OrderStatus) -> None: ...
class ConsoleNotifier(OrderObserver):
def on_status_change(self, order_id: str, customer_name: str,
restaurant_name: str, agent_name: str | None,
old_status: OrderStatus,
new_status: OrderStatus) -> None:
agent_info = f" | agent: {agent_name}" if agent_name else ""
print(f" [ORDER {order_id}] {old_status.value} -> {new_status.value} | "
f"{customer_name} from {restaurant_name}{agent_info}")
class Order:
def __init__(self, order_id: str, customer: Customer,
restaurant: Restaurant, items: list[OrderItem]):
self.id = order_id
self.customer = customer
self.restaurant = restaurant
self.items = items
self.agent: DeliveryAgent | None = None
self._status = OrderStatus.PLACED
self.total = sum(item.subtotal for item in items)
self._observers: list[OrderObserver] = []
@property
def status(self) -> OrderStatus:
return self._status
def add_observer(self, observer: OrderObserver) -> None:
self._observers.append(observer)
def _transition(self, new_status: OrderStatus) -> None:
if new_status not in VALID_TRANSITIONS[self._status]:
raise ValueError(
f"Cannot transition from {self._status.value} to "
f"{new_status.value}")
old = self._status
self._status = new_status
agent_name = self.agent.name if self.agent else None
for obs in self._observers:
obs.on_status_change(
self.id, self.customer.name, self.restaurant.name,
agent_name, old, new_status)
def confirm(self) -> None:
self._transition(OrderStatus.CONFIRMED)
def start_preparing(self) -> None:
self._transition(OrderStatus.PREPARING)
def dispatch(self, agent: DeliveryAgent) -> None:
agent.assign_order()
self.agent = agent
self._transition(OrderStatus.OUT_FOR_DELIVERY)
def deliver(self) -> None:
self._transition(OrderStatus.DELIVERED)
if self.agent:
self.agent.complete_order()
def cancel(self) -> None:
self._transition(OrderStatus.CANCELLED)
if self.agent:
self.agent.complete_order()
class AssignmentStrategy(Protocol):
def assign_agent(self, agents: list[DeliveryAgent],
restaurant: Restaurant) -> DeliveryAgent | None: ...
class NearestAgentStrategy:
"""Pick the closest available agent to the restaurant."""
def assign_agent(self, agents: list[DeliveryAgent],
restaurant: Restaurant) -> DeliveryAgent | None:
available = [a for a in agents if a.is_available()]
if not available:
return None
return min(available,
key=lambda a: a.location.distance_to(restaurant.location))
class LoadBalancedStrategy:
"""Pick the available agent with the fewest current orders."""
def assign_agent(self, agents: list[DeliveryAgent],
restaurant: Restaurant) -> DeliveryAgent | None:
available = [a for a in agents if a.is_available()]
if not available:
return None
return min(available, key=lambda a: a.current_load)
class DeliveryPlatform:
def __init__(self, assignment: AssignmentStrategy | None = None):
self._restaurants: dict[str, Restaurant] = {}
self._agents: list[DeliveryAgent] = []
self._orders: dict[str, Order] = {}
self._assignment = assignment or NearestAgentStrategy()
self._notifier = ConsoleNotifier()
self._lock = Lock()
def register_restaurant(self, restaurant: Restaurant) -> None:
self._restaurants[restaurant.id] = restaurant
def register_agent(self, agent: DeliveryAgent) -> None:
self._agents.append(agent)
def set_assignment_strategy(self, strategy: AssignmentStrategy) -> None:
self._assignment = strategy
def place_order(self, customer: Customer, restaurant_id: str,
item_requests: list[tuple[str, int]]) -> Order:
"""Place an order. item_requests is a list of (item_name, quantity)."""
with self._lock:
restaurant = self._restaurants.get(restaurant_id)
if not restaurant:
raise ValueError(f"Restaurant {restaurant_id} not found")
order_items = []
for item_name, qty in item_requests:
menu_item = restaurant.get_item(item_name)
if menu_item is None:
raise ValueError(f"Item '{item_name}' not on menu")
if not menu_item.available:
raise ValueError(f"Item '{item_name}' is unavailable")
order_items.append(OrderItem(menu_item, qty))
order_id = str(uuid.uuid4())[:8]
order = Order(order_id, customer, restaurant, order_items)
order.add_observer(self._notifier)
self._orders[order_id] = order
return order
def confirm_order(self, order_id: str) -> None:
"""Restaurant confirms it can fulfill the order."""
self._orders[order_id].confirm()
def mark_preparing(self, order_id: str) -> None:
self._orders[order_id].start_preparing()
def dispatch_order(self, order_id: str) -> None:
"""Assign an agent and send the order out for delivery."""
with self._lock:
order = self._orders[order_id]
agent = self._assignment.assign_agent(
self._agents, order.restaurant)
if agent is None:
raise RuntimeError("No delivery agents available")
order.dispatch(agent)
def complete_delivery(self, order_id: str) -> None:
with self._lock:
self._orders[order_id].deliver()
def cancel_order(self, order_id: str) -> None:
with self._lock:
self._orders[order_id].cancel()
if __name__ == "__main__":
platform = DeliveryPlatform(assignment=NearestAgentStrategy())
# Set up a restaurant
biryani_house = Restaurant("r1", "Biryani House", Location(12.9716, 77.5946), [
MenuItem("Chicken Biryani", 250.0),
MenuItem("Paneer Biryani", 220.0),
MenuItem("Raita", 40.0),
MenuItem("Gulab Jamun", 60.0),
])
platform.register_restaurant(biryani_house)
# Set up delivery agents
platform.register_agent(DeliveryAgent("a1", "Rahul", Location(12.9720, 77.5950)))
platform.register_agent(DeliveryAgent("a2", "Meena", Location(12.9600, 77.6100)))
platform.register_agent(DeliveryAgent("a3", "Vijay", Location(12.9800, 77.5800)))
# Customer places an order
customer = Customer("c1", "Priya", Location(12.9352, 77.6245))
print("=== Placing order ===")
order = platform.place_order(
customer, "r1",
[("Chicken Biryani", 2), ("Raita", 1)])
print(f" Order {order.id}: Rs {order.total}")
print("\n=== Restaurant confirms ===")
platform.confirm_order(order.id)
print("\n=== Kitchen starts preparing ===")
platform.mark_preparing(order.id)
print("\n=== Dispatching delivery agent ===")
platform.dispatch_order(order.id)
print(f" Assigned to: {order.agent.name}")
print("\n=== Delivered ===")
platform.complete_delivery(order.id)
# Second order with load balancing
print("\n=== Switching to load-balanced assignment ===")
platform.set_assignment_strategy(LoadBalancedStrategy())
customer2 = Customer("c2", "Amit", Location(12.9400, 77.6000))
order2 = platform.place_order(
customer2, "r1",
[("Paneer Biryani", 1), ("Gulab Jamun", 2)])
platform.confirm_order(order2.id)
platform.mark_preparing(order2.id)
platform.dispatch_order(order2.id)
print(f" Order {order2.id} assigned to: {order2.agent.name} "
f"(load: {order2.agent.current_load}/{order2.agent.max_load})")
platform.complete_delivery(order2.id)
print("\n=== Invalid transition test ===")
try:
platform.dispatch_order(order.id) # already delivered
except ValueError as e:
print(f" Correctly rejected: {e}")
print("\n=== Cancellation test ===")
order3 = platform.place_order(
customer, "r1", [("Gulab Jamun", 3)])
print(f" Placed order {order3.id}")
platform.cancel_order(order3.id)
print(f" Status after cancel: {order3.status.value}")Common Mistakes
- ✗Skipping the CONFIRMED state. Without restaurant confirmation, you charge the customer before knowing if the food can be made.
- ✗Assigning a delivery agent at order placement. The agent should be assigned when food is almost ready, not 30 minutes early.
- ✗Hardcoding agent assignment logic. Load balancing during peak hours and nearest-agent during off-peak need different strategies.
- ✗Not tracking agent load. Assigning five orders to one agent while others sit idle kills delivery times.
Key Points
- ✓State machine on Order enforces valid transitions. An order cannot jump from PLACED to DELIVERED.
- ✓Strategy pattern for agent assignment: swap between nearest-agent, load-balanced, or rating-based without changing order logic
- ✓Observer notifies customer and restaurant on every status change for real-time tracking
- ✓Restaurant confirmation is an explicit state. The order waits for the restaurant to accept before preparation begins.