Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, Dict
4import secrets
5import datetime
6 
7app = FastAPI()
8 
9users: Dict[int, dict] = {}
10next_user_id = 1
11tokens: Dict[str, int] = {}
12 
13plans: Dict[int, dict] = {}
14next_plan_id = 1
15 
16subscriptions: Dict[int, dict] = {}
17next_sub_id = 1
18 
19class SignupRequest(BaseModel):
20    username: str
21    password: str
22 
23class LoginRequest(BaseModel):
24    username: str
25    password: str
26 
27class PlanCreate(BaseModel):
28    name: str
29    delivery_frequency_days: int
30    theme: str
31    price_per_month: float
32 
33class SubscriptionCreate(BaseModel):
34    user_id: int
35    plan_id: int
36    start_date: str
37 
38class SubscriptionUpdate(BaseModel):
39    plan_id: Optional[int] = None
40    delivery_frequency_days: Optional[int] = None
41    theme: Optional[str] = None
42 
43class CancelSubscription(BaseModel):
44    cancellation_date: str
45 
46def get_user_id_from_token(authorization: str = Header(...)):
47    if not authorization.startswith("Bearer "):
48        raise HTTPException(status_code=401, detail="Invalid auth header")
49    token = authorization[7:]
50    user_id = tokens.get(token)
51    if user_id is None:
52        raise HTTPException(status_code=401, detail="Invalid token")
53    return user_id
54 
55@app.post("/signup")
56def signup(req: SignupRequest):
57    global next_user_id
58    for u in users.values():
59        if u["username"] == req.username:
60            raise HTTPException(status_code=400, detail="Username taken")
61    user_id = next_user_id
62    next_user_id += 1
63    users[user_id] = {"id": user_id, "username": req.username, "password": req.password}
64    return {"user_id": user_id}
65 
66@app.post("/login")
67def login(req: LoginRequest):
68    for u in users.values():
69        if u["username"] == req.username and u["password"] == req.password:
70            token = secrets.token_hex(16)
71            tokens[token] = u["id"]
72            return {"token": token}
73    raise HTTPException(status_code=401, detail="Invalid credentials")
74 
75@app.post("/plans")
76def create_plan(plan: PlanCreate, authorization: str = Header(...)):
77    get_user_id_from_token(authorization)
78    global next_plan_id
79    plan_id = next_plan_id
80    next_plan_id += 1
81    plans[plan_id] = {
82        "id": plan_id,
83        "name": plan.name,
84        "delivery_frequency_days": plan.delivery_frequency_days,
85        "theme": plan.theme,
86        "price_per_month": plan.price_per_month
87    }
88    return plans[plan_id]
89 
90@app.get("/plans/{plan_id}")
91def get_plan(plan_id: int, authorization: str = Header(...)):
92    get_user_id_from_token(authorization)
93    plan = plans.get(plan_id)
94    if not plan:
95        raise HTTPException(status_code=404, detail="Plan not found")
96    return plan
97 
98@app.post("/subscriptions")
99def create_subscription(sub: SubscriptionCreate, authorization: str = Header(...)):
100    user_id = get_user_id_from_token(authorization)
101    if sub.user_id != user_id:
102        raise HTTPException(status_code=403, detail="Cannot create subscription for another user")
103    if sub.plan_id not in plans:
104        raise HTTPException(status_code=404, detail="Plan not found")
105    global next_sub_id
106    sub_id = next_sub_id
107    next_sub_id += 1
108    subscriptions[sub_id] = {
109        "id": sub_id,
110        "user_id": sub.user_id,
111        "plan_id": sub.plan_id,
112        "start_date": sub.start_date,
113        "delivery_frequency_days": plans[sub.plan_id]["delivery_frequency_days"],
114        "theme": plans[sub.plan_id]["theme"],
115        "price_per_month": plans[sub.plan_id]["price_per_month"],
116        "active": True
117    }
118    return subscriptions[sub_id]
119 
120@app.get("/subscriptions/{sub_id}")
121def get_subscription(sub_id: int, authorization: str = Header(...)):
122    user_id = get_user_id_from_token(authorization)
123    sub = subscriptions.get(sub_id)
124    if not sub:
125        raise HTTPException(status_code=404, detail="Subscription not found")
126    if sub["user_id"] != user_id:
127        raise HTTPException(status_code=403, detail="Not your subscription")
128    return sub
129 
130@app.patch("/subscriptions/{sub_id}")
131def update_subscription(sub_id: int, update: SubscriptionUpdate, authorization: str = Header(...)):
132    user_id = get_user_id_from_token(authorization)
133    sub = subscriptions.get(sub_id)
134    if not sub:
135        raise HTTPException(status_code=404, detail="Subscription not found")
136    if sub["user_id"] != user_id:
137        raise HTTPException(status_code=403, detail="Not your subscription")
138    if update.plan_id is not None:
139        if update.plan_id not in plans:
140            raise HTTPException(status_code=404, detail="Plan not found")
141        sub["plan_id"] = update.plan_id
142        sub["delivery_frequency_days"] = plans[update.plan_id]["delivery_frequency_days"]
143        sub["theme"] = plans[update.plan_id]["theme"]
144        sub["price_per_month"] = plans[update.plan_id]["price_per_month"]
145    if update.delivery_frequency_days is not None:
146        sub["delivery_frequency_days"] = update.delivery_frequency_days
147    if update.theme is not None:
148        sub["theme"] = update.theme
149    return sub
150 
151@app.post("/subscriptions/{sub_id}/cancel")
152def cancel_subscription(sub_id: int, cancel: CancelSubscription, authorization: str = Header(...)):
153    user_id = get_user_id_from_token(authorization)
154    sub = subscriptions.get(sub_id)
155    if not sub:
156        raise HTTPException(status_code=404, detail="Subscription not found")
157    if sub["user_id"] != user_id:
158        raise HTTPException(status_code=403, detail="Not your subscription")
159    if not sub["active"]:
160        raise HTTPException(status_code=400, detail="Already cancelled")
161    sub["active"] = False
162    sub["cancellation_date"] = cancel.cancellation_date
163    start = datetime.date.fromisoformat(sub["start_date"])
164    cancel_date = datetime.date.fromisoformat(cancel.cancellation_date)
165    days_used = (cancel_date - start).days
166    total_days = 30
167    prorated_refund = max(0.0, sub["price_per_month"] * (1 - days_used / total_days))
168    return {"refund_amount": round(prorated_refund, 2), "subscription": sub}

1fastapi
2uvicorn