Problem Statement
Design the core backend of a ride-sharing app like Uber or Lyft. A rider requests a trip with a pickup location and destination. The system matches them with a nearby available driver, calculates the fare, tracks the trip state machine (REQUESTED → DRIVER_ASSIGNED → IN_PROGRESS → COMPLETED), and handles cancellation with appropriate fees.
Core Entities
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional
import math
import uuid
class TripStatus(Enum):
REQUESTED = "requested"
DRIVER_ASSIGNED = "driver_assigned"
DRIVER_ARRIVED = "driver_arrived"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
CANCELLED = "cancelled"
class VehicleClass(Enum):
ECONOMY = "economy"
COMFORT = "comfort"
PREMIUM = "premium"
@dataclass
class Location:
latitude: float
longitude: float
def distance_km(self, other: 'Location') -> float:
"""Haversine formula for great-circle distance."""
R = 6371 # Earth radius in km
lat1, lon1 = math.radians(self.latitude), math.radians(self.longitude)
lat2, lon2 = math.radians(other.latitude), math.radians(other.longitude)
dlat, dlon = lat2 - lat1, lon2 - lon1
a = math.sin(dlat/2)**2 + math.cos(lat1)*math.cos(lat2)*math.sin(dlon/2)**2
return R * 2 * math.asin(math.sqrt(a))
@dataclass
class Driver:
driver_id: str
name: str
vehicle_class: VehicleClass
license_plate: str
rating: float = 5.0
is_available: bool = True
current_location: Optional[Location] = None
@dataclass
class Rider:
rider_id: str
name: str
email: str
rating: float = 5.0
@dataclass
class Trip:
trip_id: str = field(default_factory=lambda: str(uuid.uuid4()))
rider: Optional['Rider'] = None
driver: Optional[Driver] = None
pickup: Optional[Location] = None
destination: Optional[Location] = None
vehicle_class: VehicleClass = VehicleClass.ECONOMY
status: TripStatus = TripStatus.REQUESTED
fare: float = 0.0
start_time: Optional[datetime] = None
end_time: Optional[datetime] = None
created_at: datetime = field(default_factory=datetime.now)
Fare Calculation
@dataclass
class FareConfig:
base_fare: float # Fixed charge per trip
per_km_rate: float # Per kilometer
per_min_rate: float # Per minute of trip time
minimum_fare: float
surge_multiplier: float = 1.0
FARE_CONFIGS = {
VehicleClass.ECONOMY: FareConfig(
base_fare=2.00, per_km_rate=0.80,
per_min_rate=0.15, minimum_fare=5.00
),
VehicleClass.COMFORT: FareConfig(
base_fare=3.50, per_km_rate=1.20,
per_min_rate=0.22, minimum_fare=8.00
),
VehicleClass.PREMIUM: FareConfig(
base_fare=5.00, per_km_rate=2.00,
per_min_rate=0.35, minimum_fare=15.00
),
}
class FareCalculator:
def estimate(self, pickup: Location, destination: Location,
vehicle_class: VehicleClass,
surge_multiplier: float = 1.0) -> float:
config = FARE_CONFIGS[vehicle_class]
dist_km = pickup.distance_km(destination)
# Estimate time: assume average 30 km/h in city
est_minutes = (dist_km / 30) * 60
fare = (config.base_fare
+ dist_km * config.per_km_rate
+ est_minutes * config.per_min_rate)
fare = max(fare, config.minimum_fare)
return round(fare * surge_multiplier, 2)
def calculate_final(self, trip: Trip) -> float:
if not trip.start_time or not trip.end_time:
return self.estimate(trip.pickup, trip.destination,
trip.vehicle_class)
config = FARE_CONFIGS[trip.vehicle_class]
dist_km = trip.pickup.distance_km(trip.destination)
minutes = (trip.end_time - trip.start_time).total_seconds() / 60
fare = (config.base_fare
+ dist_km * config.per_km_rate
+ minutes * config.per_min_rate)
fare = max(fare, config.minimum_fare)
return round(fare * config.surge_multiplier, 2)
Driver Matching
import heapq
class DriverMatcher:
def __init__(self):
self._drivers: dict[str, Driver] = {}
def register_driver(self, driver: Driver):
self._drivers[driver.driver_id] = driver
def update_location(self, driver_id: str, location: Location):
if driver_id in self._drivers:
self._drivers[driver_id].current_location = location
def find_nearest_drivers(self, pickup: Location,
vehicle_class: VehicleClass,
radius_km: float = 5.0,
top_n: int = 5) -> list[Driver]:
"""
Find top_n nearest available drivers within radius_km.
Uses a min-heap to efficiently find nearest without sorting all drivers.
"""
candidates = []
for driver in self._drivers.values():
if (not driver.is_available
or driver.vehicle_class != vehicle_class
or driver.current_location is None):
continue
dist = pickup.distance_km(driver.current_location)
if dist <= radius_km:
heapq.heappush(candidates, (dist, driver.driver_id, driver))
# Return top_n nearest
result = []
while candidates and len(result) < top_n:
dist, _, driver = heapq.heappop(candidates)
result.append(driver)
return result
Trip State Machine
class RideSharingService:
def __init__(self):
self._trips: dict[str, Trip] = {}
self._riders: dict[str, Rider] = {}
self.matcher = DriverMatcher()
self.fare_calc = FareCalculator()
# Valid state transitions
self._transitions = {
TripStatus.REQUESTED: [TripStatus.DRIVER_ASSIGNED, TripStatus.CANCELLED],
TripStatus.DRIVER_ASSIGNED: [TripStatus.DRIVER_ARRIVED, TripStatus.CANCELLED],
TripStatus.DRIVER_ARRIVED: [TripStatus.IN_PROGRESS, TripStatus.CANCELLED],
TripStatus.IN_PROGRESS: [TripStatus.COMPLETED],
TripStatus.COMPLETED: [],
TripStatus.CANCELLED: [],
}
def request_trip(self, rider_id: str, pickup: Location,
destination: Location,
vehicle_class: VehicleClass = VehicleClass.ECONOMY) -> Trip:
rider = self._riders.get(rider_id)
if not rider:
raise ValueError(f"Rider {rider_id} not found")
# Estimate fare before matching
estimated_fare = self.fare_calc.estimate(pickup, destination, vehicle_class)
trip = Trip(
rider = rider,
pickup = pickup,
destination = destination,
vehicle_class = vehicle_class,
fare = estimated_fare,
)
self._trips[trip.trip_id] = trip
print(f"Trip {trip.trip_id} requested. Estimated fare: $" + f"{estimated_fare:.2f}")
# Find nearest driver
nearby = self.matcher.find_nearest_drivers(pickup, vehicle_class)
if nearby:
self._assign_driver(trip, nearby[0])
else:
print(f"No drivers available nearby. Trip {trip.trip_id} queued.")
return trip
def _assign_driver(self, trip: Trip, driver: Driver):
driver.is_available = False
trip.driver = driver
self._transition(trip, TripStatus.DRIVER_ASSIGNED)
dist_to_pickup = driver.current_location.distance_km(trip.pickup)
eta_min = (dist_to_pickup / 30) * 60
print(f"Driver {driver.name} assigned. ETA: {eta_min:.0f} min")
def driver_arrived(self, trip_id: str) -> bool:
trip = self._trips.get(trip_id)
if not trip:
return False
return self._transition(trip, TripStatus.DRIVER_ARRIVED)
def start_trip(self, trip_id: str) -> bool:
trip = self._trips.get(trip_id)
if not trip:
return False
trip.start_time = datetime.now()
return self._transition(trip, TripStatus.IN_PROGRESS)
def complete_trip(self, trip_id: str) -> float:
trip = self._trips.get(trip_id)
if not trip:
return 0.0
trip.end_time = datetime.now()
final_fare = self.fare_calc.calculate_final(trip)
trip.fare = final_fare
if trip.driver:
trip.driver.is_available = True
self._transition(trip, TripStatus.COMPLETED)
print(f"Trip completed. Final fare: $" + f"{final_fare:.2f}")
return final_fare
def cancel_trip(self, trip_id: str, cancelled_by: str) -> float:
trip = self._trips.get(trip_id)
if not trip:
return 0.0
cancellation_fee = 0.0
# Cancellation fee if driver already dispatched
if (trip.status == TripStatus.DRIVER_ASSIGNED
and cancelled_by == "rider"):
cancellation_fee = 3.00
print(f"Cancellation fee: $" + f"{cancellation_fee:.2f}")
if trip.driver:
trip.driver.is_available = True
self._transition(trip, TripStatus.CANCELLED)
return cancellation_fee
def _transition(self, trip: Trip, new_status: TripStatus) -> bool:
allowed = self._transitions.get(trip.status, [])
if new_status not in allowed:
print(f"Invalid transition: {trip.status} -> {new_status}")
return False
trip.status = new_status
return True
Surge Pricing
class SurgePricing:
SURGE_THRESHOLDS = [
(0.5, 1.0), # demand/supply ratio 2.0: 2.5x cap
]
def calculate_multiplier(self, demand: int, available_drivers: int) -> float:
if available_drivers == 0:
return 2.5 # Max surge when no drivers
ratio = demand / available_drivers
for threshold, multiplier in self.SURGE_THRESHOLDS:
if ratio <= threshold:
return multiplier
return 2.5
Interview Checklist
- State machine: explicit valid transitions prevent illegal state changes
- Location: Haversine formula for accurate great-circle distance
- Driver matching: min-heap by distance, filter by availability and vehicle class
- Fare: base + per-km + per-minute; minimum fare; surge multiplier
- Cancellation: fee only if driver already dispatched and rider cancels
- Surge: demand/supply ratio determines multiplier; cap at 2.5x
- Production: replace in-memory with geospatial index (Redis GEOADD/GEORADIUS or H3 hexagons); real-time driver location via WebSocket
🏢 Asked at: Uber Interview Guide 2026: Dispatch Systems, Geospatial Algorithms, and Marketplace Engineering
🏢 Asked at: Lyft Interview Guide 2026: Rideshare Engineering, Real-Time Dispatch, and Safety Systems
🏢 Asked at: DoorDash Interview Guide
🏢 Asked at: Airbnb Interview Guide 2026: Search Systems, Trust and Safety, and Full-Stack Engineering
🏢 Asked at: Stripe Interview Guide