Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5import time
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11reports = {}
12report_id_counter = 1
13 
14class SignupRequest(BaseModel):
15    username: str
16    password: str
17 
18class LoginRequest(BaseModel):
19    username: str
20    password: str
21 
22class ReportCreate(BaseModel):
23    title: str
24    description: str
25    location: str
26    category: str
27 
28class ReportUpdate(BaseModel):
29    status: Optional[str] = None
30    title: Optional[str] = None
31    description: Optional[str] = None
32    location: Optional[str] = None
33    category: Optional[str] = None
34 
35def get_current_user(authorization: str = Header(None)):
36    if not authorization:
37        raise HTTPException(status_code=401, detail="No auth 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    if req.username in users:
46        raise HTTPException(status_code=400, detail="User exists")
47    users[req.username] = {"password": req.password}
48    return {"msg": "ok"}
49 
50@app.post("/login")
51def login(req: LoginRequest):
52    user = users.get(req.username)
53    if not user or user["password"] != req.password:
54        raise HTTPException(status_code=401, detail="Bad credentials")
55    token = secrets.token_hex(16)
56    tokens[token] = req.username
57    return {"token": token}
58 
59@app.get("/reports")
60def get_reports(authorization: str = Header(None)):
61    user = get_current_user(authorization)
62    open_reports = []
63    for rid, r in reports.items():
64        if r["status"] == "open":
65            open_reports.append({
66                "id": rid,
67                "location": r["location"],
68                "category": r["category"],
69                "status": r["status"],
70                "title": r["title"],
71                "description": r["description"],
72                "reported_by": r["reported_by"]
73            })
74    return open_reports
75 
76@app.post("/reports")
77def create_report(req: ReportCreate, authorization: str = Header(None)):
78    user = get_current_user(authorization)
79    global report_id_counter
80    new_id = report_id_counter
81    report_id_counter += 1
82    reports[new_id] = {
83        "title": req.title,
84        "description": req.description,
85        "location": req.location,
86        "category": req.category,
87        "status": "open",
88        "reported_by": user,
89        "created_at": int(time.time())
90    }
91    return {"id": new_id, "status": "open"}
92 
93@app.get("/reports/{report_id}")
94def get_report(report_id: int, authorization: str = Header(None)):
95    user = get_current_user(authorization)
96    r = reports.get(report_id)
97    if not r:
98        raise HTTPException(status_code=404, detail="Report not found")
99    return {"id": report_id, **r}
100 
101@app.put("/reports/{report_id}")
102def update_report(report_id: int, req: ReportUpdate, authorization: str = Header(None)):
103    user = get_current_user(authorization)
104    r = reports.get(report_id)
105    if not r:
106        raise HTTPException(status_code=404, detail="Report not found")
107    if req.status is not None:
108        r["status"] = req.status
109    if req.title is not None:
110        r["title"] = req.title
111    if req.description is not None:
112        r["description"] = req.description
113    if req.location is not None:
114        r["location"] = req.location
115    if req.category is not None:
116        r["category"] = req.category
117    return {"id": report_id, **r}

1fastapi
2uvicorn