Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, Dict
4import secrets
5 
6app = FastAPI()
7 
8# In-memory stores
9users: Dict[int, dict] = {}
10subscribers: Dict[int, dict] = {}
11tokens: Dict[str, int] = {}  # token -> user_id
12next_user_id = 1
13next_subscriber_id = 1
14 
15# Auth helper
16def get_user_id(authorization: str = Header(...)) -> int:
17    if not authorization.startswith("Bearer "):
18        raise HTTPException(status_code=401, detail="Invalid auth header")
19    token = authorization[7:]
20    user_id = tokens.get(token)
21    if user_id is None:
22        raise HTTPException(status_code=401, detail="Invalid token")
23    return user_id
24 
25# User models
26class SignupRequest(BaseModel):
27    username: str
28    password: str
29 
30class LoginRequest(BaseModel):
31    username: str
32    password: str
33 
34# Endpoints
35@app.post("/signup")
36def signup(req: SignupRequest):
37    global next_user_id
38    user_id = next_user_id
39    next_user_id += 1
40    users[user_id] = {"username": req.username, "password": req.password}
41    return {"id": user_id, "username": req.username}
42 
43@app.post("/login")
44def login(req: LoginRequest):
45    for uid, u in users.items():
46        if u["username"] == req.username and u["password"] == req.password:
47            token = secrets.token_hex(16)
48            tokens[token] = uid
49            return {"token": token}
50    raise HTTPException(status_code=401, detail="Invalid credentials")
51 
52# Subscriber models
53class SubscriberCreate(BaseModel):
54    name: str
55    email: str
56    tier: Optional[str] = "free"
57    is_active: Optional[bool] = True
58 
59class SubscriberUpdate(BaseModel):
60    name: Optional[str] = None
61    email: Optional[str] = None
62    tier: Optional[str] = None
63    is_active: Optional[bool] = None
64 
65@app.post("/subscribers")
66def create_subscriber(req: SubscriberCreate, authorization: str = Header(...)):
67    get_user_id(authorization)
68    global next_subscriber_id
69    sub_id = next_subscriber_id
70    next_subscriber_id += 1
71    subscribers[sub_id] = {
72        "id": sub_id,
73        "name": req.name,
74        "email": req.email,
75        "tier": req.tier,
76        "is_active": req.is_active
77    }
78    return subscribers[sub_id]
79 
80@app.get("/subscribers/{sub_id}")
81def get_subscriber(sub_id: int, authorization: str = Header(...)):
82    get_user_id(authorization)
83    sub = subscribers.get(sub_id)
84    if not sub:
85        raise HTTPException(status_code=404, detail="Subscriber not found")
86    return sub
87 
88@app.patch("/subscribers/{sub_id}")
89def update_subscriber(sub_id: int, req: SubscriberUpdate, authorization: str = Header(...)):
90    get_user_id(authorization)
91    sub = subscribers.get(sub_id)
92    if not sub:
93        raise HTTPException(status_code=404, detail="Subscriber not found")
94    if req.name is not None:
95        sub["name"] = req.name
96    if req.email is not None:
97        sub["email"] = req.email
98    if req.tier is not None:
99        sub["tier"] = req.tier
100    if req.is_active is not None:
101        sub["is_active"] = req.is_active
102    return sub

1fastapi
2uvicorn