Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from typing import Optional
3import hashlib
4import secrets
5 
6app = FastAPI()
7 
8users = {}
9tokens = {}
10agencies = {}
11incidents = {}
12routes = {}
13next_user_id = 1
14next_agency_id = 1
15next_incident_id = 1
16next_route_id = 1
17 
18def hash_password(password: str) -> str:
19    return hashlib.sha256(password.encode()).hexdigest()
20 
21def get_current_user(authorization: Optional[str] = Header(None)):
22    if not authorization:
23        raise HTTPException(status_code=401, detail="Missing Authorization header")
24    token = authorization.replace("Bearer ", "")
25    user_id = tokens.get(token)
26    if not user_id:
27        raise HTTPException(status_code=401, detail="Invalid token")
28    return user_id
29 
30@app.post("/signup")
31def signup(username: str, password: str):
32    global next_user_id
33    for u in users.values():
34        if u["username"] == username:
35            raise HTTPException(status_code=400, detail="Username already exists")
36    user_id = next_user_id
37    next_user_id += 1
38    users[user_id] = {"id": user_id, "username": username, "password_hash": hash_password(password)}
39    return {"user_id": user_id, "username": username}
40 
41@app.post("/login")
42def login(username: str, password: str):
43    for u in users.values():
44        if u["username"] == username and u["password_hash"] == hash_password(password):
45            token = secrets.token_hex(16)
46            tokens[token] = u["id"]
47            return {"token": token}
48    raise HTTPException(status_code=401, detail="Invalid credentials")
49 
50@app.post("/agency")
51def create_agency(name: str, authorization: Optional[str] = Header(None)):
52    get_current_user(authorization)
53    global next_agency_id
54    agency_id = next_agency_id
55    next_agency_id += 1
56    agencies[agency_id] = {"id": agency_id, "name": name}
57    return agencies[agency_id]
58 
59@app.get("/agency/{agency_id}")
60def get_agency(agency_id: int, authorization: Optional[str] = Header(None)):
61    get_current_user(authorization)
62    agency = agencies.get(agency_id)
63    if not agency:
64        raise HTTPException(status_code=404, detail="Agency not found")
65    return agency
66 
67@app.post("/incident")
68def create_incident(route_id: int, description: str, delay_minutes: int, authorization: Optional[str] = Header(None)):
69    get_current_user(authorization)
70    global next_incident_id
71    incident_id = next_incident_id
72    next_incident_id += 1
73    incidents[incident_id] = {
74        "id": incident_id,
75        "route_id": route_id,
76        "description": description,
77        "delay_minutes": delay_minutes
78    }
79    return incidents[incident_id]
80 
81@app.get("/incident/{incident_id}")
82def get_incident(incident_id: int, authorization: Optional[str] = Header(None)):
83    get_current_user(authorization)
84    incident = incidents.get(incident_id)
85    if not incident:
86        raise HTTPException(status_code=404, detail="Incident not found")
87    return incident
88 
89@app.post("/route")
90def create_route(agency_id: int, name: str, authorization: Optional[str] = Header(None)):
91    get_current_user(authorization)
92    global next_route_id
93    route_id = next_route_id
94    next_route_id += 1
95    routes[route_id] = {"id": route_id, "agency_id": agency_id, "name": name}
96    return routes[route_id]
97 
98@app.get("/route/{route_id}")
99def get_route(route_id: int, authorization: Optional[str] = Header(None)):
100    get_current_user(authorization)
101    route = routes.get(route_id)
102    if not route:
103        raise HTTPException(status_code=404, detail="Route not found")
104    return route
105 
106@app.get("/route/{route_id}/delays")
107def get_route_delays(route_id: int, authorization: Optional[str] = Header(None)):
108    get_current_user(authorization)
109    route = routes.get(route_id)
110    if not route:
111        raise HTTPException(status_code=404, detail="Route not found")
112    route_incidents = [inc for inc in incidents.values() if inc["route_id"] == route_id]
113    total_delay = sum(inc["delay_minutes"] for inc in route_incidents)
114    return {"route_id": route_id, "total_delay_minutes": total_delay, "incidents": route_incidents}
115 
116@app.get("/route/{route_id}/alternatives")
117def get_route_alternatives(route_id: int, authorization: Optional[str] = Header(None)):
118    get_current_user(authorization)
119    route = routes.get(route_id)
120    if not route:
121        raise HTTPException(status_code=404, detail="Route not found")
122    same_agency_routes = [r for r in routes.values() if r["agency_id"] == route["agency_id"] and r["id"] != route_id]
123    if not same_agency_routes:
124        return {"route_id": route_id, "alternatives": []}
125    alternatives = []
126    for alt_route in same_agency_routes:
127        alt_incidents = [inc for inc in incidents.values() if inc["route_id"] == alt_route["id"]]
128        total_delay = sum(inc["delay_minutes"] for inc in alt_incidents)
129        alternatives.append({
130            "route_id": alt_route["id"],
131            "name": alt_route["name"],
132            "total_delay_minutes": total_delay
133        })
134    alternatives.sort(key=lambda x: x["total_delay_minutes"])
135    return {"route_id": route_id, "alternatives": alternatives}

1fastapi
2uvicorn