Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from typing import Optional
3import secrets
4 
5app = FastAPI()
6 
7users = {}
8tokens = {}
9deals = {}
10editors = {}
11deal_docs = {}
12 
13user_id_seq = 1
14deal_id_seq = 1
15doc_id_seq = 1
16 
17def get_current_user(authorization: Optional[str] = Header(None)):
18    if not authorization:
19        raise HTTPException(status_code=401, detail="Missing auth header")
20    token = authorization.replace("Bearer ", "")
21    if token not in tokens:
22        raise HTTPException(status_code=401, detail="Invalid token")
23    return tokens[token]
24 
25@app.post("/signup")
26def signup(username: str, password: str):
27    global user_id_seq
28    if any(u["username"] == username for u in users.values()):
29        raise HTTPException(status_code=400, detail="Username taken")
30    uid = user_id_seq
31    users[uid] = {"id": uid, "username": username, "password": password}
32    user_id_seq += 1
33    return {"id": uid, "username": username}
34 
35@app.post("/login")
36def login(username: str, password: str):
37    for u in users.values():
38        if u["username"] == username and u["password"] == password:
39            token = secrets.token_hex(16)
40            tokens[token] = u["id"]
41            return {"token": token}
42    raise HTTPException(status_code=401, detail="Invalid credentials")
43 
44@app.get("/deal/{deal_id}")
45def get_deal(deal_id: int, authorization: Optional[str] = Header(None)):
46    user_id = get_current_user(authorization)
47    if deal_id not in deals:
48        raise HTTPException(status_code=404, detail="Deal not found")
49    return deals[deal_id]
50 
51@app.post("/deal")
52def create_deal(name: str, authorization: Optional[str] = Header(None)):
53    global deal_id_seq
54    user_id = get_current_user(authorization)
55    deal_id = deal_id_seq
56    deals[deal_id] = {"id": deal_id, "name": name, "creator_id": user_id, "docs": []}
57    editors[deal_id] = [user_id]
58    deal_id_seq += 1
59    return deals[deal_id]
60 
61@app.post("/deal/{deal_id}/promote")
62def promote_to_editor(deal_id: int, target_username: str, authorization: Optional[str] = Header(None)):
63    user_id = get_current_user(authorization)
64    if deal_id not in deals:
65        raise HTTPException(status_code=404, detail="Deal not found")
66    if deals[deal_id]["creator_id"] != user_id:
67        raise HTTPException(status_code=403, detail="Only creator can promote")
68    target_user = None
69    for u in users.values():
70        if u["username"] == target_username:
71            target_user = u
72            break
73    if not target_user:
74        raise HTTPException(status_code=404, detail="User not found")
75    if target_user["id"] not in editors[deal_id]:
76        editors[deal_id].append(target_user["id"])
77    return {"status": "promoted", "user_id": target_user["id"]}
78 
79@app.get("/deal/{deal_id}/doc/{doc_id}")
80def get_doc(deal_id: int, doc_id: int, authorization: Optional[str] = Header(None)):
81    user_id = get_current_user(authorization)
82    if deal_id not in deals:
83        raise HTTPException(status_code=404, detail="Deal not found")
84    if user_id not in editors.get(deal_id, []):
85        raise HTTPException(status_code=403, detail="Not an editor")
86    doc_key = (deal_id, doc_id)
87    if doc_key not in deal_docs:
88        raise HTTPException(status_code=404, detail="Doc not found")
89    return deal_docs[doc_key]
90 
91@app.post("/deal/{deal_id}/doc")
92def upload_doc(deal_id: int, filename: str, content: str, authorization: Optional[str] = Header(None)):
93    global doc_id_seq
94    user_id = get_current_user(authorization)
95    if deal_id not in deals:
96        raise HTTPException(status_code=404, detail="Deal not found")
97    if user_id not in editors.get(deal_id, []):
98        raise HTTPException(status_code=403, detail="Not an editor")
99    doc_id = doc_id_seq
100    deal_docs[(deal_id, doc_id)] = {"id": doc_id, "deal_id": deal_id, "filename": filename, "content": content, "uploaded_by": user_id}
101    deals[deal_id]["docs"].append(doc_id)
102    doc_id_seq += 1
103    return deal_docs[(deal_id, doc_id)]
104 
105@app.post("/deal/{deal_id}/add_editor")
106def add_editor(deal_id: int, target_username: str, authorization: Optional[str] = Header(None)):
107    user_id = get_current_user(authorization)
108    if deal_id not in deals:
109        raise HTTPException(status_code=404, detail="Deal not found")
110    if user_id not in editors.get(deal_id, []):
111        raise HTTPException(status_code=403, detail="Not an editor")
112    target_user = None
113    for u in users.values():
114        if u["username"] == target_username:
115            target_user = u
116            break
117    if not target_user:
118        raise HTTPException(status_code=404, detail="User not found")
119    if target_user["id"] not in editors[deal_id]:
120        editors[deal_id].append(target_user["id"])
121    return {"status": "added", "user_id": target_user["id"]}

1fastapi
2uvicorn