Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List
4import secrets
5import time
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11zones = {}
12next_user_id = 1
13next_zone_id = 1
14 
15class SignupRequest(BaseModel):
16    username: str
17    password: str
18 
19class LoginRequest(BaseModel):
20    username: str
21    password: str
22 
23class ZoneCreate(BaseModel):
24    name: str
25    devices: List[str] = []
26    temperature_settings: dict = {}
27    user_control_permissions: List[str] = []
28 
29class ZoneUpdate(BaseModel):
30    name: Optional[str] = None
31    devices: Optional[List[str]] = None
32    temperature_settings: Optional[dict] = None
33    user_control_permissions: Optional[List[str]] = None
34 
35def get_current_user(authorization: str = Header(...)):
36    token = authorization.replace("Bearer ", "")
37    if token not in tokens:
38        raise HTTPException(status_code=401, detail="Invalid token")
39    user_id = tokens[token]
40    if time.time() > users[user_id]["token_expiry"]:
41        del tokens[token]
42        raise HTTPException(status_code=401, detail="Token expired")
43    return user_id
44 
45@app.post("/signup")
46def signup(req: SignupRequest):
47    global next_user_id
48    user_id = next_user_id
49    next_user_id += 1
50    users[user_id] = {
51        "username": req.username,
52        "password": req.password,
53        "token_expiry": 0
54    }
55    return {"user_id": user_id, "message": "User created"}
56 
57@app.post("/login")
58def login(req: LoginRequest):
59    for uid, u in users.items():
60        if u["username"] == req.username and u["password"] == req.password:
61            token = secrets.token_hex(32)
62            tokens[token] = uid
63            users[uid]["token_expiry"] = time.time() + 86400
64            return {"token": token}
65    raise HTTPException(status_code=401, detail="Invalid credentials")
66 
67@app.get("/zones/{zone_id}")
68def get_zone(zone_id: int, authorization: str = Header(...)):
69    get_current_user(authorization)
70    if zone_id not in zones:
71        raise HTTPException(status_code=404, detail="Zone not found")
72    return zones[zone_id]
73 
74@app.post("/zones")
75def create_zone(zone: ZoneCreate, authorization: str = Header(...)):
76    get_current_user(authorization)
77    global next_zone_id
78    zone_id = next_zone_id
79    next_zone_id += 1
80    zones[zone_id] = {
81        "id": zone_id,
82        "name": zone.name,
83        "devices": zone.devices,
84        "temperature_settings": zone.temperature_settings,
85        "user_control_permissions": zone.user_control_permissions
86    }
87    return zones[zone_id]
88 
89@app.patch("/zones/{zone_id}")
90def update_zone(zone_id: int, update: ZoneUpdate, authorization: str = Header(...)):
91    get_current_user(authorization)
92    if zone_id not in zones:
93        raise HTTPException(status_code=404, detail="Zone not found")
94    zone = zones[zone_id]
95    if update.name is not None:
96        zone["name"] = update.name
97    if update.devices is not None:
98        zone["devices"] = update.devices
99    if update.temperature_settings is not None:
100        zone["temperature_settings"] = update.temperature_settings
101    if update.user_control_permissions is not None:
102        zone["user_control_permissions"] = update.user_control_permissions
103    return zone

1fastapi
2uvicorn