Browse the data

1import os
2import shutil
3import time
4import hashlib
5import secrets
6from pathlib import Path
7from typing import Optional
8 
9from fastapi import FastAPI, HTTPException, Header, UploadFile, File, Form
10from fastapi.responses import FileResponse
11 
12app = FastAPI()
13 
14users = {}
15tokens = {}
16teams = {}
17doc_versions = {}
18 
19BASE_DIR = "/var/teams"
20 
21def get_current_user(authorization: Optional[str] = Header(None)):
22    if not authorization:
23        raise HTTPException(status_code=401, detail="No auth header")
24    token = authorization.replace("Bearer ", "")
25    if token not in tokens:
26        raise HTTPException(status_code=401, detail="Invalid token")
27    return tokens[token]
28 
29@app.post("/signup")
30async def signup(username: str = Form(...), password: str = Form(...)):
31    if username in users:
32        raise HTTPException(status_code=400, detail="User exists")
33    user_id = len(users) + 1
34    users[username] = {"id": user_id, "password": password}
35    return {"id": user_id, "username": username}
36 
37@app.post("/login")
38async def login(username: str = Form(...), password: str = Form(...)):
39    if username not in users or users[username]["password"] != password:
40        raise HTTPException(status_code=401, detail="Bad credentials")
41    token = secrets.token_hex(32)
42    tokens[token] = users[username]
43    return {"token": token}
44 
45@app.post("/team")
46async def create_team(name: str = Form(...), authorization: Optional[str] = Header(None)):
47    user = get_current_user(authorization)
48    team_id = len(teams) + 1
49    teams[team_id] = {"id": team_id, "name": name, "owner": user["id"]}
50    os.makedirs(f"{BASE_DIR}/{team_id}", exist_ok=True)
51    return teams[team_id]
52 
53@app.get("/team/{team_id}")
54async def get_team(team_id: int, authorization: Optional[str] = Header(None)):
55    user = get_current_user(authorization)
56    if team_id not in teams:
57        raise HTTPException(status_code=404, detail="Team not found")
58    return teams[team_id]
59 
60@app.post("/team/{team_id}/docs")
61async def upload_doc(team_id: int, file: UploadFile = File(...), authorization: Optional[str] = Header(None)):
62    user = get_current_user(authorization)
63    if team_id not in teams:
64        raise HTTPException(status_code=404, detail="Team not found")
65    team_dir = f"{BASE_DIR}/{team_id}"
66    os.makedirs(team_dir, exist_ok=True)
67    file_path = f"{team_dir}/{file.filename}"
68    content = await file.read()
69    with open(file_path, "wb") as f:
70        f.write(content)
71    if team_id not in doc_versions:
72        doc_versions[team_id] = {}
73    if file.filename not in doc_versions[team_id]:
74        doc_versions[team_id][file.filename] = []
75    version = len(doc_versions[team_id][file.filename]) + 1
76    version_dir = f"{team_dir}/.versions/{file.filename}"
77    os.makedirs(version_dir, exist_ok=True)
78    shutil.copy2(file_path, f"{version_dir}/v{version}")
79    doc_versions[team_id][file.filename].append({"version": version, "timestamp": time.time()})
80    return {"filename": file.filename, "version": version}
81 
82@app.get("/team/{team_id}/docs/{file_path:path}")
83async def get_doc(team_id: int, file_path: str, v: Optional[int] = None, authorization: Optional[str] = Header(None)):
84    user = get_current_user(authorization)
85    if team_id not in teams:
86        raise HTTPException(status_code=404, detail="Team not found")
87    if v:
88        version_dir = f"{BASE_DIR}/{team_id}/.versions/{file_path}"
89        version_file = f"{version_dir}/v{v}"
90        if not os.path.exists(version_file):
91            raise HTTPException(status_code=404, detail="Version not found")
92        return FileResponse(version_file)
93    full_path = f"{BASE_DIR}/{team_id}/{file_path}"
94    if not os.path.exists(full_path):
95        raise HTTPException(status_code=404, detail="File not found")
96    return FileResponse(full_path)

1fastapi
2uvicorn
3python-multipart