Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List
4from datetime import date
5import secrets
6import time
7 
8app = FastAPI()
9 
10users = {}
11tokens = {}
12trips = {}
13trip_id_counter = 1
14change_logs = []
15 
16def get_current_user(authorization: str = Header(None)):
17    if not authorization:
18        raise HTTPException(status_code=401, detail="Missing auth token")
19    token = authorization.replace("Bearer ", "")
20    user_id = tokens.get(token)
21    if not user_id:
22        raise HTTPException(status_code=401, detail="Invalid token")
23    return user_id
24 
25class SignupRequest(BaseModel):
26    username: str
27    password: str
28 
29class LoginRequest(BaseModel):
30    username: str
31    password: str
32 
33class TripCreate(BaseModel):
34    destination: str
35    start_date: date
36    end_date: date
37    budget: float
38    traveler_names: List[str]
39    lodging_tier: str
40 
41class TripUpdate(BaseModel):
42    destination: Optional[str] = None
43    start_date: Optional[date] = None
44    end_date: Optional[date] = None
45    budget: Optional[float] = None
46    traveler_names: Optional[List[str]] = None
47    lodging_tier: Optional[str] = None
48 
49@app.post("/signup")
50def signup(req: SignupRequest):
51    if req.username in users:
52        raise HTTPException(status_code=400, detail="Username already exists")
53    user_id = len(users) + 1
54    users[req.username] = {"id": user_id, "password": req.password, "username": req.username}
55    return {"id": user_id, "username": req.username}
56 
57@app.post("/login")
58def login(req: LoginRequest):
59    user = users.get(req.username)
60    if not user or user["password"] != req.password:
61        raise HTTPException(status_code=401, detail="Invalid credentials")
62    token = secrets.token_hex(32)
63    tokens[token] = user["id"]
64    return {"token": token}
65 
66@app.post("/trips")
67def create_trip(trip: TripCreate, authorization: str = Header(None)):
68    user_id = get_current_user(authorization)
69    global trip_id_counter
70    trip_id = trip_id_counter
71    trip_id_counter += 1
72    trip_data = trip.dict()
73    trip_data["id"] = trip_id
74    trip_data["user_id"] = user_id
75    trips[trip_id] = trip_data
76    return trip_data
77 
78@app.get("/trips/{trip_id}")
79def get_trip(trip_id: int, authorization: str = Header(None)):
80    user_id = get_current_user(authorization)
81    trip = trips.get(trip_id)
82    if not trip:
83        raise HTTPException(status_code=404, detail="Trip not found")
84    return trip
85 
86@app.patch("/trips/{trip_id}")
87def update_trip(trip_id: int, update: TripUpdate, authorization: str = Header(None)):
88    user_id = get_current_user(authorization)
89    trip = trips.get(trip_id)
90    if not trip:
91        raise HTTPException(status_code=404, detail="Trip not found")
92    if trip["user_id"] != user_id:
93        raise HTTPException(status_code=403, detail="Not your trip")
94    
95    changes = {}
96    for field, value in update.dict(exclude_unset=True).items():
97        old_value = trip.get(field)
98        if old_value != value:
99            changes[field] = {"old": old_value, "new": value}
100            trip[field] = value
101    
102    if changes:
103        change_logs.append({
104            "trip_id": trip_id,
105            "timestamp": time.time(),
106            "changes": changes,
107            "user_id": user_id
108        })
109    
110    return trip
111 
112@app.get("/trips/{trip_id}/change-logs")
113def get_trip_change_logs(trip_id: int, authorization: str = Header(None)):
114    user_id = get_current_user(authorization)
115    trip = trips.get(trip_id)
116    if not trip:
117        raise HTTPException(status_code=404, detail="Trip not found")
118    if trip["user_id"] != user_id:
119        raise HTTPException(status_code=403, detail="Not your trip")
120    return [log for log in change_logs if log["trip_id"] == trip_id]

1fastapi
2uvicorn
3pydantic