Browse the data

1import httpx
2import asyncio
3from fastapi import FastAPI, HTTPException, Header
4from pydantic import BaseModel
5from typing import Optional
6import secrets
7 
8app = FastAPI()
9 
10users = {}
11tokens = {}
12next_user_id = 1
13 
14schedules = {}
15next_schedule_id = 1
16 
17class SignupRequest(BaseModel):
18    username: str
19    password: str
20 
21class LoginRequest(BaseModel):
22    username: str
23    password: str
24 
25class ScheduleCreate(BaseModel):
26    name: str
27    url: str
28 
29class ItineraryRequest(BaseModel):
30    schedule_id: int
31    origin_stop: str
32    destination_stop: str
33    departure_time: str
34 
35def get_current_user(authorization: Optional[str] = Header(None)):
36    if not authorization:
37        raise HTTPException(status_code=401, detail="Missing authorization 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 already exists")
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("/schedule")
64def create_schedule(req: ScheduleCreate, authorization: Optional[str] = Header(None)):
65    get_current_user(authorization)
66    global next_schedule_id
67    sid = next_schedule_id
68    next_schedule_id += 1
69    schedules[sid] = {"id": sid, "name": req.name, "url": req.url}
70    return schedules[sid]
71 
72@app.get("/schedule/{schedule_id}")
73def get_schedule(schedule_id: int, authorization: Optional[str] = Header(None)):
74    get_current_user(authorization)
75    if schedule_id not in schedules:
76        raise HTTPException(status_code=404, detail="Schedule not found")
77    return schedules[schedule_id]
78 
79@app.post("/itinerary/validate")
80def validate_itinerary(req: ItineraryRequest, authorization: Optional[str] = Header(None)):
81    get_current_user(authorization)
82    if req.schedule_id not in schedules:
83        raise HTTPException(status_code=404, detail="Schedule not found")
84    url = schedules[req.schedule_id]["url"]
85    try:
86        resp = httpx.get(url, timeout=10)
87        resp.raise_for_status()
88        data = resp.json()
89    except Exception as e:
90        raise HTTPException(status_code=400, detail=f"Failed to fetch schedule: {str(e)}")
91 
92    stops = data.get("stops", [])
93    trips = data.get("trips", [])
94 
95    origin_found = False
96    dest_found = False
97    for stop in stops:
98        if stop["name"] == req.origin_stop:
99            origin_found = True
100        if stop["name"] == req.destination_stop:
101            dest_found = True
102 
103    if not origin_found:
104        return {"feasible": False, "reason": "Origin stop not found"}
105    if not dest_found:
106        return {"feasible": False, "reason": "Destination stop not found"}
107 
108    feasible_trip = None
109    for trip in trips:
110        if trip["origin"] == req.origin_stop and trip["destination"] == req.destination_stop:
111            if trip["departure"] >= req.departure_time:
112                if feasible_trip is None or trip["departure"] < feasible_trip["departure"]:
113                    feasible_trip = trip
114 
115    if feasible_trip:
116        return {
117            "feasible": True,
118            "departure": feasible_trip["departure"],
119            "arrival": feasible_trip["arrival"]
120        }
121    else:
122        return {"feasible": False, "reason": "No feasible trip found"}

1fastapi
2uvicorn
3httpx