Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5 
6app = FastAPI()
7 
8users = {}
9tokens = {}
10channels = {}
11channel_id_counter = 1
12user_id_counter = 1
13 
14class SignupRequest(BaseModel):
15    username: str
16    password: str
17 
18class LoginRequest(BaseModel):
19    username: str
20    password: str
21 
22class ChannelCreate(BaseModel):
23    name: str
24    description: str = ""
25    category: str = ""
26 
27class ChannelUpdate(BaseModel):
28    name: Optional[str] = None
29    description: Optional[str] = None
30    category: Optional[str] = None
31    monetization_enabled: Optional[bool] = None
32    subscriber_tier_required: Optional[str] = None
33 
34@app.post("/signup")
35def signup(req: SignupRequest):
36    global user_id_counter
37    for u in users.values():
38        if u["username"] == req.username:
39            raise HTTPException(400, "Username already exists")
40    user_id = user_id_counter
41    user_id_counter += 1
42    users[user_id] = {"id": user_id, "username": req.username, "password": req.password}
43    return {"id": user_id, "username": req.username}
44 
45@app.post("/login")
46def login(req: LoginRequest):
47    for u in users.values():
48        if u["username"] == req.username and u["password"] == req.password:
49            token = secrets.token_hex(16)
50            tokens[token] = u["id"]
51            return {"token": token}
52    raise HTTPException(401, "Invalid credentials")
53 
54def get_user_id(token: str):
55    if token not in tokens:
56        raise HTTPException(401, "Invalid token")
57    return tokens[token]
58 
59@app.get("/channels/{channel_id}")
60def get_channel(channel_id: int, authorization: str = Header(None)):
61    get_user_id(authorization)
62    if channel_id not in channels:
63        raise HTTPException(404, "Channel not found")
64    return channels[channel_id]
65 
66@app.post("/channels")
67def create_channel(channel: ChannelCreate, authorization: str = Header(None)):
68    global channel_id_counter
69    user_id = get_user_id(authorization)
70    cid = channel_id_counter
71    channel_id_counter += 1
72    channels[cid] = {
73        "id": cid,
74        "owner_id": user_id,
75        "name": channel.name,
76        "description": channel.description,
77        "category": channel.category,
78        "monetization_enabled": False,
79        "subscriber_tier_required": None
80    }
81    return channels[cid]
82 
83@app.put("/channels/{channel_id}")
84def update_channel(channel_id: int, update: ChannelUpdate, authorization: str = Header(None)):
85    user_id = get_user_id(authorization)
86    if channel_id not in channels:
87        raise HTTPException(404, "Channel not found")
88    ch = channels[channel_id]
89    if ch["owner_id"] != user_id:
90        raise HTTPException(403, "Not your channel")
91    if update.name is not None:
92        ch["name"] = update.name
93    if update.description is not None:
94        ch["description"] = update.description
95    if update.category is not None:
96        ch["category"] = update.category
97    if update.monetization_enabled is not None:
98        ch["monetization_enabled"] = update.monetization_enabled
99    if update.subscriber_tier_required is not None:
100        ch["subscriber_tier_required"] = update.subscriber_tier_required
101    return ch

1fastapi
2uvicorn