Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3import secrets
4from typing import Optional
5 
6app = FastAPI()
7 
8users = {}
9tokens = {}
10photos = {}
11albums = {}
12photo_id_counter = 1
13album_id_counter = 1
14user_id_counter = 1
15 
16promotions = {}
17 
18class SignupRequest(BaseModel):
19    username: str
20    password: str
21 
22class LoginRequest(BaseModel):
23    username: str
24    password: str
25 
26class PhotoCreate(BaseModel):
27    url: str
28    album_id: int
29 
30class AlbumCreate(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="Missing auth header")
36    token = authorization.replace("Bearer ", "")
37    user_id = tokens.get(token)
38    if not user_id:
39        raise HTTPException(status_code=401, detail="Invalid token")
40    return user_id
41 
42@app.post("/signup")
43def signup(req: SignupRequest):
44    global user_id_counter
45    for u in users.values():
46        if u["username"] == req.username:
47            raise HTTPException(status_code=400, detail="Username taken")
48    user_id = user_id_counter
49    user_id_counter += 1
50    users[user_id] = {"id": user_id, "username": req.username, "password": req.password}
51    return {"user_id": user_id}
52 
53@app.post("/login")
54def login(req: LoginRequest):
55    for uid, u in users.items():
56        if u["username"] == req.username and u["password"] == req.password:
57            token = secrets.token_hex(16)
58            tokens[token] = uid
59            return {"token": token}
60    raise HTTPException(status_code=401, detail="Invalid credentials")
61 
62@app.get("/photos/{photo_id}")
63def get_photo(photo_id: int):
64    photo = photos.get(photo_id)
65    if not photo:
66        raise HTTPException(status_code=404, detail="Photo not found")
67    return photo
68 
69@app.post("/photos")
70def create_photo(photo: PhotoCreate, authorization: Optional[str] = Header(None)):
71    global photo_id_counter
72    user_id = get_current_user(authorization)
73    album = albums.get(photo.album_id)
74    if not album:
75        raise HTTPException(status_code=404, detail="Album not found")
76    if photo.album_id not in album.get("allowed_curators", {}) and album.get("owner") != user_id:
77        raise HTTPException(status_code=403, detail="Not authorized to add photos to this album")
78    pid = photo_id_counter
79    photo_id_counter += 1
80    photos[pid] = {"id": pid, "url": photo.url, "album_id": photo.album_id, "uploader": user_id, "approved": False}
81    return {"photo_id": pid}
82 
83@app.get("/albums/{album_id}")
84def get_album(album_id: int):
85    album = albums.get(album_id)
86    if not album:
87        raise HTTPException(status_code=404, detail="Album not found")
88    return album
89 
90@app.post("/albums")
91def create_album(album: AlbumCreate, authorization: Optional[str] = Header(None)):
92    global album_id_counter
93    user_id = get_current_user(authorization)
94    aid = album_id_counter
95    album_id_counter += 1
96    albums[aid] = {"id": aid, "name": album.name, "owner": user_id, "allowed_curators": {}}
97    return {"album_id": aid}
98 
99@app.post("/promote/{user_id}")
100def promote_user(user_id: int, authorization: Optional[str] = Header(None)):
101    admin_id = get_current_user(authorization)
102    # first user is admin
103    if admin_id != 1:
104        raise HTTPException(status_code=403, detail="Only the first user (admin) can promote")
105    if user_id not in users:
106        raise HTTPException(status_code=404, detail="User not found")
107    promotions[user_id] = {"role": "curator"}
108    return {"status": "promoted", "user_id": user_id}
109 
110@app.post("/curate/{photo_id}")
111def curate_photo(photo_id: int, approve: bool, authorization: Optional[str] = Header(None)):
112    user_id = get_current_user(authorization)
113    if user_id not in promotions:
114        raise HTTPException(status_code=403, detail="Not a curator")
115    photo = photos.get(photo_id)
116    if not photo:
117        raise HTTPException(status_code=404, detail="Photo not found")
118    album = albums.get(photo["album_id"])
119    if not album:
120        raise HTTPException(status_code=404, detail="Album not found")
121    if album["owner"] != user_id and user_id not in album.get("allowed_curators", {}):
122        raise HTTPException(status_code=403, detail="Not authorized to curate this album")
123    photo["approved"] = approve
124    return {"photo_id": photo_id, "approved": approve}
125 
126@app.post("/albums/{album_id}/curators/{curator_id}")
127def add_curator_to_album(album_id: int, curator_id: int, authorization: Optional[str] = Header(None)):
128    user_id = get_current_user(authorization)
129    album = albums.get(album_id)
130    if not album:
131        raise HTTPException(status_code=404, detail="Album not found")
132    if album["owner"] != user_id:
133        raise HTTPException(status_code=403, detail="Only album owner can add curators")
134    if curator_id not in users:
135        raise HTTPException(status_code=404, detail="Curator user not found")
136    album["allowed_curators"][curator_id] = True
137    return {"status": "curator added", "album_id": album_id, "curator_id": curator_id}

1fastapi
2uvicorn