Browse the data

1from fastapi import FastAPI, HTTPException, Header, status
2from pydantic import BaseModel
3from typing import Optional, List
4import secrets
5from datetime import datetime, timezone
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11alerts = {}
12alert_id_counter = 0
13alert_history = {}
14 
15class SignupRequest(BaseModel):
16    username: str
17    password: str
18 
19class LoginRequest(BaseModel):
20    username: str
21    password: str
22 
23class AlertCreate(BaseModel):
24    location: str
25    severity: str
26    message: str
27    expiration: str
28    channels: List[str] = ["email"]
29 
30class AlertUpdate(BaseModel):
31    location: Optional[str] = None
32    severity: Optional[str] = None
33    message: Optional[str] = None
34    expiration: Optional[str] = None
35    channels: Optional[List[str]] = None
36 
37def get_current_user(authorization: str = Header(None)):
38    if not authorization:
39        raise HTTPException(status_code=401, detail="Missing auth header")
40    token = authorization.replace("Bearer ", "")
41    if token not in tokens:
42        raise HTTPException(status_code=401, detail="Invalid token")
43    return tokens[token]
44 
45@app.post("/signup")
46def signup(req: SignupRequest):
47    if req.username in users:
48        raise HTTPException(status_code=400, detail="User exists")
49    users[req.username] = req.password
50    return {"message": "User created"}
51 
52@app.post("/login")
53def login(req: LoginRequest):
54    if users.get(req.username) != req.password:
55        raise HTTPException(status_code=401, detail="Invalid credentials")
56    token = secrets.token_hex(16)
57    tokens[token] = req.username
58    return {"token": token}
59 
60@app.post("/alerts")
61def create_alert(req: AlertCreate, authorization: str = Header(None)):
62    user = get_current_user(authorization)
63    global alert_id_counter
64    alert_id_counter += 1
65    now = datetime.now(timezone.utc).isoformat()
66    alert = {
67        "id": alert_id_counter,
68        "location": req.location,
69        "severity": req.severity,
70        "message": req.message,
71        "expiration": req.expiration,
72        "channels": req.channels,
73        "created_by": user,
74        "created_at": now,
75        "updated_at": now
76    }
77    alerts[alert_id_counter] = alert
78    alert_history[alert_id_counter] = [alert.copy()]
79    return alert
80 
81@app.get("/alerts/{alert_id}")
82def get_alert(alert_id: int, authorization: str = Header(None)):
83    user = get_current_user(authorization)
84    if alert_id not in alerts:
85        raise HTTPException(status_code=404, detail="Alert not found")
86    return alerts[alert_id]
87 
88@app.patch("/alerts/{alert_id}")
89def update_alert(alert_id: int, req: AlertUpdate, authorization: str = Header(None)):
90    user = get_current_user(authorization)
91    if alert_id not in alerts:
92        raise HTTPException(status_code=404, detail="Alert not found")
93    alert = alerts[alert_id]
94    if req.location is not None:
95        alert["location"] = req.location
96    if req.severity is not None:
97        alert["severity"] = req.severity
98    if req.message is not None:
99        alert["message"] = req.message
100    if req.expiration is not None:
101        alert["expiration"] = req.expiration
102    if req.channels is not None:
103        alert["channels"] = req.channels
104    alert["updated_at"] = datetime.now(timezone.utc).isoformat()
105    alert_history[alert_id].append(alert.copy())
106    return alert
107 
108@app.get("/alerts/{alert_id}/history")
109def get_alert_history(alert_id: int, authorization: str = Header(None)):
110    user = get_current_user(authorization)
111    if alert_id not in alert_history:
112        raise HTTPException(status_code=404, detail="Alert not found")
113    return alert_history[alert_id]

1fastapi
2uvicorn