Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, Dict, List
4import secrets
5import uvicorn
6 
7app = FastAPI()
8 
9# In-memory stores
10users = {}
11drivers = {}
12tokens = {}
13next_user_id = 1
14next_driver_id = 1
15 
16# Simple auth dependency
17def get_current_user(authorization: str = Header(None)):
18    if not authorization:
19        raise HTTPException(status_code=401, detail="Missing auth token")
20    token = authorization.replace("Bearer ", "")
21    user_id = tokens.get(token)
22    if not user_id:
23        raise HTTPException(status_code=401, detail="Invalid token")
24    return user_id
25 
26# Models
27class SignupRequest(BaseModel):
28    username: str
29    password: str
30 
31class LoginRequest(BaseModel):
32    username: str
33    password: str
34 
35class DriverCreate(BaseModel):
36    name: str
37    license_info: Optional[str] = None
38    route_assignments: Optional[List[str]] = []
39    is_admin: Optional[bool] = False
40 
41class DriverUpdate(BaseModel):
42    name: Optional[str] = None
43    license_info: Optional[str] = None
44    route_assignments: Optional[List[str]] = None
45    is_admin: Optional[bool] = None
46 
47# Auth endpoints
48@app.post("/signup")
49def signup(req: SignupRequest):
50    global next_user_id
51    for u in users.values():
52        if u["username"] == req.username:
53            raise HTTPException(status_code=400, detail="Username taken")
54    user_id = next_user_id
55    next_user_id += 1
56    users[user_id] = {"id": user_id, "username": req.username, "password": req.password}
57    token = secrets.token_hex(32)
58    tokens[token] = user_id
59    return {"user_id": user_id, "token": token}
60 
61@app.post("/login")
62def login(req: LoginRequest):
63    for u in users.values():
64        if u["username"] == req.username and u["password"] == req.password:
65            token = secrets.token_hex(32)
66            tokens[token] = u["id"]
67            return {"user_id": u["id"], "token": token}
68    raise HTTPException(status_code=401, detail="Invalid credentials")
69 
70# Driver endpoints
71@app.get("/drivers/{driver_id}")
72def get_driver(driver_id: int, authorization: str = Header(None)):
73    get_current_user(authorization)
74    driver = drivers.get(driver_id)
75    if not driver:
76        raise HTTPException(status_code=404, detail="Driver not found")
77    return driver
78 
79@app.post("/drivers")
80def create_driver(driver: DriverCreate, authorization: str = Header(None)):
81    get_current_user(authorization)
82    global next_driver_id
83    driver_id = next_driver_id
84    next_driver_id += 1
85    drivers[driver_id] = {
86        "id": driver_id,
87        "name": driver.name,
88        "license_info": driver.license_info,
89        "route_assignments": driver.route_assignments or [],
90        "is_admin": driver.is_admin
91    }
92    return drivers[driver_id]
93 
94@app.put("/drivers/{driver_id}")
95def update_driver(driver_id: int, driver: DriverUpdate, authorization: str = Header(None)):
96    get_current_user(authorization)
97    existing = drivers.get(driver_id)
98    if not existing:
99        raise HTTPException(status_code=404, detail="Driver not found")
100    if driver.name is not None:
101        existing["name"] = driver.name
102    if driver.license_info is not None:
103        existing["license_info"] = driver.license_info
104    if driver.route_assignments is not None:
105        existing["route_assignments"] = driver.route_assignments
106    if driver.is_admin is not None:
107        existing["is_admin"] = driver.is_admin
108    return existing

1fastapi
2uvicorn