Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import List, Optional, Dict
4import hashlib
5import secrets
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11expenses = {}
12groups = {}
13settlements = {}
14group_members = {}
15next_user_id = 1
16next_expense_id = 1
17next_group_id = 1
18next_settlement_id = 1
19 
20class SignupRequest(BaseModel):
21    username: str
22    password: str
23 
24class LoginRequest(BaseModel):
25    username: str
26    password: str
27 
28class ExpenseCreate(BaseModel):
29    group_id: int
30    payer_id: int
31    amount: float
32    split_members: List[int]
33 
34class SettlementCreate(BaseModel):
35    group_id: int
36    from_user_id: int
37    to_user_id: int
38    amount: float
39 
40class GroupCreate(BaseModel):
41    name: str
42    member_ids: List[int]
43 
44def get_current_user(authorization: str = Header(None)):
45    if not authorization:
46        raise HTTPException(status_code=401, detail="Missing auth header")
47    token = authorization.replace("Bearer ", "")
48    if token not in tokens:
49        raise HTTPException(status_code=401, detail="Invalid token")
50    return tokens[token]
51 
52@app.post("/signup")
53def signup(req: SignupRequest):
54    global next_user_id
55    for u in users.values():
56        if u["username"] == req.username:
57            raise HTTPException(status_code=400, detail="Username taken")
58    user_id = next_user_id
59    next_user_id += 1
60    users[user_id] = {"id": user_id, "username": req.username, "password": hashlib.sha256(req.password.encode()).hexdigest()}
61    return {"user_id": user_id}
62 
63@app.post("/login")
64def login(req: LoginRequest):
65    for u in users.values():
66        if u["username"] == req.username and u["password"] == hashlib.sha256(req.password.encode()).hexdigest():
67            token = secrets.token_hex(16)
68            tokens[token] = u["id"]
69            return {"token": token}
70    raise HTTPException(status_code=401, detail="Invalid credentials")
71 
72@app.post("/groups")
73def create_group(group: GroupCreate, authorization: str = Header(None)):
74    current_user = get_current_user(authorization)
75    global next_group_id
76    group_id = next_group_id
77    next_group_id += 1
78    groups[group_id] = {"id": group_id, "name": group.name, "created_by": current_user}
79    group_members[group_id] = group.member_ids
80    return {"group_id": group_id}
81 
82@app.get("/groups/{group_id}")
83def get_group(group_id: int, authorization: str = Header(None)):
84    get_current_user(authorization)
85    if group_id not in groups:
86        raise HTTPException(status_code=404, detail="Group not found")
87    return groups[group_id]
88 
89@app.post("/expenses")
90def create_expense(expense: ExpenseCreate, authorization: str = Header(None)):
91    get_current_user(authorization)
92    global next_expense_id
93    if expense.group_id not in groups:
94        raise HTTPException(status_code=404, detail="Group not found")
95    if expense.payer_id not in group_members[expense.group_id]:
96        raise HTTPException(status_code=400, detail="Payer not in group")
97    for mid in expense.split_members:
98        if mid not in group_members[expense.group_id]:
99            raise HTTPException(status_code=400, detail=f"Member {mid} not in group")
100    expense_id = next_expense_id
101    next_expense_id += 1
102    expenses[expense_id] = {
103        "id": expense_id,
104        "group_id": expense.group_id,
105        "payer_id": expense.payer_id,
106        "amount": expense.amount,
107        "split_members": expense.split_members
108    }
109    return {"expense_id": expense_id}
110 
111@app.get("/expenses/{expense_id}")
112def get_expense(expense_id: int, authorization: str = Header(None)):
113    get_current_user(authorization)
114    if expense_id not in expenses:
115        raise HTTPException(status_code=404, detail="Expense not found")
116    return expenses[expense_id]
117 
118@app.get("/groups/{group_id}/balances")
119def get_balances(group_id: int, authorization: str = Header(None)):
120    get_current_user(authorization)
121    if group_id not in groups:
122        raise HTTPException(status_code=404, detail="Group not found")
123    members = group_members[group_id]
124    balance = {uid: 0.0 for uid in members}
125    for eid, exp in expenses.items():
126        if exp["group_id"] != group_id:
127            continue
128        share = exp["amount"] / len(exp["split_members"])
129        balance[exp["payer_id"]] += exp["amount"]
130        for mid in exp["split_members"]:
131            balance[mid] -= share
132    for s in settlements.values():
133        if s["group_id"] != group_id:
134            continue
135        balance[s["from_user_id"]] += s["amount"]
136        balance[s["to_user_id"]] -= s["amount"]
137    debts = []
138    for uid, bal in balance.items():
139        if bal > 0:
140            for uid2, bal2 in balance.items():
141                if bal2 < 0 and abs(bal2) > 0.01:
142                    amount = min(bal, abs(bal2))
143                    if amount > 0.01:
144                        debts.append({"from": uid2, "to": uid, "amount": round(amount, 2)})
145                        bal -= amount
146                        balance[uid2] += amount
147    return {"group_id": group_id, "debts": debts}
148 
149@app.post("/settlements")
150def create_settlement(settlement: SettlementCreate, authorization: str = Header(None)):
151    get_current_user(authorization)
152    global next_settlement_id
153    if settlement.group_id not in groups:
154        raise HTTPException(status_code=404, detail="Group not found")
155    if settlement.from_user_id not in group_members[settlement.group_id] or settlement.to_user_id not in group_members[settlement.group_id]:
156        raise HTTPException(status_code=400, detail="Users not in group")
157    settlement_id = next_settlement_id
158    next_settlement_id += 1
159    settlements[settlement_id] = {
160        "id": settlement_id,
161        "group_id": settlement.group_id,
162        "from_user_id": settlement.from_user_id,
163        "to_user_id": settlement.to_user_id,
164        "amount": settlement.amount
165    }
166    return {"settlement_id": settlement_id}
167 
168@app.get("/settlements/{settlement_id}")
169def get_settlement(settlement_id: int, authorization: str = Header(None)):
170    get_current_user(authorization)
171    if settlement_id not in settlements:
172        raise HTTPException(status_code=404, detail="Settlement not found")
173    return settlements[settlement_id]

1fastapi
2uvicorn
3pydantic