Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from typing import Optional
3import secrets
4 
5app = FastAPI()
6 
7users = {}
8tokens = {}
9subscriptions = {}
10invoices = {}
11next_ids = {"users": 1, "subscriptions": 1, "invoices": 1}
12 
13def get_current_user(authorization: Optional[str] = Header(None)):
14    if not authorization or not authorization.startswith("Bearer "):
15        raise HTTPException(status_code=401, detail="Invalid auth")
16    token = authorization.split(" ")[1]
17    for uid, t in tokens.items():
18        if t == token:
19            return uid
20    raise HTTPException(status_code=401, detail="Invalid token")
21 
22@app.post("/signup")
23def signup(username: str, password: str):
24    uid = next_ids["users"]
25    next_ids["users"] += 1
26    users[uid] = {"username": username, "password": password}
27    return {"user_id": uid}
28 
29@app.post("/login")
30def login(username: str, password: str):
31    for uid, u in users.items():
32        if u["username"] == username and u["password"] == password:
33            token = secrets.token_hex(16)
34            tokens[uid] = token
35            return {"token": token}
36    raise HTTPException(status_code=401, detail="Invalid credentials")
37 
38@app.get("/subscriptions/{id}")
39def get_subscription(id: int, authorization: Optional[str] = Header(None)):
40    get_current_user(authorization)
41    if id not in subscriptions:
42        raise HTTPException(status_code=404, detail="Subscription not found")
43    return subscriptions[id]
44 
45@app.post("/subscriptions")
46def create_subscription(plan: str, billing_cycle: str, payment_method: str, account_tier: str, discount_code: str = "", authorization: Optional[str] = Header(None)):
47    uid = get_current_user(authorization)
48    sid = next_ids["subscriptions"]
49    next_ids["subscriptions"] += 1
50    subscriptions[sid] = {
51        "id": sid,
52        "user_id": uid,
53        "plan": plan,
54        "billing_cycle": billing_cycle,
55        "payment_method": payment_method,
56        "account_tier": account_tier,
57        "discount_code": discount_code
58    }
59    return subscriptions[sid]
60 
61@app.put("/subscriptions/{id}")
62def update_subscription(id: int, plan: Optional[str] = None, billing_cycle: Optional[str] = None, payment_method: Optional[str] = None, account_tier: Optional[str] = None, discount_code: Optional[str] = None, authorization: Optional[str] = Header(None)):
63    get_current_user(authorization)
64    if id not in subscriptions:
65        raise HTTPException(status_code=404, detail="Subscription not found")
66    sub = subscriptions[id]
67    if plan is not None:
68        sub["plan"] = plan
69    if billing_cycle is not None:
70        sub["billing_cycle"] = billing_cycle
71    if payment_method is not None:
72        sub["payment_method"] = payment_method
73    if account_tier is not None:
74        sub["account_tier"] = account_tier
75    if discount_code is not None:
76        sub["discount_code"] = discount_code
77    return sub
78 
79@app.get("/invoices/{id}")
80def get_invoice(id: int, authorization: Optional[str] = Header(None)):
81    get_current_user(authorization)
82    if id not in invoices:
83        raise HTTPException(status_code=404, detail="Invoice not found")
84    return invoices[id]
85 
86@app.post("/invoices")
87def create_invoice(subscription_id: int, amount: float, authorization: Optional[str] = Header(None)):
88    get_current_user(authorization)
89    iid = next_ids["invoices"]
90    next_ids["invoices"] += 1
91    invoices[iid] = {
92        "id": iid,
93        "subscription_id": subscription_id,
94        "amount": amount,
95        "status": "pending"
96    }
97    return invoices[iid]

1fastapi
2uvicorn