File size: 4,333 Bytes
fcf8749 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | """
Drivers API endpoint.
Handles GET /api/v1/drivers/{id} for driver details.
"""
from datetime import date, timedelta
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.database import get_db
from app.models import Driver, DriverStatsDaily
from app.schemas.driver import DriverResponse, DriverStatsResponse
router = APIRouter(prefix="/drivers", tags=["Drivers"])
@router.get(
"/{driver_id}",
response_model=DriverResponse,
summary="Get driver details",
description="Returns driver details including recent fairness statistics (last 7 days).",
)
async def get_driver(
driver_id: UUID,
db: AsyncSession = Depends(get_db),
) -> DriverResponse:
"""Get driver details by ID."""
# Fetch driver with recent stats
result = await db.execute(
select(Driver)
.where(Driver.id == driver_id)
.options(selectinload(Driver.daily_stats))
)
driver = result.scalar_one_or_none()
if not driver:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Driver with ID {driver_id} not found",
)
# Get stats for last 7 days
seven_days_ago = date.today() - timedelta(days=7)
recent_stats = [
DriverStatsResponse(
date=stat.date,
avg_workload_score=stat.avg_workload_score,
total_routes=stat.total_routes,
gini_contribution=stat.gini_contribution,
reported_stress_level=stat.reported_stress_level,
reported_fairness_score=stat.reported_fairness_score,
)
for stat in driver.daily_stats
if stat.date >= seven_days_ago
]
# Sort by date descending
recent_stats.sort(key=lambda x: x.date, reverse=True)
return DriverResponse(
id=driver.id,
external_id=driver.external_id,
name=driver.name,
phone=driver.phone,
whatsapp_number=driver.whatsapp_number,
preferred_language=driver.preferred_language.value,
vehicle_type=driver.vehicle_type.value,
vehicle_capacity_kg=driver.vehicle_capacity_kg,
license_number=driver.license_number,
created_at=driver.created_at,
updated_at=driver.updated_at,
recent_stats=recent_stats,
)
@router.get(
"/external/{external_id}",
response_model=DriverResponse,
summary="Get driver by external ID",
description="Returns driver details by external ID (from integration system).",
)
async def get_driver_by_external_id(
external_id: str,
db: AsyncSession = Depends(get_db),
) -> DriverResponse:
"""Get driver details by external ID."""
result = await db.execute(
select(Driver)
.where(Driver.external_id == external_id)
.options(selectinload(Driver.daily_stats))
)
driver = result.scalar_one_or_none()
if not driver:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Driver with external ID {external_id} not found",
)
# Get stats for last 7 days
seven_days_ago = date.today() - timedelta(days=7)
recent_stats = [
DriverStatsResponse(
date=stat.date,
avg_workload_score=stat.avg_workload_score,
total_routes=stat.total_routes,
gini_contribution=stat.gini_contribution,
reported_stress_level=stat.reported_stress_level,
reported_fairness_score=stat.reported_fairness_score,
)
for stat in driver.daily_stats
if stat.date >= seven_days_ago
]
recent_stats.sort(key=lambda x: x.date, reverse=True)
return DriverResponse(
id=driver.id,
external_id=driver.external_id,
name=driver.name,
phone=driver.phone,
whatsapp_number=driver.whatsapp_number,
preferred_language=driver.preferred_language.value,
vehicle_type=driver.vehicle_type.value,
vehicle_capacity_kg=driver.vehicle_capacity_kg,
license_number=driver.license_number,
created_at=driver.created_at,
updated_at=driver.updated_at,
recent_stats=recent_stats,
)
|