Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import time
5import random
6import string
7 
8app = FastAPI()
9 
10users = {}
11tokens = {}
12reports = {}
13report_counter = 0
14 
15def generate_token():
16    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
17 
18def get_current_user(authorization: Optional[str] = Header(None)):
19    if not authorization:
20        raise HTTPException(status_code=401, detail="Missing authorization header")
21    token = authorization.replace("Bearer ", "")
22    user_id = tokens.get(token)
23    if not user_id:
24        raise HTTPException(status_code=401, detail="Invalid token")
25    return user_id
26 
27class SignupRequest(BaseModel):
28    username: str
29    password: str
30 
31class LoginRequest(BaseModel):
32    username: str
33    password: str
34 
35class ReportCreate(BaseModel):
36    lat: float
37    lng: float
38    issue_type: str  # "pothole" or "broken_light"
39    description: Optional[str] = ""
40 
41@app.post("/signup")
42def signup(req: SignupRequest):
43    if req.username in users:
44        raise HTTPException(status_code=400, detail="User already exists")
45    users[req.username] = req.password
46    token = generate_token()
47    tokens[token] = req.username
48    return {"token": token}
49 
50@app.post("/login")
51def login(req: LoginRequest):
52    if users.get(req.username) != req.password:
53        raise HTTPException(status_code=401, detail="Invalid credentials")
54    token = generate_token()
55    tokens[token] = req.username
56    return {"token": token}
57 
58@app.post("/reports")
59def create_report(report: ReportCreate, authorization: Optional[str] = Header(None)):
60    user = get_current_user(authorization)
61    global report_counter
62    report_counter += 1
63    now = int(time.time())
64    reports[report_counter] = {
65        "id": report_counter,
66        "lat": report.lat,
67        "lng": report.lng,
68        "issue_type": report.issue_type,
69        "description": report.description,
70        "reported_by": user,
71        "created_at": now,
72        "status": "open",
73        "escalated": False
74    }
75    return reports[report_counter]
76 
77@app.get("/reports/{report_id}")
78def get_report(report_id: int, authorization: Optional[str] = Header(None)):
79    get_current_user(authorization)
80    report = reports.get(report_id)
81    if not report:
82        raise HTTPException(status_code=404, detail="Report not found")
83    # Check escalation
84    now = int(time.time())
85    if report["status"] == "open" and (now - report["created_at"]) > 7 * 24 * 3600:
86        if not report["escalated"]:
87            report["escalated"] = True
88            report["status"] = "escalated"
89    return report
90 
91@app.get("/reports")
92def list_reports(authorization: Optional[str] = Header(None)):
93    get_current_user(authorization)
94    now = int(time.time())
95    for r in reports.values():
96        if r["status"] == "open" and (now - r["created_at"]) > 7 * 24 * 3600:
97            if not r["escalated"]:
98                r["escalated"] = True
99                r["status"] = "escalated"
100    return list(reports.values())

1fastapi
2uvicorn