Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5import datetime
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11tiers = {}
12subscriptions = {}
13shipments = {}
14preferences = {}
15next_user_id = 1
16next_tier_id = 1
17next_subscription_id = 1
18next_shipment_id = 1
19next_preference_id = 1
20 
21def get_current_user(authorization: Optional[str] = Header(None)):
22    if not authorization:
23        raise HTTPException(status_code=401, detail="Missing auth header")
24    token = authorization.replace("Bearer ", "")
25    if token not in tokens:
26        raise HTTPException(status_code=401, detail="Invalid token")
27    return tokens[token]
28 
29class SignupRequest(BaseModel):
30    username: str
31    password: str
32 
33class LoginRequest(BaseModel):
34    username: str
35    password: str
36 
37class TierCreate(BaseModel):
38    name: str
39    price: float
40    description: Optional[str] = None
41 
42class SubscriptionCreate(BaseModel):
43    user_id: int
44    tier_id: int
45 
46class ShipmentCreate(BaseModel):
47    subscription_id: int
48    status: str = "pending"
49    tracking_number: Optional[str] = None
50 
51class PreferenceCreate(BaseModel):
52    subscription_id: int
53    preferences: dict
54 
55@app.post("/signup")
56def signup(req: SignupRequest):
57    global next_user_id
58    user_id = next_user_id
59    next_user_id += 1
60    users[user_id] = {"id": user_id, "username": req.username, "password": req.password}
61    return {"user_id": user_id}
62 
63@app.post("/login")
64def login(req: LoginRequest):
65    for uid, u in users.items():
66        if u["username"] == req.username and u["password"] == req.password:
67            token = secrets.token_hex(16)
68            tokens[token] = uid
69            return {"token": token}
70    raise HTTPException(status_code=401, detail="Invalid credentials")
71 
72@app.post("/tiers")
73def create_tier(tier: TierCreate, authorization: Optional[str] = Header(None)):
74    get_current_user(authorization)
75    global next_tier_id
76    tier_id = next_tier_id
77    next_tier_id += 1
78    tiers[tier_id] = {"id": tier_id, "name": tier.name, "price": tier.price, "description": tier.description}
79    return tiers[tier_id]
80 
81@app.get("/tiers/{tier_id}")
82def get_tier(tier_id: int, authorization: Optional[str] = Header(None)):
83    get_current_user(authorization)
84    if tier_id not in tiers:
85        raise HTTPException(status_code=404, detail="Tier not found")
86    return tiers[tier_id]
87 
88@app.post("/subscriptions")
89def create_subscription(sub: SubscriptionCreate, authorization: Optional[str] = Header(None)):
90    get_current_user(authorization)
91    global next_subscription_id
92    sub_id = next_subscription_id
93    next_subscription_id += 1
94    subscriptions[sub_id] = {"id": sub_id, "user_id": sub.user_id, "tier_id": sub.tier_id, "created_at": str(datetime.datetime.now())}
95    return subscriptions[sub_id]
96 
97@app.get("/subscriptions/{sub_id}")
98def get_subscription(sub_id: int, authorization: Optional[str] = Header(None)):
99    get_current_user(authorization)
100    if sub_id not in subscriptions:
101        raise HTTPException(status_code=404, detail="Subscription not found")
102    return subscriptions[sub_id]
103 
104@app.post("/shipments")
105def create_shipment(ship: ShipmentCreate, authorization: Optional[str] = Header(None)):
106    get_current_user(authorization)
107    global next_shipment_id
108    ship_id = next_shipment_id
109    next_shipment_id += 1
110    shipments[ship_id] = {"id": ship_id, "subscription_id": ship.subscription_id, "status": ship.status, "tracking_number": ship.tracking_number, "created_at": str(datetime.datetime.now())}
111    return shipments[ship_id]
112 
113@app.get("/shipments/{ship_id}")
114def get_shipment(ship_id: int, authorization: Optional[str] = Header(None)):
115    get_current_user(authorization)
116    if ship_id not in shipments:
117        raise HTTPException(status_code=404, detail="Shipment not found")
118    return shipments[ship_id]
119 
120@app.post("/preferences")
121def create_preference(pref: PreferenceCreate, authorization: Optional[str] = Header(None)):
122    get_current_user(authorization)
123    global next_preference_id
124    pref_id = next_preference_id
125    next_preference_id += 1
126    preferences[pref_id] = {"id": pref_id, "subscription_id": pref.subscription_id, "preferences": pref.preferences, "updated_at": str(datetime.datetime.now())}
127    return preferences[pref_id]
128 
129@app.get("/preferences/{pref_id}")
130def get_preference(pref_id: int, authorization: Optional[str] = Header(None)):
131    get_current_user(authorization)
132    if pref_id not in preferences:
133        raise HTTPException(status_code=404, detail="Preference not found")
134    return preferences[pref_id]

1fastapi
2uvicorn