Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5import uvicorn
6 
7app = FastAPI()
8 
9users = {}
10rides = {}
11tokens = {}
12user_id_counter = 1
13ride_id_counter = 1
14 
15class UserCreate(BaseModel):
16    username: str
17    password: str
18 
19class LoginRequest(BaseModel):
20    username: str
21    password: str
22 
23class RideCreate(BaseModel):
24    origin: str
25    destination: str
26    time: str
27    seats: int
28 
29class RideSearch(BaseModel):
30    from_: str
31    to: str
32 
33def get_current_user(authorization: Optional[str] = Header(None)):
34    if not authorization:
35        raise HTTPException(status_code=401, detail="Missing Authorization header")
36    token = authorization.replace("Bearer ", "")
37    user_id = tokens.get(token)
38    if not user_id:
39        raise HTTPException(status_code=401, detail="Invalid token")
40    return user_id
41 
42@app.post("/signup")
43def signup(user: UserCreate):
44    global user_id_counter
45    if user.username in users:
46        raise HTTPException(status_code=400, detail="Username already exists")
47    users[user.username] = {"id": user_id_counter, "username": user.username, "password": user.password}
48    user_id_counter += 1
49    return {"id": users[user.username]["id"], "username": user.username}
50 
51@app.post("/login")
52def login(req: LoginRequest):
53    user = users.get(req.username)
54    if not user or user["password"] != req.password:
55        raise HTTPException(status_code=401, detail="Invalid credentials")
56    token = secrets.token_hex(16)
57    tokens[token] = user["id"]
58    return {"token": token}
59 
60@app.get("/rides/{ride_id}")
61def get_ride(ride_id: int, authorization: Optional[str] = Header(None)):
62    get_current_user(authorization)
63    ride = rides.get(ride_id)
64    if not ride:
65        raise HTTPException(status_code=404, detail="Ride not found")
66    return ride
67 
68@app.post("/rides")
69def create_ride(ride: RideCreate, authorization: Optional[str] = Header(None)):
70    global ride_id_counter
71    driver_id = get_current_user(authorization)
72    ride_data = ride.dict()
73    ride_data["id"] = ride_id_counter
74    ride_data["driver_id"] = driver_id
75    rides[ride_id_counter] = ride_data
76    ride_id_counter += 1
77    return ride_data
78 
79@app.get("/rides/search")
80def search_rides(from_: str, to: str, authorization: Optional[str] = Header(None)):
81    get_current_user(authorization)
82    results = []
83    for ride_id, ride in rides.items():
84        if ride["origin"] == from_ and ride["destination"] == to:
85            driver = users.get(ride["driver_id"])
86            if driver:
87                ride_with_driver = {**ride, "driver_username": driver["username"]}
88            else:
89                ride_with_driver = {**ride, "driver_username": "unknown"}
90            results.append(ride_with_driver)
91    return results
92 
93@app.get("/users/{user_id}")
94def get_user(user_id: int, authorization: Optional[str] = Header(None)):
95    get_current_user(authorization)
96    for username, user in users.items():
97        if user["id"] == user_id:
98            return user
99    raise HTTPException(status_code=404, detail="User not found")
100 
101@app.post("/users")
102def create_user(user: UserCreate):
103    return signup(user)

1fastapi
2uvicorn