Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import hashlib
5import random
6import string
7 
8app = FastAPI()
9 
10users = {}
11tokens = {}
12creators = {}
13next_user_id = 1
14next_creator_id = 1
15 
16def generate_token():
17    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
18 
19def get_current_user(authorization: str = Header(...)):
20    if not authorization.startswith("Bearer "):
21        raise HTTPException(status_code=401, detail="Invalid auth header")
22    token = authorization[7:]
23    if token not in tokens:
24        raise HTTPException(status_code=401, detail="Invalid token")
25    return tokens[token]
26 
27class SignupRequest(BaseModel):
28    username: str
29    password: str
30 
31class LoginRequest(BaseModel):
32    username: str
33    password: str
34 
35class CreatorCreate(BaseModel):
36    name: str
37    channel_description: str = ""
38    monetization_enabled: bool = False
39    content_tier: str = "free"
40 
41class CreatorUpdate(BaseModel):
42    name: Optional[str] = None
43    channel_description: Optional[str] = None
44    monetization_enabled: Optional[bool] = None
45    content_tier: Optional[str] = None
46 
47@app.post("/signup")
48def signup(req: SignupRequest):
49    global next_user_id
50    if any(u["username"] == req.username for u in users.values()):
51        raise HTTPException(status_code=400, detail="Username taken")
52    user_id = next_user_id
53    next_user_id += 1
54    users[user_id] = {
55        "id": user_id,
56        "username": req.username,
57        "password": hashlib.sha256(req.password.encode()).hexdigest()
58    }
59    return {"id": user_id, "username": req.username}
60 
61@app.post("/login")
62def login(req: LoginRequest):
63    for uid, u in users.items():
64        if u["username"] == req.username and u["password"] == hashlib.sha256(req.password.encode()).hexdigest():
65            token = generate_token()
66            tokens[token] = uid
67            return {"token": token}
68    raise HTTPException(status_code=401, detail="Invalid credentials")
69 
70@app.post("/creators")
71def create_creator(creator: CreatorCreate, authorization: str = Header(...)):
72    global next_creator_id
73    user_id = get_current_user(authorization)
74    creator_id = next_creator_id
75    next_creator_id += 1
76    creators[creator_id] = {
77        "id": creator_id,
78        "user_id": user_id,
79        "name": creator.name,
80        "channel_description": creator.channel_description,
81        "monetization_enabled": creator.monetization_enabled,
82        "content_tier": creator.content_tier
83    }
84    return creators[creator_id]
85 
86@app.get("/creators/{creator_id}")
87def get_creator(creator_id: int, authorization: str = Header(...)):
88    user_id = get_current_user(authorization)
89    if creator_id not in creators:
90        raise HTTPException(status_code=404, detail="Creator not found")
91    return creators[creator_id]
92 
93@app.patch("/creators/{creator_id}")
94def update_creator(creator_id: int, update: CreatorUpdate, authorization: str = Header(...)):
95    user_id = get_current_user(authorization)
96    if creator_id not in creators:
97        raise HTTPException(status_code=404, detail="Creator not found")
98    creator = creators[creator_id]
99    if creator["user_id"] != user_id:
100        raise HTTPException(status_code=403, detail="Not your creator")
101    if update.name is not None:
102        creator["name"] = update.name
103    if update.channel_description is not None:
104        creator["channel_description"] = update.channel_description
105    if update.monetization_enabled is not None:
106        creator["monetization_enabled"] = update.monetization_enabled
107    if update.content_tier is not None:
108        creator["content_tier"] = update.content_tier
109    return creator

1fastapi
2uvicorn