Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from typing import Optional
3import secrets
4 
5app = FastAPI()
6 
7users = {}
8tokens = {}
9subscriptions = {}
10billing_history = {}
11next_user_id = 1
12next_sub_id = 1
13next_bill_id = 1
14 
15def get_current_user(authorization: Optional[str] = Header(None)):
16    if not authorization or not authorization.startswith("Bearer "):
17        raise HTTPException(status_code=401, detail="Unauthorized")
18    token = authorization[7:]
19    user_id = tokens.get(token)
20    if not user_id:
21        raise HTTPException(status_code=401, detail="Invalid token")
22    return user_id
23 
24@app.post("/signup")
25def signup(username: str, password: str):
26    global next_user_id
27    user_id = next_user_id
28    next_user_id += 1
29    users[user_id] = {"id": user_id, "username": username, "password": password}
30    token = secrets.token_hex(16)
31    tokens[token] = user_id
32    return {"user_id": user_id, "token": token}
33 
34@app.post("/login")
35def login(username: str, password: str):
36    for uid, u in users.items():
37        if u["username"] == username and u["password"] == password:
38            token = secrets.token_hex(16)
39            tokens[token] = uid
40            return {"token": token}
41    raise HTTPException(status_code=401, detail="Invalid credentials")
42 
43@app.get("/subscribers")
44def get_subscribers(plan: Optional[str] = None, status: Optional[str] = None, authorization: Optional[str] = Header(None)):
45    user_id = get_current_user(authorization)
46    results = []
47    for sid, sub in subscriptions.items():
48        if plan and sub.get("plan") != plan:
49            continue
50        if status and sub.get("status") != status:
51            continue
52        bills = [b for b in billing_history.values() if b["subscription_id"] == sid]
53        latest_bill = max(bills, key=lambda b: b["id"]) if bills else None
54        results.append({
55            "subscription_id": sid,
56            "user_id": sub["user_id"],
57            "plan": sub["plan"],
58            "status": sub["status"],
59            "renewal_date": sub.get("renewal_date"),
60            "payment_method": latest_bill["payment_method"] if latest_bill else None
61        })
62    return results
63 
64@app.get("/subscriptions/{subscription_id}")
65def get_subscription(subscription_id: int, authorization: Optional[str] = Header(None)):
66    user_id = get_current_user(authorization)
67    sub = subscriptions.get(subscription_id)
68    if not sub:
69        raise HTTPException(status_code=404, detail="Subscription not found")
70    return sub
71 
72@app.post("/subscriptions")
73def create_subscription(plan: str, status: str, renewal_date: str, authorization: Optional[str] = Header(None)):
74    user_id = get_current_user(authorization)
75    global next_sub_id
76    sub_id = next_sub_id
77    next_sub_id += 1
78    subscriptions[sub_id] = {
79        "id": sub_id,
80        "user_id": user_id,
81        "plan": plan,
82        "status": status,
83        "renewal_date": renewal_date
84    }
85    return {"subscription_id": sub_id}
86 
87@app.post("/billing")
88def add_billing(subscription_id: int, payment_method: str, amount: float, authorization: Optional[str] = Header(None)):
89    user_id = get_current_user(authorization)
90    if subscription_id not in subscriptions:
91        raise HTTPException(status_code=404, detail="Subscription not found")
92    global next_bill_id
93    bill_id = next_bill_id
94    next_bill_id += 1
95    billing_history[bill_id] = {
96        "id": bill_id,
97        "subscription_id": subscription_id,
98        "payment_method": payment_method,
99        "amount": amount
100    }
101    return {"billing_id": bill_id}

1fastapi
2uvicorn