Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from datetime import datetime, timedelta
3import secrets
4import math
5 
6app = FastAPI()
7 
8users = {}
9tokens = {}
10subscriptions = {}
11churn_tracker = {}
12sub_id_counter = 1
13user_id_counter = 1
14 
15PLANS = {
16    "free": {"price": 0, "billing_cycle": "monthly"},
17    "basic": {"price": 9.99, "billing_cycle": "monthly"},
18    "pro": {"price": 29.99, "billing_cycle": "monthly"},
19    "enterprise": {"price": 99.99, "billing_cycle": "yearly"},
20}
21 
22def get_user_from_token(authorization: str = Header(...)):
23    token = authorization.replace("Bearer ", "")
24    if token not in tokens:
25        raise HTTPException(status_code=401, detail="Invalid token")
26    return tokens[token]
27 
28@app.post("/signup")
29def signup(username: str, password: str):
30    global user_id_counter
31    for u in users.values():
32        if u["username"] == username:
33            raise HTTPException(status_code=400, detail="Username already exists")
34    user_id = user_id_counter
35    user_id_counter += 1
36    users[user_id] = {"id": user_id, "username": username, "password": password, "tier": "free"}
37    subscriptions[user_id] = {
38        "user_id": user_id,
39        "plan": "free",
40        "price": 0,
41        "billing_cycle": "monthly",
42        "active": True,
43        "start_date": datetime.now().isoformat(),
44    }
45    churn_tracker[user_id] = datetime.now()
46    return {"user_id": user_id, "username": username}
47 
48@app.post("/login")
49def login(username: str, password: str):
50    for uid, u in users.items():
51        if u["username"] == username and u["password"] == password:
52            token = secrets.token_hex(16)
53            tokens[token] = uid
54            return {"token": token}
55    raise HTTPException(status_code=401, detail="Invalid credentials")
56 
57@app.get("/subscriptions")
58def get_subscriptions(authorization: str = Header(...)):
59    user_id = get_user_from_token(authorization)
60    sub = subscriptions.get(user_id)
61    if not sub:
62        raise HTTPException(status_code=404, detail="No subscription found")
63    return {
64        "plan": sub["plan"],
65        "price": sub["price"],
66        "billing_cycle": sub["billing_cycle"],
67        "active": sub["active"],
68    }
69 
70@app.post("/upgrade")
71def upgrade(tier: str, authorization: str = Header(...)):
72    user_id = get_user_from_token(authorization)
73    if tier not in PLANS:
74        raise HTTPException(status_code=400, detail="Invalid tier")
75    sub = subscriptions.get(user_id)
76    if not sub:
77        raise HTTPException(status_code=404, detail="No subscription found")
78    old_plan = sub["plan"]
79    new_plan = tier
80    old_price = PLANS[old_plan]["price"]
81    new_price = PLANS[new_plan]["price"]
82    if new_price <= old_price:
83        prorated_charge = 0
84    else:
85        days_in_cycle = 30
86        days_used = (datetime.now() - datetime.fromisoformat(sub["start_date"])).days
87        days_remaining = max(0, days_in_cycle - days_used)
88        daily_rate = old_price / days_in_cycle
89        credit = daily_rate * days_remaining
90        prorated_charge = round(new_price - credit, 2)
91    sub["plan"] = new_plan
92    sub["price"] = new_price
93    sub["billing_cycle"] = PLANS[new_plan]["billing_cycle"]
94    sub["start_date"] = datetime.now().isoformat()
95    users[user_id]["tier"] = new_plan
96    return {"new_plan": new_plan, "prorated_charge": prorated_charge}
97 
98@app.get("/churn-risk")
99def get_churn_risk(authorization: str = Header(...)):
100    user_id = get_user_from_token(authorization)
101    cutoff = datetime.now() - timedelta(days=45)
102    at_risk = []
103    for uid, last_active in churn_tracker.items():
104        if last_active < cutoff:
105            u = users.get(uid)
106            if u:
107                at_risk.append({"user_id": uid, "username": u["username"]})
108    return {"churn_risk_users": at_risk}
109 
110@app.get("/{resource}/{resource_id}")
111def get_resource(resource: str, resource_id: int, authorization: str = Header(...)):
112    user_id = get_user_from_token(authorization)
113    if resource == "users":
114        if resource_id not in users:
115            raise HTTPException(status_code=404, detail="User not found")
116        return users[resource_id]
117    elif resource == "subscriptions":
118        if resource_id not in subscriptions:
119            raise HTTPException(status_code=404, detail="Subscription not found")
120        return subscriptions[resource_id]
121    raise HTTPException(status_code=400, detail="Unknown resource")
122 
123@app.post("/{resource}")
124def create_resource(resource: str, data: dict, authorization: str = Header(...)):
125    user_id = get_user_from_token(authorization)
126    if resource == "users":
127        raise HTTPException(status_code=400, detail="Use /signup for users")
128    elif resource == "subscriptions":
129        raise HTTPException(status_code=400, detail="Use /upgrade for subscriptions")
130    raise HTTPException(status_code=400, detail="Unknown resource")

1fastapi
2uvicorn