Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5import uvicorn
6 
7app = FastAPI()
8 
9users = {}
10user_id_counter = 1
11tokens = {}
12tracks = {}
13track_id_counter = 1
14labels = {}
15label_id_counter = 1
16 
17class SignupRequest(BaseModel):
18    username: str
19    password: str
20 
21class LoginRequest(BaseModel):
22    username: str
23    password: str
24 
25class TrackCreate(BaseModel):
26    title: str
27    artist: str
28    duration: float
29 
30class LabelCreate(BaseModel):
31    name: str
32 
33def get_current_user(authorization: Optional[str] = Header(None)):
34    if not authorization:
35        raise HTTPException(status_code=401, detail="No auth header")
36    token = authorization.replace("Bearer ", "")
37    if token not in tokens:
38        raise HTTPException(status_code=401, detail="Invalid token")
39    return tokens[token]
40 
41@app.post("/signup")
42def signup(req: SignupRequest):
43    global user_id_counter
44    for u in users.values():
45        if u["username"] == req.username:
46            raise HTTPException(status_code=400, detail="Username taken")
47    user_id = user_id_counter
48    user_id_counter += 1
49    users[user_id] = {"id": user_id, "username": req.username, "password": req.password, "role": "listener"}
50    if user_id == 1:
51        users[user_id]["role"] = "label_admin"
52    return {"user_id": user_id, "username": req.username, "role": users[user_id]["role"]}
53 
54@app.post("/login")
55def login(req: LoginRequest):
56    for uid, u in users.items():
57        if u["username"] == req.username and u["password"] == req.password:
58            token = secrets.token_hex(16)
59            tokens[token] = uid
60            return {"token": token, "user_id": uid}
61    raise HTTPException(status_code=401, detail="Invalid credentials")
62 
63@app.post("/promote/{user_id}")
64def promote_user(user_id: int, authorization: Optional[str] = Header(None)):
65    current_user_id = get_current_user(authorization)
66    if users[current_user_id]["role"] != "label_admin":
67        raise HTTPException(status_code=403, detail="Only label admin can promote")
68    if user_id not in users:
69        raise HTTPException(status_code=404, detail="User not found")
70    users[user_id]["role"] = "producer"
71    return {"user_id": user_id, "role": "producer"}
72 
73@app.post("/tracks")
74def create_track(track: TrackCreate, authorization: Optional[str] = Header(None)):
75    current_user_id = get_current_user(authorization)
76    global track_id_counter
77    track_id = track_id_counter
78    track_id_counter += 1
79    tracks[track_id] = {"id": track_id, "title": track.title, "artist": track.artist, "duration": track.duration, "owner_id": current_user_id}
80    return tracks[track_id]
81 
82@app.get("/tracks/{track_id}")
83def get_track(track_id: int):
84    if track_id not in tracks:
85        raise HTTPException(status_code=404, detail="Track not found")
86    return tracks[track_id]
87 
88@app.post("/tracks/merge")
89def merge_tracks(track_ids: list[int], authorization: Optional[str] = Header(None)):
90    current_user_id = get_current_user(authorization)
91    if users[current_user_id]["role"] not in ("producer", "label_admin"):
92        raise HTTPException(status_code=403, detail="Only producers or admin can merge")
93    merged_title = "Merged Track"
94    merged_duration = 0.0
95    for tid in track_ids:
96        if tid not in tracks:
97            raise HTTPException(status_code=404, detail=f"Track {tid} not found")
98        merged_duration += tracks[tid]["duration"]
99    global track_id_counter
100    new_id = track_id_counter
101    track_id_counter += 1
102    tracks[new_id] = {"id": new_id, "title": merged_title, "artist": "Merged", "duration": merged_duration, "owner_id": current_user_id, "merged_from": track_ids}
103    return tracks[new_id]
104 
105@app.post("/tracks/bulk")
106def bulk_upload(tracks_data: list[TrackCreate], authorization: Optional[str] = Header(None)):
107    current_user_id = get_current_user(authorization)
108    if users[current_user_id]["role"] not in ("producer", "label_admin"):
109        raise HTTPException(status_code=403, detail="Only producers or admin can bulk upload")
110    created = []
111    global track_id_counter
112    for t in tracks_data:
113        tid = track_id_counter
114        track_id_counter += 1
115        tracks[tid] = {"id": tid, "title": t.title, "artist": t.artist, "duration": t.duration, "owner_id": current_user_id}
116        created.append(tracks[tid])
117    return created
118 
119@app.get("/users/{user_id}")
120def get_user(user_id: int):
121    if user_id not in users:
122        raise HTTPException(status_code=404, detail="User not found")
123    return {"id": users[user_id]["id"], "username": users[user_id]["username"], "role": users[user_id]["role"]}
124 
125@app.post("/labels")
126def create_label(label: LabelCreate, authorization: Optional[str] = Header(None)):
127    current_user_id = get_current_user(authorization)
128    global label_id_counter
129    label_id = label_id_counter
130    label_id_counter += 1
131    labels[label_id] = {"id": label_id, "name": label.name, "owner_id": current_user_id}
132    return labels[label_id]
133 
134@app.get("/labels/{label_id}")
135def get_label(label_id: int):
136    if label_id not in labels:
137        raise HTTPException(status_code=404, detail="Label not found")
138    return labels[label_id]

1fastapi
2uvicorn