Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5import datetime
6 
7app = FastAPI()
8 
9users = {}
10sensors = {}
11readings = {}
12alerts = {}
13tokens = {}
14ids = {"users": 0, "sensors": 0, "readings": 0, "alerts": 0}
15 
16class SignupRequest(BaseModel):
17    username: str
18    password: str
19 
20class LoginRequest(BaseModel):
21    username: str
22    password: str
23 
24class SensorCreate(BaseModel):
25    name: str
26    location: str
27 
28class ReadingCreate(BaseModel):
29    sensor_id: int
30    value: float
31    unit: str = "celsius"
32 
33class AlertCreate(BaseModel):
34    sensor_id: int
35    min_value: Optional[float] = None
36    max_value: Optional[float] = None
37 
38def get_current_user(authorization: str = Header(...)):
39    token = authorization.replace("Bearer ", "")
40    if token not in tokens:
41        raise HTTPException(status_code=401, detail="Invalid token")
42    return tokens[token]
43 
44@app.post("/signup")
45def signup(req: SignupRequest):
46    if req.username in users:
47        raise HTTPException(status_code=400, detail="User exists")
48    users[req.username] = {"username": req.username, "password": req.password}
49    return {"message": "User created"}
50 
51@app.post("/login")
52def login(req: LoginRequest):
53    user = users.get(req.username)
54    if not user or user["password"] != 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("/sensors")
61def create_sensor(sensor: SensorCreate, authorization: str = Header(...)):
62    user = get_current_user(authorization)
63    ids["sensors"] += 1
64    sensor_id = ids["sensors"]
65    sensors[sensor_id] = {"id": sensor_id, "name": sensor.name, "location": sensor.location, "owner": user}
66    return sensors[sensor_id]
67 
68@app.get("/sensors/{sensor_id}")
69def get_sensor(sensor_id: int, authorization: str = Header(...)):
70    get_current_user(authorization)
71    sensor = sensors.get(sensor_id)
72    if not sensor:
73        raise HTTPException(status_code=404, detail="Sensor not found")
74    return sensor
75 
76@app.post("/readings")
77def create_reading(reading: ReadingCreate, authorization: str = Header(...)):
78    get_current_user(authorization)
79    if reading.sensor_id not in sensors:
80        raise HTTPException(status_code=404, detail="Sensor not found")
81    ids["readings"] += 1
82    reading_id = ids["readings"]
83    readings[reading_id] = {
84        "id": reading_id,
85        "sensor_id": reading.sensor_id,
86        "value": reading.value,
87        "unit": reading.unit,
88        "timestamp": datetime.datetime.utcnow().isoformat()
89    }
90    # Check alerts
91    for alert_id, alert in alerts.items():
92        if alert["sensor_id"] == reading.sensor_id:
93            if alert["min_value"] is not None and reading.value < alert["min_value"]:
94                pass  # trigger
95            if alert["max_value"] is not None and reading.value > alert["max_value"]:
96                pass  # trigger
97    return readings[reading_id]
98 
99@app.get("/readings/{reading_id}")
100def get_reading(reading_id: int, authorization: str = Header(...)):
101    get_current_user(authorization)
102    reading = readings.get(reading_id)
103    if not reading:
104        raise HTTPException(status_code=404, detail="Reading not found")
105    return reading
106 
107@app.get("/readings/history/{sensor_id}")
108def get_reading_history(sensor_id: int, authorization: str = Header(...)):
109    get_current_user(authorization)
110    if sensor_id not in sensors:
111        raise HTTPException(status_code=404, detail="Sensor not found")
112    sensor_readings = [r for r in readings.values() if r["sensor_id"] == sensor_id]
113    return sensor_readings
114 
115@app.post("/alerts")
116def create_alert(alert: AlertCreate, authorization: str = Header(...)):
117    get_current_user(authorization)
118    if alert.sensor_id not in sensors:
119        raise HTTPException(status_code=404, detail="Sensor not found")
120    ids["alerts"] += 1
121    alert_id = ids["alerts"]
122    alerts[alert_id] = {"id": alert_id, "sensor_id": alert.sensor_id, "min_value": alert.min_value, "max_value": alert.max_value}
123    return alerts[alert_id]
124 
125@app.get("/alerts/{alert_id}")
126def get_alert(alert_id: int, authorization: str = Header(...)):
127    get_current_user(authorization)
128    alert = alerts.get(alert_id)
129    if not alert:
130        raise HTTPException(status_code=404, detail="Alert not found")
131    return alert

1fastapi
2uvicorn