Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List
4from datetime import datetime
5import random
6import string
7 
8app = FastAPI()
9 
10# In-memory stores
11users = {}
12user_id_counter = 0
13tokens = {}
14 
15newsletters = {}
16newsletter_id_counter = 0
17 
18analytics = {}
19 
20def generate_token():
21    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
22 
23def get_current_user(authorization: str = Header(None)):
24    if not authorization:
25        raise HTTPException(status_code=401, detail="Missing auth token")
26    token = authorization.replace("Bearer ", "")
27    user_id = tokens.get(token)
28    if not user_id:
29        raise HTTPException(status_code=401, detail="Invalid token")
30    return user_id
31 
32class SignupRequest(BaseModel):
33    email: str
34    password: str
35 
36class LoginRequest(BaseModel):
37    email: str
38    password: str
39 
40class NewsletterCreate(BaseModel):
41    subject: str
42    body: str
43    send_schedule: Optional[str] = None
44    subscriber_segments: Optional[List[str]] = None
45    a_test_enabled: bool = False
46    b_test_enabled: bool = False
47    a_subject: Optional[str] = None
48    a_body: Optional[str] = None
49    b_subject: Optional[str] = None
50    b_body: Optional[str] = None
51 
52class NewsletterUpdate(BaseModel):
53    subject: Optional[str] = None
54    body: Optional[str] = None
55    send_schedule: Optional[str] = None
56    subscriber_segments: Optional[List[str]] = None
57    a_test_enabled: Optional[bool] = None
58    b_test_enabled: Optional[bool] = None
59    a_subject: Optional[str] = None
60    a_body: Optional[str] = None
61    b_subject: Optional[str] = None
62    b_body: Optional[str] = None
63 
64@app.post("/signup")
65def signup(req: SignupRequest):
66    global user_id_counter
67    user_id_counter += 1
68    users[user_id_counter] = {"email": req.email, "password": req.password}
69    return {"user_id": user_id_counter}
70 
71@app.post("/login")
72def login(req: LoginRequest):
73    for uid, u in users.items():
74        if u["email"] == req.email and u["password"] == req.password:
75            token = generate_token()
76            tokens[token] = uid
77            return {"token": token}
78    raise HTTPException(status_code=401, detail="Invalid credentials")
79 
80@app.post("/newsletters")
81def create_newsletter(req: NewsletterCreate, authorization: str = Header(None)):
82    user_id = get_current_user(authorization)
83    global newsletter_id_counter
84    newsletter_id_counter += 1
85    nid = newsletter_id_counter
86    newsletters[nid] = {
87        "id": nid,
88        "user_id": user_id,
89        "subject": req.subject,
90        "body": req.body,
91        "send_schedule": req.send_schedule,
92        "subscriber_segments": req.subscriber_segments or [],
93        "a_test_enabled": req.a_test_enabled,
94        "b_test_enabled": req.b_test_enabled,
95        "a_subject": req.a_subject,
96        "a_body": req.a_body,
97        "b_subject": req.b_subject,
98        "b_body": req.b_body,
99        "created_at": datetime.now().isoformat()
100    }
101    # Initialize analytics for this newsletter
102    analytics[nid] = {
103        "sent": 0,
104        "opened": 0,
105        "clicked": 0,
106        "a_sent": 0,
107        "a_opened": 0,
108        "a_clicked": 0,
109        "b_sent": 0,
110        "b_opened": 0,
111        "b_clicked": 0
112    }
113    return newsletters[nid]
114 
115@app.get("/newsletters/{newsletter_id}")
116def get_newsletter(newsletter_id: int, authorization: str = Header(None)):
117    user_id = get_current_user(authorization)
118    n = newsletters.get(newsletter_id)
119    if not n:
120        raise HTTPException(status_code=404, detail="Newsletter not found")
121    if n["user_id"] != user_id:
122        raise HTTPException(status_code=403, detail="Not your newsletter")
123    return n
124 
125@app.put("/newsletters/{newsletter_id}")
126def update_newsletter(newsletter_id: int, req: NewsletterUpdate, authorization: str = Header(None)):
127    user_id = get_current_user(authorization)
128    n = newsletters.get(newsletter_id)
129    if not n:
130        raise HTTPException(status_code=404, detail="Newsletter not found")
131    if n["user_id"] != user_id:
132        raise HTTPException(status_code=403, detail="Not your newsletter")
133    if req.subject is not None:
134        n["subject"] = req.subject
135    if req.body is not None:
136        n["body"] = req.body
137    if req.send_schedule is not None:
138        n["send_schedule"] = req.send_schedule
139    if req.subscriber_segments is not None:
140        n["subscriber_segments"] = req.subscriber_segments
141    if req.a_test_enabled is not None:
142        n["a_test_enabled"] = req.a_test_enabled
143    if req.b_test_enabled is not None:
144        n["b_test_enabled"] = req.b_test_enabled
145    if req.a_subject is not None:
146        n["a_subject"] = req.a_subject
147    if req.a_body is not None:
148        n["a_body"] = req.a_body
149    if req.b_subject is not None:
150        n["b_subject"] = req.b_subject
151    if req.b_body is not None:
152        n["b_body"] = req.b_body
153    return n
154 
155@app.get("/analytics/{newsletter_id}")
156def get_analytics(newsletter_id: int, authorization: str = Header(None)):
157    user_id = get_current_user(authorization)
158    n = newsletters.get(newsletter_id)
159    if not n:
160        raise HTTPException(status_code=404, detail="Newsletter not found")
161    if n["user_id"] != user_id:
162        raise HTTPException(status_code=403, detail="Not your newsletter")
163    return analytics.get(newsletter_id, {})

1fastapi
2uvicorn
3pydantic