Coverage for modules\trucking\api.py: 32%

1from ninja import Router 

2from typing import List, Optional 

3from uuid import UUID 

4from core.auth import JWTAuth 

5from core.permissions import require_tenant, require_admin 

6from core.exceptions import NotFoundException 

7from .schemas import ( 

8 VehicleIn, VehicleOut, 

9 DriverIn, DriverOut, 

10 FuelCardIn, FuelCardOut, 

11 FuelLogIn, FuelLogOut, 

12 MaintenanceLogIn, MaintenanceLogOut, 

13 TripIn, TripOut, 

14 DeliveryIn, DeliveryOut, DeliveryUpdateProof, 

15) 

16from .models import Vehicle, Driver, FuelCard, FuelLog, MaintenanceLog, Trip, Delivery 

17 

18router = Router(tags=["Trucking"]) 

19 

20# ── Vehicles ────────────────────────────────────────────────────────────────── 

21@router.get("/vehicles", auth=JWTAuth(), response={200: List[VehicleOut]}) 

22def list_vehicles(request, status: Optional[str] = None): 

23 tenant_id = require_tenant(request) 

24 qs = Vehicle.objects.filter(tenant_id=tenant_id) 

25 if status: 

26 qs = qs.filter(status=status) 

27 return list(qs) 

28 

29 

30@router.post("/vehicles", auth=JWTAuth(), response={201: VehicleOut}) 

31@require_admin 

32def create_vehicle(request, payload: VehicleIn): 

33 tenant_id = require_tenant(request) 

34 vehicle = Vehicle.objects.create(tenant_id=tenant_id, **payload.dict()) 

35 return 201, vehicle 

36 

37 

38@router.get("/vehicles/{vehicle_id}", auth=JWTAuth(), response={200: VehicleOut}) 

39def get_vehicle(request, vehicle_id: UUID): 

40 tenant_id = require_tenant(request) 

41 vehicle = Vehicle.objects.filter(id=vehicle_id, tenant_id=tenant_id).first() 

42 if not vehicle: 

43 raise NotFoundException("Vehicle not found.") 

44 return vehicle 

45 

46 

47@router.put("/vehicles/{vehicle_id}", auth=JWTAuth(), response={200: VehicleOut}) 

48@require_admin 

49def update_vehicle(request, vehicle_id: UUID, payload: VehicleIn): 

50 tenant_id = require_tenant(request) 

51 vehicle = Vehicle.objects.filter(id=vehicle_id, tenant_id=tenant_id).first() 

52 if not vehicle: 

53 raise NotFoundException("Vehicle not found.") 

54 for key, value in payload.dict().items(): 

55 setattr(vehicle, key, value) 

56 vehicle.save() 

57 return vehicle 

58 

59 

60@router.delete("/vehicles/{vehicle_id}", auth=JWTAuth()) 

61@require_admin 

62def delete_vehicle(request, vehicle_id: UUID): 

63 tenant_id = require_tenant(request) 

64 vehicle = Vehicle.objects.filter(id=vehicle_id, tenant_id=tenant_id).first() 

65 if not vehicle: 

66 raise NotFoundException("Vehicle not found.") 

67 vehicle.delete() 

68 return {"success": True} 

69 

70 

71# ── Drivers ──────────────────────────────────────────────────────────────────── 

72@router.get("/drivers", auth=JWTAuth(), response={200: List[DriverOut]}) 

73def list_drivers(request, status: Optional[str] = None): 

74 tenant_id = require_tenant(request) 

75 qs = Driver.objects.filter(tenant_id=tenant_id) 

76 if status: 

77 qs = qs.filter(status=status) 

78 return list(qs) 

79 

80 

81@router.post("/drivers", auth=JWTAuth(), response={201: DriverOut}) 

82@require_admin 

83def create_driver(request, payload: DriverIn): 

84 tenant_id = require_tenant(request) 

85 driver = Driver.objects.create(tenant_id=tenant_id, **payload.dict()) 

86 return 201, driver 

87 

88 

89@router.get("/drivers/{driver_id}", auth=JWTAuth(), response={200: DriverOut}) 

90def get_driver(request, driver_id: UUID): 

91 tenant_id = require_tenant(request) 

92 driver = Driver.objects.filter(id=driver_id, tenant_id=tenant_id).first() 

93 if not driver: 

94 raise NotFoundException("Driver not found.") 

95 return driver 

96 

97 

98@router.put("/drivers/{driver_id}", auth=JWTAuth(), response={200: DriverOut}) 

99def update_driver(request, driver_id: UUID, payload: DriverIn): 

100 tenant_id = require_tenant(request) 

101 driver = Driver.objects.filter(id=driver_id, tenant_id=tenant_id).first() 

102 if not driver: 

103 raise NotFoundException("Driver not found.") 

104 for key, value in payload.dict().items(): 

105 setattr(driver, key, value) 

106 driver.save() 

107 return driver 

108 

109 

110# ── Fuel Cards ──────────────────────────────────────────────────────────────── 

111@router.get("/fuel-cards", auth=JWTAuth(), response={200: List[FuelCardOut]}) 

112def list_fuel_cards(request): 

113 tenant_id = require_tenant(request) 

114 return list(FuelCard.objects.filter(tenant_id=tenant_id)) 

115 

116 

117@router.post("/fuel-cards", auth=JWTAuth(), response={201: FuelCardOut}) 

118@require_admin 

119def create_fuel_card(request, payload: FuelCardIn): 

120 tenant_id = require_tenant(request) 

121 card = FuelCard.objects.create(tenant_id=tenant_id, **payload.dict()) 

122 return 201, card 

123 

124 

125@router.delete("/fuel-cards/{card_id}", auth=JWTAuth()) 

126def delete_fuel_card(request, card_id: UUID): 

127 tenant_id = require_tenant(request) 

128 card = FuelCard.objects.filter(id=card_id, tenant_id=tenant_id).first() 

129 if not card: 

130 raise NotFoundException("Fuel card not found.") 

131 card.delete() 

132 return {"success": True} 

133 

134 

135# ── Fuel Logs ───────────────────────────────────────────────────────────────── 

136@router.get("/fuel-logs", auth=JWTAuth(), response={200: List[FuelLogOut]}) 

137def list_fuel_logs(request, vehicle_id: Optional[UUID] = None): 

138 tenant_id = require_tenant(request) 

139 qs = FuelLog.objects.filter(tenant_id=tenant_id) 

140 if vehicle_id: 

141 qs = qs.filter(vehicle_id=vehicle_id) 

142 return list(qs.select_related("vehicle", "driver")) 

143 

144 

145@router.post("/fuel-logs", auth=JWTAuth(), response={201: FuelLogOut}) 

146@require_admin 

147def create_fuel_log(request, payload: FuelLogIn): 

148 tenant_id = require_tenant(request) 

149 log = FuelLog.objects.create(tenant_id=tenant_id, **payload.dict()) 

150 return 201, log 

151 

152 

153# ── Maintenance Logs ─────────────────────────────────────────────────────────── 

154@router.get("/maintenance-logs", auth=JWTAuth(), response={200: List[MaintenanceLogOut]}) 

155def list_maintenance_logs(request, vehicle_id: Optional[UUID] = None): 

156 tenant_id = require_tenant(request) 

157 qs = MaintenanceLog.objects.filter(tenant_id=tenant_id) 

158 if vehicle_id: 

159 qs = qs.filter(vehicle_id=vehicle_id) 

160 return list(qs.select_related("vehicle")) 

161 

162 

163@router.post("/maintenance-logs", auth=JWTAuth(), response={201: MaintenanceLogOut}) 

164@require_admin 

165def create_maintenance_log(request, payload: MaintenanceLogIn): 

166 tenant_id = require_tenant(request) 

167 log = MaintenanceLog.objects.create(tenant_id=tenant_id, **payload.dict()) 

168 return 201, log 

169 

170 

171@router.put("/maintenance-logs/{log_id}", auth=JWTAuth(), response={200: MaintenanceLogOut}) 

172def update_maintenance_log(request, log_id: UUID, payload: MaintenanceLogIn): 

173 tenant_id = require_tenant(request) 

174 log = MaintenanceLog.objects.filter(id=log_id, tenant_id=tenant_id).first() 

175 if not log: 

176 raise NotFoundException("Maintenance log not found.") 

177 for key, value in payload.dict().items(): 

178 setattr(log, key, value) 

179 log.save() 

180 return log 

181 

182 

183# ── Trips ───────────────────────────────────────────────────────────────────── 

184@router.get("/trips", auth=JWTAuth(), response={200: List[TripOut]}) 

185def list_trips(request, status: Optional[str] = None, driver_id: Optional[UUID] = None): 

186 tenant_id = require_tenant(request) 

187 qs = Trip.objects.filter(tenant_id=tenant_id) 

188 if status: 

189 qs = qs.filter(status=status) 

190 if driver_id: 

191 qs = qs.filter(driver_id=driver_id) 

192 return list(qs.select_related("vehicle", "driver")) 

193 

194 

195@router.post("/trips", auth=JWTAuth(), response={201: TripOut}) 

196@require_admin 

197def create_trip(request, payload: TripIn): 

198 tenant_id = require_tenant(request) 

199 trip = Trip.objects.create(tenant_id=tenant_id, **payload.dict()) 

200 return 201, trip 

201 

202 

203@router.get("/trips/{trip_id}", auth=JWTAuth(), response={200: TripOut}) 

204def get_trip(request, trip_id: UUID): 

205 tenant_id = require_tenant(request) 

206 trip = Trip.objects.filter(id=trip_id, tenant_id=tenant_id).first() 

207 if not trip: 

208 raise NotFoundException("Trip not found.") 

209 return trip 

210 

211 

212@router.put("/trips/{trip_id}", auth=JWTAuth(), response={200: TripOut}) 

213def update_trip(request, trip_id: UUID, payload: TripIn): 

214 tenant_id = require_tenant(request) 

215 trip = Trip.objects.filter(id=trip_id, tenant_id=tenant_id).first() 

216 if not trip: 

217 raise NotFoundException("Trip not found.") 

218 for key, value in payload.dict().items(): 

219 setattr(trip, key, value) 

220 trip.save() 

221 return trip 

222 

223 

224# ── Deliveries ───────────────────────────────────────────────────────────────── 

225@router.get("/deliveries", auth=JWTAuth(), response={200: List[DeliveryOut]}) 

226def list_deliveries(request, trip_id: Optional[UUID] = None): 

227 tenant_id = require_tenant(request) 

228 qs = Delivery.objects.filter(tenant_id=tenant_id) 

229 if trip_id: 

230 qs = qs.filter(trip_id=trip_id) 

231 return qs 

232 

233 

234@router.post("/deliveries", auth=JWTAuth(), response={201: DeliveryOut}) 

235def create_delivery(request, payload: DeliveryIn): 

236 tenant_id = require_tenant(request) 

237 data = payload.dict() 

238 trip_id = data.pop("trip_id", None) 

239 if trip_id: 

240 data["trip_id"] = trip_id 

241 delivery = Delivery.objects.create(tenant_id=tenant_id, **data) 

242 return delivery 

243 

244 

245@router.get("/deliveries/{delivery_id}", auth=JWTAuth(), response={200: DeliveryOut}) 

246def get_delivery(request, delivery_id: UUID): 

247 tenant_id = require_tenant(request) 

248 delivery = Delivery.objects.filter(id=delivery_id, tenant_id=tenant_id).first() 

249 if not delivery: 

250 raise NotFoundException("Delivery not found.") 

251 return delivery 

252 

253 

254@router.put("/deliveries/{delivery_id}", auth=JWTAuth(), response={200: DeliveryOut}) 

255def update_delivery(request, delivery_id: UUID, payload: DeliveryIn): 

256 tenant_id = require_tenant(request) 

257 delivery = Delivery.objects.filter(id=delivery_id, tenant_id=tenant_id).first() 

258 if not delivery: 

259 raise NotFoundException("Delivery not found.") 

260 data = payload.dict() 

261 trip_id = data.pop("trip_id", None) 

262 if trip_id is not None: 

263 delivery.trip_id = trip_id 

264 for key, value in data.items(): 

265 if key != "trip_id": 

266 setattr(delivery, key, value) 

267 delivery.save() 

268 return delivery 

269 

270 

271@router.patch("/deliveries/{delivery_id}/proof", auth=JWTAuth(), response={200: DeliveryOut}) 

272def update_delivery_proof(request, delivery_id: UUID, payload: DeliveryUpdateProof): 

273 tenant_id = require_tenant(request) 

274 delivery = Delivery.objects.filter(id=delivery_id, tenant_id=tenant_id).first() 

275 if not delivery: 

276 raise NotFoundException("Delivery not found.") 

277 for key, value in payload.dict(exclude_unset=True).items(): 

278 setattr(delivery, key, value) 

279 delivery.save() 

280 return delivery 

281 

282 

283@router.delete("/deliveries/{delivery_id}", auth=JWTAuth(), response={204: None}) 

284def delete_delivery(request, delivery_id: UUID): 

285 tenant_id = require_tenant(request) 

286 delivery = Delivery.objects.filter(id=delivery_id, tenant_id=tenant_id).first() 

287 if not delivery: 

288 raise NotFoundException("Delivery not found.") 

289 delivery.delete() 

290 return None