Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import random
5import string
6 
7app = FastAPI()
8 
9users = {}
10user_tokens = {}
11newsletters = {}
12subscribers = {}
13subscriptions = {}
14open_rates = {}
15 
16next_user_id = 1
17next_newsletter_id = 1
18next_subscriber_id = 1
19next_subscription_id = 1
20 
21def generate_token():
22    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
23 
24def get_current_user(authorization: Optional[str] = Header(None)):
25    if not authorization:
26        raise HTTPException(status_code=401, detail="Missing auth header")
27    token = authorization.replace("Bearer ", "")
28    for uid, t in user_tokens.items():
29        if t == token:
30            return uid
31    raise HTTPException(status_code=401, detail="Invalid token")
32 
33class SignupRequest(BaseModel):
34    username: str
35    password: str
36 
37class LoginRequest(BaseModel):
38    username: str
39    password: str
40 
41class NewsletterCreate(BaseModel):
42    name: str
43    topic: str
44 
45class SubscriberCreate(BaseModel):
46    email: str
47 
48class SubscriptionCreate(BaseModel):
49    newsletter_id: int
50    subscriber_id: int
51 
52@app.post("/signup")
53def signup(req: SignupRequest):
54    global next_user_id
55    for u in users.values():
56        if u["username"] == req.username:
57            raise HTTPException(status_code=400, detail="Username taken")
58    uid = next_user_id
59    next_user_id += 1
60    users[uid] = {"id": uid, "username": req.username, "password": req.password}
61    token = generate_token()
62    user_tokens[uid] = token
63    return {"user_id": uid, "token": token}
64 
65@app.post("/login")
66def login(req: LoginRequest):
67    for uid, u in users.items():
68        if u["username"] == req.username and u["password"] == req.password:
69            token = generate_token()
70            user_tokens[uid] = token
71            return {"user_id": uid, "token": token}
72    raise HTTPException(status_code=401, detail="Invalid credentials")
73 
74@app.post("/newsletters")
75def create_newsletter(newsletter: NewsletterCreate, authorization: Optional[str] = Header(None)):
76    get_current_user(authorization)
77    global next_newsletter_id
78    nid = next_newsletter_id
79    next_newsletter_id += 1
80    newsletters[nid] = {"id": nid, "name": newsletter.name, "topic": newsletter.topic}
81    return newsletters[nid]
82 
83@app.get("/newsletters/{newsletter_id}")
84def get_newsletter(newsletter_id: int):
85    n = newsletters.get(newsletter_id)
86    if not n:
87        raise HTTPException(status_code=404, detail="Newsletter not found")
88    return n
89 
90@app.post("/subscribers")
91def create_subscriber(subscriber: SubscriberCreate, authorization: Optional[str] = Header(None)):
92    get_current_user(authorization)
93    global next_subscriber_id
94    sid = next_subscriber_id
95    next_subscriber_id += 1
96    subscribers[sid] = {"id": sid, "email": subscriber.email}
97    return subscribers[sid]
98 
99@app.get("/subscribers/{subscriber_id}")
100def get_subscriber(subscriber_id: int):
101    s = subscribers.get(subscriber_id)
102    if not s:
103        raise HTTPException(status_code=404, detail="Subscriber not found")
104    return s
105 
106@app.post("/subscriptions")
107def create_subscription(subscription: SubscriptionCreate, authorization: Optional[str] = Header(None)):
108    get_current_user(authorization)
109    if subscription.newsletter_id not in newsletters:
110        raise HTTPException(status_code=404, detail="Newsletter not found")
111    if subscription.subscriber_id not in subscribers:
112        raise HTTPException(status_code=404, detail="Subscriber not found")
113    global next_subscription_id
114    sub_id = next_subscription_id
115    next_subscription_id += 1
116    subscriptions[sub_id] = {
117        "id": sub_id,
118        "newsletter_id": subscription.newsletter_id,
119        "subscriber_id": subscription.subscriber_id
120    }
121    open_rates[sub_id] = {"opens": 0, "sends": 0}
122    return subscriptions[sub_id]
123 
124@app.get("/subscriptions/{subscription_id}")
125def get_subscription(subscription_id: int):
126    sub = subscriptions.get(subscription_id)
127    if not sub:
128        raise HTTPException(status_code=404, detail="Subscription not found")
129    rate = open_rates.get(subscription_id, {"opens": 0, "sends": 0})
130    return {**sub, "open_rate": rate["opens"] / rate["sends"] if rate["sends"] > 0 else 0.0}
131 
132@app.post("/subscriptions/{subscription_id}/open")
133def record_open(subscription_id: int):
134    if subscription_id not in subscriptions:
135        raise HTTPException(status_code=404, detail="Subscription not found")
136    if subscription_id not in open_rates:
137        open_rates[subscription_id] = {"opens": 0, "sends": 0}
138    open_rates[subscription_id]["opens"] += 1
139    open_rates[subscription_id]["sends"] += 1
140    return {"status": "ok"}
141 
142@app.post("/subscriptions/{subscription_id}/send")
143def record_send(subscription_id: int):
144    if subscription_id not in subscriptions:
145        raise HTTPException(status_code=404, detail="Subscription not found")
146    if subscription_id not in open_rates:
147        open_rates[subscription_id] = {"opens": 0, "sends": 0}
148    open_rates[subscription_id]["sends"] += 1
149    return {"status": "ok"}

1fastapi
2uvicorn