Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List
4import secrets
5 
6app = FastAPI()
7 
8users = {}
9rides = {}
10bookings = {}
11tokens = {}
12next_user_id = 1
13next_ride_id = 1
14next_booking_id = 1
15 
16class SignupRequest(BaseModel):
17    username: str
18    password: str
19 
20class LoginRequest(BaseModel):
21    username: str
22    password: str
23 
24class RideCreate(BaseModel):
25    driver_id: int
26    origin: str
27    destination: str
28    departure_time: str
29    available_seats: int
30 
31class BookingCreate(BaseModel):
32    ride_id: int
33    passenger_id: int
34 
35def get_current_user(authorization: Optional[str] = Header(None)):
36    if not authorization:
37        raise HTTPException(status_code=401, detail="Missing auth header")
38    token = authorization.replace("Bearer ", "")
39    if token not in tokens:
40        raise HTTPException(status_code=401, detail="Invalid token")
41    return tokens[token]
42 
43@app.post("/signup")
44def signup(req: SignupRequest):
45    global next_user_id
46    for u in users.values():
47        if u["username"] == req.username:
48            raise HTTPException(status_code=400, detail="Username taken")
49    user_id = next_user_id
50    next_user_id += 1
51    users[user_id] = {"id": user_id, "username": req.username, "password": req.password}
52    return {"id": user_id, "username": req.username}
53 
54@app.post("/login")
55def login(req: LoginRequest):
56    for u in users.values():
57        if u["username"] == req.username and u["password"] == req.password:
58            token = secrets.token_hex(16)
59            tokens[token] = u["id"]
60            return {"token": token}
61    raise HTTPException(status_code=401, detail="Invalid credentials")
62 
63@app.post("/rides")
64def create_ride(ride: RideCreate, authorization: Optional[str] = Header(None)):
65    user_id = get_current_user(authorization)
66    if ride.driver_id != user_id:
67        raise HTTPException(status_code=403, detail="Driver ID mismatch")
68    global next_ride_id
69    ride_id = next_ride_id
70    next_ride_id += 1
71    rides[ride_id] = {
72        "id": ride_id,
73        "driver_id": ride.driver_id,
74        "origin": ride.origin,
75        "destination": ride.destination,
76        "departure_time": ride.departure_time,
77        "available_seats": ride.available_seats
78    }
79    return rides[ride_id]
80 
81@app.get("/rides/{ride_id}")
82def get_ride(ride_id: int, authorization: Optional[str] = Header(None)):
83    user_id = get_current_user(authorization)
84    if ride_id not in rides:
85        raise HTTPException(status_code=404, detail="Ride not found")
86    ride = rides[ride_id].copy()
87    passenger_ids = [b["passenger_id"] for b in bookings.values() if b["ride_id"] == ride_id]
88    ride["passengers"] = [users[pid] for pid in passenger_ids if pid in users]
89    return ride
90 
91@app.post("/bookings")
92def create_booking(booking: BookingCreate, authorization: Optional[str] = Header(None)):
93    user_id = get_current_user(authorization)
94    if booking.passenger_id != user_id:
95        raise HTTPException(status_code=403, detail="Passenger ID mismatch")
96    if booking.ride_id not in rides:
97        raise HTTPException(status_code=404, detail="Ride not found")
98    ride = rides[booking.ride_id]
99    if ride["available_seats"] <= 0:
100        raise HTTPException(status_code=400, detail="No available seats")
101    global next_booking_id
102    booking_id = next_booking_id
103    next_booking_id += 1
104    bookings[booking_id] = {
105        "id": booking_id,
106        "ride_id": booking.ride_id,
107        "passenger_id": booking.passenger_id
108    }
109    ride["available_seats"] -= 1
110    return bookings[booking_id]
111 
112@app.get("/bookings/{booking_id}")
113def get_booking(booking_id: int, authorization: Optional[str] = Header(None)):
114    user_id = get_current_user(authorization)
115    if booking_id not in bookings:
116        raise HTTPException(status_code=404, detail="Booking not found")
117    return bookings[booking_id]
118 
119@app.get("/users/{user_id}")
120def get_user(user_id: int, authorization: Optional[str] = Header(None)):
121    user_id_token = get_current_user(authorization)
122    if user_id not in users:
123        raise HTTPException(status_code=404, detail="User not found")
124    return {"id": users[user_id]["id"], "username": users[user_id]["username"]}

1fastapi
2uvicorn