Browse the data

1import os
2import json
3import secrets
4import shutil
5from pathlib import Path
6from fastapi import FastAPI, HTTPException, Header, UploadFile, File
7from fastapi.responses import FileResponse, StreamingResponse
8from typing import Optional
9 
10app = FastAPI()
11 
12users = {}
13tokens = {}
14workouts = {}
15workout_id_counter = 1
16 
17WORKOUT_DIR = "/var/workouts"
18os.makedirs(WORKOUT_DIR, exist_ok=True)
19 
20def auth_required(authorization: Optional[str] = Header(None)):
21    if not authorization:
22        raise HTTPException(status_code=401, detail="Missing auth header")
23    token = authorization.replace("Bearer ", "")
24    if token not in tokens:
25        raise HTTPException(status_code=401, detail="Invalid token")
26    return tokens[token]
27 
28@app.post("/signup")
29def signup(username: str, password: str):
30    if username in users:
31        raise HTTPException(status_code=400, detail="User exists")
32    users[username] = {"username": username, "password": password}
33    return {"message": "User created"}
34 
35@app.post("/login")
36def login(username: str, password: str):
37    if username not in users or users[username]["password"] != password:
38        raise HTTPException(status_code=401, detail="Invalid credentials")
39    token = secrets.token_hex(16)
40    tokens[token] = username
41    return {"token": token}
42 
43@app.post("/workout")
44def create_workout(name: str, description: str = "", authorization: Optional[str] = Header(None)):
45    auth_required(authorization)
46    global workout_id_counter
47    wid = workout_id_counter
48    workout_id_counter += 1
49    workouts[wid] = {
50        "id": wid,
51        "name": name,
52        "description": description,
53        "clips": {}
54    }
55    workout_dir = Path(WORKOUT_DIR) / str(wid)
56    workout_dir.mkdir(parents=True, exist_ok=True)
57    return workouts[wid]
58 
59@app.post("/workout/{workout_id}/clips")
60def upload_clip(workout_id: int, file: UploadFile = File(...), authorization: Optional[str] = Header(None)):
61    auth_required(authorization)
62    if workout_id not in workouts:
63        raise HTTPException(status_code=404, detail="Workout not found")
64    workout_dir = Path(WORKOUT_DIR) / str(workout_id)
65    file_path = workout_dir / file.filename
66    with open(file_path, "wb") as f:
67        shutil.copyfileobj(file.file, f)
68    clip_id = len(workouts[workout_id]["clips"]) + 1
69    workouts[workout_id]["clips"][clip_id] = {
70        "id": clip_id,
71        "filename": file.filename,
72        "filepath": str(file_path),
73        "views": 0
74    }
75    return workouts[workout_id]["clips"][clip_id]
76 
77@app.get("/workout/{workout_id}/clips/{file}")
78def stream_clip(workout_id: int, file: str, authorization: Optional[str] = Header(None)):
79    auth_required(authorization)
80    if workout_id not in workouts:
81        raise HTTPException(status_code=404, detail="Workout not found")
82    workout_dir = Path(WORKOUT_DIR) / str(workout_id)
83    file_path = workout_dir / file
84    if not file_path.exists():
85        raise HTTPException(status_code=404, detail="File not found")
86    for clip in workouts[workout_id]["clips"].values():
87        if clip["filename"] == file:
88            clip["views"] += 1
89            break
90    def iterfile():
91        with open(file_path, "rb") as f:
92            yield from f
93    return StreamingResponse(iterfile(), media_type="video/mp4")
94 
95@app.get("/workout/{workout_id}")
96def get_workout(workout_id: int, authorization: Optional[str] = Header(None)):
97    auth_required(authorization)
98    if workout_id not in workouts:
99        raise HTTPException(status_code=404, detail="Workout not found")
100    return workouts[workout_id]
101 
102@app.get("/workout/{workout_id}/clips")
103def list_clips(workout_id: int, authorization: Optional[str] = Header(None)):
104    auth_required(authorization)
105    if workout_id not in workouts:
106        raise HTTPException(status_code=404, detail="Workout not found")
107    return list(workouts[workout_id]["clips"].values())
108 
109@app.get("/workout/{workout_id}/clips/{clip_id}/views")
110def get_clip_views(workout_id: int, clip_id: int, authorization: Optional[str] = Header(None)):
111    auth_required(authorization)
112    if workout_id not in workouts:
113        raise HTTPException(status_code=404, detail="Workout not found")
114    if clip_id not in workouts[workout_id]["clips"]:
115        raise HTTPException(status_code=404, detail="Clip not found")
116    return {"views": workouts[workout_id]["clips"][clip_id]["views"]}

1fastapi
2uvicorn
3python-multipart