Browse the data

1import os
2import hashlib
3import secrets
4from fastapi import FastAPI, HTTPException, Header, Path, Query
5from fastapi.responses import StreamingResponse, FileResponse
6from pydantic import BaseModel
7 
8app = FastAPI()
9 
10users = {}
11tokens = {}
12assets = {}
13asset_versions = {}
14asset_id_counter = 1
15 
16GAME_ASSETS_DIR = "/game_assets/"
17 
18class SignupRequest(BaseModel):
19    username: str
20    password: str
21 
22class LoginRequest(BaseModel):
23    username: str
24    password: str
25 
26class AssetCreate(BaseModel):
27    name: str
28    file_path: str
29 
30def hash_password(password: str) -> str:
31    return hashlib.sha256(password.encode()).hexdigest()
32 
33def generate_token() -> str:
34    return secrets.token_hex(32)
35 
36def verify_token(authorization: str = Header(None)):
37    if not authorization:
38        raise HTTPException(status_code=401, detail="Missing authorization header")
39    token = authorization.replace("Bearer ", "")
40    if token not in tokens:
41        raise HTTPException(status_code=401, detail="Invalid token")
42    return tokens[token]
43 
44@app.post("/signup")
45def signup(req: SignupRequest):
46    if req.username in users:
47        raise HTTPException(status_code=400, detail="Username already exists")
48    users[req.username] = hash_password(req.password)
49    return {"message": "User created"}
50 
51@app.post("/login")
52def login(req: LoginRequest):
53    if req.username not in users or users[req.username] != hash_password(req.password):
54        raise HTTPException(status_code=401, detail="Invalid credentials")
55    token = generate_token()
56    tokens[token] = req.username
57    return {"token": token}
58 
59@app.post("/assets")
60def create_asset(req: AssetCreate, authorization: str = Header(None)):
61    user = verify_token(authorization)
62    global asset_id_counter
63    asset_id = asset_id_counter
64    asset_id_counter += 1
65    assets[asset_id] = {"id": asset_id, "name": req.name, "file_path": req.file_path, "owner": user}
66    asset_versions[asset_id] = [{"version": 1, "file_path": req.file_path}]
67    return {"id": asset_id, "name": req.name}
68 
69@app.get("/assets/{asset_id}")
70def get_asset(asset_id: int, authorization: str = Header(None)):
71    verify_token(authorization)
72    if asset_id not in assets:
73        raise HTTPException(status_code=404, detail="Asset not found")
74    return assets[asset_id]
75 
76@app.get("/assets/{game_id}/{asset_path:path}")
77def stream_asset(game_id: int, asset_path: str, version: int = Query(None), authorization: str = Header(None)):
78    verify_token(authorization)
79    full_path = os.path.join(GAME_ASSETS_DIR, asset_path)
80    if not os.path.exists(full_path):
81        raise HTTPException(status_code=404, detail="File not found")
82    if version is not None:
83        if game_id not in asset_versions:
84            raise HTTPException(status_code=404, detail="No versions for this asset")
85        version_data = next((v for v in asset_versions[game_id] if v["version"] == version), None)
86        if not version_data:
87            raise HTTPException(status_code=404, detail="Version not found")
88        full_path = os.path.join(GAME_ASSETS_DIR, version_data["file_path"])
89        if not os.path.exists(full_path):
90            raise HTTPException(status_code=404, detail="Version file not found")
91    return FileResponse(full_path, media_type="application/octet-stream")

1fastapi
2uvicorn