Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from typing import Optional
3import secrets
4 
5app = FastAPI()
6 
7users = {}
8next_user_id = 1
9trips = {}
10next_trip_id = 1
11destinations = {}
12next_destination_id = 1
13tokens = {}
14 
15def get_current_user(authorization: Optional[str] = Header(None)):
16    if not authorization or not authorization.startswith("Bearer "):
17        raise HTTPException(status_code=401, detail="Invalid auth")
18    token = authorization.split(" ")[1]
19    user_id = tokens.get(token)
20    if not user_id:
21        raise HTTPException(status_code=401, detail="Invalid token")
22    return user_id
23 
24@app.post("/signup")
25def signup(username: str, password: str):
26    global next_user_id
27    user_id = next_user_id
28    next_user_id += 1
29    users[user_id] = {"id": user_id, "username": username, "password": password, "role": "admin"}
30    return {"user_id": user_id, "role": "admin"}
31 
32@app.post("/login")
33def login(username: str, password: str):
34    for uid, u in users.items():
35        if u["username"] == username and u["password"] == password:
36            token = secrets.token_hex(16)
37            tokens[token] = uid
38            return {"token": token}
39    raise HTTPException(status_code=401, detail="Invalid credentials")
40 
41@app.post("/trip")
42def create_trip(name: str, authorization: Optional[str] = Header(None)):
43    user_id = get_current_user(authorization)
44    global next_trip_id
45    trip_id = next_trip_id
46    next_trip_id += 1
47    trips[trip_id] = {"id": trip_id, "name": name, "admin_id": user_id, "co_planners": [], "dates": None}
48    return {"trip_id": trip_id}
49 
50@app.get("/trip/{trip_id}")
51def get_trip(trip_id: int, authorization: Optional[str] = Header(None)):
52    user_id = get_current_user(authorization)
53    trip = trips.get(trip_id)
54    if not trip:
55        raise HTTPException(status_code=404, detail="Trip not found")
56    return trip
57 
58@app.post("/promote/{user_id}")
59def promote_to_co_planner(user_id: int, trip_id: int, authorization: Optional[str] = Header(None)):
60    admin_id = get_current_user(authorization)
61    trip = trips.get(trip_id)
62    if not trip:
63        raise HTTPException(status_code=404, detail="Trip not found")
64    if trip["admin_id"] != admin_id:
65        raise HTTPException(status_code=403, detail="Only admin can promote")
66    if user_id not in users:
67        raise HTTPException(status_code=404, detail="User not found")
68    if user_id not in trip["co_planners"]:
69        trip["co_planners"].append(user_id)
70    return {"status": "ok"}
71 
72@app.post("/trip/{trip_id}/dates")
73def set_trip_dates(trip_id: int, start_date: str, end_date: str, authorization: Optional[str] = Header(None)):
74    user_id = get_current_user(authorization)
75    trip = trips.get(trip_id)
76    if not trip:
77        raise HTTPException(status_code=404, detail="Trip not found")
78    if user_id != trip["admin_id"] and user_id not in trip["co_planners"]:
79        raise HTTPException(status_code=403, detail="Only admin or co-planner can edit dates")
80    trip["dates"] = {"start": start_date, "end": end_date}
81    return {"status": "ok"}
82 
83@app.post("/destination")
84def create_destination(trip_id: int, name: str, authorization: Optional[str] = Header(None)):
85    user_id = get_current_user(authorization)
86    trip = trips.get(trip_id)
87    if not trip:
88        raise HTTPException(status_code=404, detail="Trip not found")
89    if user_id != trip["admin_id"] and user_id not in trip["co_planners"]:
90        raise HTTPException(status_code=403, detail="Only admin or co-planner can add destinations")
91    global next_destination_id
92    dest_id = next_destination_id
93    next_destination_id += 1
94    destinations[dest_id] = {"id": dest_id, "trip_id": trip_id, "name": name}
95    return {"destination_id": dest_id}
96 
97@app.get("/destination/{dest_id}")
98def get_destination(dest_id: int, authorization: Optional[str] = Header(None)):
99    user_id = get_current_user(authorization)
100    dest = destinations.get(dest_id)
101    if not dest:
102        raise HTTPException(status_code=404, detail="Destination not found")
103    return dest
104 
105@app.post("/trip/{trip_id}/role")
106def assign_role(trip_id: int, user_id: int, role: str, authorization: Optional[str] = Header(None)):
107    admin_id = get_current_user(authorization)
108    trip = trips.get(trip_id)
109    if not trip:
110        raise HTTPException(status_code=404, detail="Trip not found")
111    if trip["admin_id"] != admin_id:
112        raise HTTPException(status_code=403, detail="Only admin can assign roles")
113    if user_id not in users:
114        raise HTTPException(status_code=404, detail="User not found")
115    if role not in ["guide", "guest"]:
116        raise HTTPException(status_code=400, detail="Role must be guide or guest")
117    if "roles" not in trip:
118        trip["roles"] = {}
119    trip["roles"][user_id] = role
120    return {"status": "ok"}

1fastapi
2uvicorn