Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List
4import random
5import string
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11rides = {}
12ride_id_counter = 1
13user_id_counter = 1
14 
15def generate_token():
16    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
17 
18def get_current_user(authorization: str = Header(None)):
19    if not authorization:
20        raise HTTPException(status_code=401, detail="Missing auth token")
21    token = authorization.replace("Bearer ", "")
22    user_id = tokens.get(token)
23    if not user_id:
24        raise HTTPException(status_code=401, detail="Invalid auth token")
25    return user_id
26 
27class SignupRequest(BaseModel):
28    name: str
29    email: str
30    password: str
31 
32class LoginRequest(BaseModel):
33    email: str
34    password: str
35 
36class RideCreate(BaseModel):
37    start_location: str
38    time: str
39    seats: int
40    driver_id: Optional[int] = None
41    status: str = "active"
42 
43class RideUpdate(BaseModel):
44    start_location: Optional[str] = None
45    time: Optional[str] = None
46    seats: Optional[int] = None
47    driver_id: Optional[int] = None
48    status: Optional[str] = None
49 
50@app.post("/signup")
51def signup(req: SignupRequest):
52    global user_id_counter
53    user_id = user_id_counter
54    user_id_counter += 1
55    users[user_id] = {"id": user_id, "name": req.name, "email": req.email, "password": req.password}
56    token = generate_token()
57    tokens[token] = user_id
58    return {"user_id": user_id, "token": token}
59 
60@app.post("/login")
61def login(req: LoginRequest):
62    for uid, u in users.items():
63        if u["email"] == req.email and u["password"] == req.password:
64            token = generate_token()
65            tokens[token] = uid
66            return {"user_id": uid, "token": token}
67    raise HTTPException(status_code=401, detail="Invalid credentials")
68 
69@app.get("/users/{user_id}")
70def get_user(user_id: int, authorization: str = Header(None)):
71    get_current_user(authorization)
72    user = users.get(user_id)
73    if not user:
74        raise HTTPException(status_code=404, detail="User not found")
75    return user
76 
77@app.post("/rides")
78def create_ride(ride: RideCreate, authorization: str = Header(None)):
79    global ride_id_counter
80    get_current_user(authorization)
81    ride_id = ride_id_counter
82    ride_id_counter += 1
83    rides[ride_id] = {
84        "id": ride_id,
85        "start_location": ride.start_location,
86        "time": ride.time,
87        "seats": ride.seats,
88        "driver_id": ride.driver_id,
89        "status": ride.status
90    }
91    return rides[ride_id]
92 
93@app.get("/rides/{ride_id}")
94def get_ride(ride_id: int, authorization: str = Header(None)):
95    get_current_user(authorization)
96    ride = rides.get(ride_id)
97    if not ride:
98        raise HTTPException(status_code=404, detail="Ride not found")
99    return ride
100 
101@app.patch("/rides/{ride_id}")
102def update_ride(ride_id: int, update: RideUpdate, authorization: str = Header(None)):
103    get_current_user(authorization)
104    ride = rides.get(ride_id)
105    if not ride:
106        raise HTTPException(status_code=404, detail="Ride not found")
107    if update.start_location is not None:
108        ride["start_location"] = update.start_location
109    if update.time is not None:
110        ride["time"] = update.time
111    if update.seats is not None:
112        ride["seats"] = update.seats
113    if update.driver_id is not None:
114        ride["driver_id"] = update.driver_id
115    if update.status is not None:
116        ride["status"] = update.status
117    return ride

1fastapi
2uvicorn