Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List
4import secrets
5import datetime
6 
7app = FastAPI()
8 
9users = {}
10pledges = {}
11installments = {}
12tokens = {}
13pledge_id_counter = 1
14installment_id_counter = 1
15user_id_counter = 1
16 
17class SignupRequest(BaseModel):
18    username: str
19    password: str
20 
21class LoginRequest(BaseModel):
22    username: str
23    password: str
24 
25class PledgeCreate(BaseModel):
26    donor_name: str
27    total_amount: float
28    installment_plan: Optional[List[float]] = None
29 
30class InstallmentCreate(BaseModel):
31    pledge_id: int
32    amount: float
33    due_date: str
34 
35def get_current_user(authorization: str = Header(None)):
36    if not authorization:
37        raise HTTPException(status_code=401, detail="Missing auth header")
38    token = authorization.replace("Bearer ", "")
39    user_id = tokens.get(token)
40    if not user_id:
41        raise HTTPException(status_code=401, detail="Invalid token")
42    return user_id
43 
44@app.post("/signup")
45def signup(req: SignupRequest):
46    global user_id_counter
47    for u in users.values():
48        if u["username"] == req.username:
49            raise HTTPException(status_code=400, detail="Username exists")
50    user_id = user_id_counter
51    user_id_counter += 1
52    users[user_id] = {"id": user_id, "username": req.username, "password": req.password}
53    return {"user_id": user_id, "username": req.username}
54 
55@app.post("/login")
56def login(req: LoginRequest):
57    for uid, u in users.items():
58        if u["username"] == req.username and u["password"] == req.password:
59            token = secrets.token_hex(16)
60            tokens[token] = uid
61            return {"token": token}
62    raise HTTPException(status_code=401, detail="Invalid credentials")
63 
64@app.post("/pledges")
65def create_pledge(pledge: PledgeCreate, authorization: str = Header(None)):
66    get_current_user(authorization)
67    global pledge_id_counter
68    pid = pledge_id_counter
69    pledge_id_counter += 1
70    pledges[pid] = {
71        "id": pid,
72        "donor_name": pledge.donor_name,
73        "total_amount": pledge.total_amount,
74        "fulfilled": False,
75        "created_at": datetime.datetime.now().isoformat()
76    }
77    if pledge.installment_plan:
78        for amount in pledge.installment_plan:
79            global installment_id_counter
80            iid = installment_id_counter
81            installment_id_counter += 1
82            installments[iid] = {
83                "id": iid,
84                "pledge_id": pid,
85                "amount": amount,
86                "paid": False,
87                "due_date": None
88            }
89    return pledges[pid]
90 
91@app.get("/pledges/{pledge_id}")
92def get_pledge(pledge_id: int, authorization: str = Header(None)):
93    get_current_user(authorization)
94    p = pledges.get(pledge_id)
95    if not p:
96        raise HTTPException(status_code=404, detail="Pledge not found")
97    return p
98 
99@app.post("/installments")
100def create_installment(inst: InstallmentCreate, authorization: str = Header(None)):
101    get_current_user(authorization)
102    if inst.pledge_id not in pledges:
103        raise HTTPException(status_code=404, detail="Pledge not found")
104    global installment_id_counter
105    iid = installment_id_counter
106    installment_id_counter += 1
107    installments[iid] = {
108        "id": iid,
109        "pledge_id": inst.pledge_id,
110        "amount": inst.amount,
111        "due_date": inst.due_date,
112        "paid": False
113    }
114    return installments[iid]
115 
116@app.get("/installments/{installment_id}")
117def get_installment(installment_id: int, authorization: str = Header(None)):
118    get_current_user(authorization)
119    i = installments.get(installment_id)
120    if not i:
121        raise HTTPException(status_code=404, detail="Installment not found")
122    return i
123 
124@app.post("/reconcile/{pledge_id}")
125def reconcile_pledge(pledge_id: int, authorization: str = Header(None)):
126    get_current_user(authorization)
127    p = pledges.get(pledge_id)
128    if not p:
129        raise HTTPException(status_code=404, detail="Pledge not found")
130    total_paid = sum(i["amount"] for i in installments.values() if i["pledge_id"] == pledge_id and i["paid"])
131    if total_paid >= p["total_amount"]:
132        p["fulfilled"] = True
133        return {"status": "fulfilled", "pledge_id": pledge_id}
134    else:
135        remaining = p["total_amount"] - total_paid
136        return {"status": "partial", "fulfilled": False, "remaining": remaining}
137 
138@app.get("/alerts")
139def get_alerts(authorization: str = Header(None)):
140    get_current_user(authorization)
141    alerts = []
142    for pid, p in pledges.items():
143        if p["fulfilled"]:
144            continue
145        due_soon = []
146        for iid, i in installments.items():
147            if i["pledge_id"] == pid and not i["paid"] and i["due_date"]:
148                due_date = datetime.datetime.fromisoformat(i["due_date"])
149                if due_date <= datetime.datetime.now() + datetime.timedelta(days=7):
150                    due_soon.append(i)
151        if due_soon:
152            alerts.append({"pledge_id": pid, "donor": p["donor_name"], "due_installments": due_soon})
153    return alerts

1fastapi
2uvicorn
3pydantic