Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5 
6app = FastAPI()
7 
8users = {}
9stations = {}
10bikes = {}
11rentals = {}
12maintenance_logs = {}
13tokens = {}
14next_ids = {"users": 1, "stations": 1, "bikes": 1, "rentals": 1, "maintenance_logs": 1}
15 
16class UserSignup(BaseModel):
17    username: str
18    password: str
19 
20class UserLogin(BaseModel):
21    username: str
22    password: str
23 
24class StationCreate(BaseModel):
25    name: str
26    location: str
27    available_bikes: int = 0
28 
29class BikeCreate(BaseModel):
30    model: str
31    station_id: int
32 
33class RentalStart(BaseModel):
34    bike_id: int
35    user_id: int
36 
37class RentalEnd(BaseModel):
38    rental_id: int
39 
40class MaintenanceLogCreate(BaseModel):
41    bike_id: int
42    description: str
43 
44def get_current_user(authorization: Optional[str] = Header(None)):
45    if not authorization:
46        raise HTTPException(status_code=401, detail="Missing token")
47    token = authorization.replace("Bearer ", "")
48    for uid, t in tokens.items():
49        if t == token:
50            return uid
51    raise HTTPException(status_code=401, detail="Invalid token")
52 
53@app.post("/signup")
54def signup(user: UserSignup):
55    uid = next_ids["users"]
56    next_ids["users"] += 1
57    users[uid] = {"id": uid, "username": user.username, "password": user.password}
58    token = secrets.token_hex(16)
59    tokens[uid] = token
60    return {"user_id": uid, "token": token}
61 
62@app.post("/login")
63def login(user: UserLogin):
64    for uid, u in users.items():
65        if u["username"] == user.username and u["password"] == user.password:
66            token = secrets.token_hex(16)
67            tokens[uid] = token
68            return {"user_id": uid, "token": token}
69    raise HTTPException(status_code=401, detail="Invalid credentials")
70 
71@app.post("/stations")
72def create_station(station: StationCreate, authorization: Optional[str] = Header(None)):
73    get_current_user(authorization)
74    sid = next_ids["stations"]
75    next_ids["stations"] += 1
76    stations[sid] = {"id": sid, "name": station.name, "location": station.location, "available_bikes": station.available_bikes}
77    return stations[sid]
78 
79@app.get("/stations/{station_id}")
80def get_station(station_id: int, authorization: Optional[str] = Header(None)):
81    get_current_user(authorization)
82    if station_id not in stations:
83        raise HTTPException(status_code=404, detail="Station not found")
84    return stations[station_id]
85 
86@app.post("/bikes")
87def create_bike(bike: BikeCreate, authorization: Optional[str] = Header(None)):
88    get_current_user(authorization)
89    bid = next_ids["bikes"]
90    next_ids["bikes"] += 1
91    bikes[bid] = {"id": bid, "model": bike.model, "station_id": bike.station_id, "status": "available"}
92    return bikes[bid]
93 
94@app.get("/bikes/{bike_id}")
95def get_bike(bike_id: int, authorization: Optional[str] = Header(None)):
96    get_current_user(authorization)
97    if bike_id not in bikes:
98        raise HTTPException(status_code=404, detail="Bike not found")
99    return bikes[bike_id]
100 
101@app.post("/rentals/start")
102def start_rental(rental: RentalStart, authorization: Optional[str] = Header(None)):
103    get_current_user(authorization)
104    if rental.bike_id not in bikes:
105        raise HTTPException(status_code=404, detail="Bike not found")
106    if bikes[rental.bike_id]["status"] != "available":
107        raise HTTPException(status_code=400, detail="Bike not available")
108    rid = next_ids["rentals"]
109    next_ids["rentals"] += 1
110    rentals[rid] = {"id": rid, "bike_id": rental.bike_id, "user_id": rental.user_id, "start_time": "now", "end_time": None}
111    bikes[rental.bike_id]["status"] = "rented"
112    return rentals[rid]
113 
114@app.post("/rentals/end")
115def end_rental(rental: RentalEnd, authorization: Optional[str] = Header(None)):
116    get_current_user(authorization)
117    if rental.rental_id not in rentals:
118        raise HTTPException(status_code=404, detail="Rental not found")
119    r = rentals[rental.rental_id]
120    if r["end_time"] is not None:
121        raise HTTPException(status_code=400, detail="Rental already ended")
122    r["end_time"] = "now"
123    bikes[r["bike_id"]]["status"] = "available"
124    return r
125 
126@app.get("/rentals/{rental_id}")
127def get_rental(rental_id: int, authorization: Optional[str] = Header(None)):
128    get_current_user(authorization)
129    if rental_id not in rentals:
130        raise HTTPException(status_code=404, detail="Rental not found")
131    return rentals[rental_id]
132 
133@app.post("/maintenance_logs")
134def create_maintenance_log(log: MaintenanceLogCreate, authorization: Optional[str] = Header(None)):
135    get_current_user(authorization)
136    if log.bike_id not in bikes:
137        raise HTTPException(status_code=404, detail="Bike not found")
138    mid = next_ids["maintenance_logs"]
139    next_ids["maintenance_logs"] += 1
140    maintenance_logs[mid] = {"id": mid, "bike_id": log.bike_id, "description": log.description, "timestamp": "now"}
141    return maintenance_logs[mid]
142 
143@app.get("/maintenance_logs/{log_id}")
144def get_maintenance_log(log_id: int, authorization: Optional[str] = Header(None)):
145    get_current_user(authorization)
146    if log_id not in maintenance_logs:
147        raise HTTPException(status_code=404, detail="Maintenance log not found")
148    return maintenance_logs[log_id]

1fastapi
2uvicorn