Browse the data

1import os
2import hashlib
3import secrets
4from fastapi import FastAPI, HTTPException, Header
5from fastapi.responses import Response
6from pydantic import BaseModel
7 
8app = FastAPI()
9 
10users = {}
11tokens = {}
12replays = {}
13next_user_id = 1
14next_token_id = 1
15next_replay_id = 1
16 
17class SignupRequest(BaseModel):
18    username: str
19    password: str
20 
21class LoginRequest(BaseModel):
22    username: str
23    password: str
24 
25class ReplayCreateRequest(BaseModel):
26    filename: str
27 
28@app.post("/signup")
29def signup(req: SignupRequest):
30    global next_user_id
31    for u in users.values():
32        if u["username"] == req.username:
33            raise HTTPException(400, "User exists")
34    uid = next_user_id
35    next_user_id += 1
36    users[uid] = {"id": uid, "username": req.username, "password": req.password}
37    return {"id": uid, "username": req.username}
38 
39@app.post("/login")
40def login(req: LoginRequest):
41    for u in users.values():
42        if u["username"] == req.username and u["password"] == req.password:
43            token = secrets.token_hex(16)
44            tokens[token] = u["id"]
45            return {"token": token}
46    raise HTTPException(401, "Bad credentials")
47 
48def get_user_id(authorization: str = Header(None)):
49    if not authorization:
50        raise HTTPException(401, "Missing auth header")
51    token = authorization.replace("Bearer ", "")
52    uid = tokens.get(token)
53    if uid is None:
54        raise HTTPException(401, "Invalid token")
55    return uid
56 
57@app.post("/replay")
58def create_replay(req: ReplayCreateRequest, authorization: str = Header(None)):
59    global next_replay_id
60    uid = get_user_id(authorization)
61    rid = next_replay_id
62    next_replay_id += 1
63    replays[rid] = {"id": rid, "filename": req.filename, "user_id": uid}
64    return {"id": rid, "filename": req.filename}
65 
66@app.get("/replay/{replay_id}")
67def get_replay(replay_id: int, authorization: str = Header(None)):
68    uid = get_user_id(authorization)
69    r = replays.get(replay_id)
70    if r is None:
71        raise HTTPException(404, "Not found")
72    return r
73 
74@app.get("/replays/{match_id}/{filename}")
75def get_replay_file(match_id: int, filename: str, authorization: str = Header(None)):
76    uid = get_user_id(authorization)
77    filepath = f"/game_data/replays/{match_id}/{filename}"
78    if not os.path.exists(filepath):
79        raise HTTPException(404, "File not found")
80    with open(filepath, "rb") as f:
81        data = f.read()
82    return Response(content=data, media_type="application/octet-stream")

1fastapi
2uvicorn