Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from datetime import datetime
3from typing import Optional
4from pydantic import BaseModel
5import secrets
6 
7app = FastAPI()
8 
9subscribers = {}
10next_id = 1
11tokens = {}
12 
13class SignupRequest(BaseModel):
14    email: str
15    name: str
16 
17class LoginRequest(BaseModel):
18    email: str
19    password: str
20 
21users = {}
22user_next_id = 1
23 
24@app.post("/signup")
25def signup(req: SignupRequest):
26    global user_next_id
27    for u in users.values():
28        if u["email"] == req.email:
29            raise HTTPException(400, "Email already exists")
30    user_id = user_next_id
31    user_next_id += 1
32    token = secrets.token_hex(16)
33    users[user_id] = {"id": user_id, "email": req.email, "name": req.name, "password": req.email}
34    tokens[token] = user_id
35    return {"user_id": user_id, "token": token}
36 
37@app.post("/login")
38def login(req: LoginRequest):
39    for u in users.values():
40        if u["email"] == req.email and u["password"] == req.password:
41            token = secrets.token_hex(16)
42            tokens[token] = u["id"]
43            return {"token": token}
44    raise HTTPException(401, "Invalid credentials")
45 
46def get_current_user(authorization: Optional[str] = Header(None)):
47    if not authorization:
48        raise HTTPException(401, "Missing auth header")
49    token = authorization.replace("Bearer ", "")
50    if token not in tokens:
51        raise HTTPException(401, "Invalid token")
52    return tokens[token]
53 
54@app.post("/subscribers")
55def create_subscriber(req: SignupRequest, authorization: Optional[str] = Header(None)):
56    get_current_user(authorization)
57    global next_id
58    sub_id = next_id
59    next_id += 1
60    subscribers[sub_id] = {
61        "id": sub_id,
62        "email": req.email,
63        "name": req.name,
64        "subscribed_at": datetime.utcnow().isoformat()
65    }
66    return subscribers[sub_id]
67 
68@app.get("/subscribers")
69def list_subscribers(authorization: Optional[str] = Header(None)):
70    get_current_user(authorization)
71    return list(subscribers.values())
72 
73@app.get("/subscribers/{sub_id}")
74def get_subscriber(sub_id: int, authorization: Optional[str] = Header(None)):
75    get_current_user(authorization)
76    if sub_id not in subscribers:
77        raise HTTPException(404, "Subscriber not found")
78    return subscribers[sub_id]
79 
80@app.delete("/subscribers/{sub_id}")
81def delete_subscriber(sub_id: int, authorization: Optional[str] = Header(None)):
82    get_current_user(authorization)
83    if sub_id not in subscribers:
84        raise HTTPException(404, "Subscriber not found")
85    del subscribers[sub_id]
86    return {"ok": True}
87 
88@app.get("/users/{user_id}")
89def get_user(user_id: int, authorization: Optional[str] = Header(None)):
90    get_current_user(authorization)
91    if user_id not in users:
92        raise HTTPException(404, "User not found")
93    return users[user_id]
94 
95@app.post("/users")
96def create_user(req: SignupRequest, authorization: Optional[str] = Header(None)):
97    get_current_user(authorization)
98    global user_next_id
99    for u in users.values():
100        if u["email"] == req.email:
101            raise HTTPException(400, "Email already exists")
102    user_id = user_next_id
103    user_next_id += 1
104    users[user_id] = {"id": user_id, "email": req.email, "name": req.name}
105    return users[user_id]

1fastapi
2uvicorn