Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from typing import Optional
3from datetime import datetime
4import uuid
5 
6app = FastAPI()
7 
8users = {}
9weddings = {}
10tasks = {}
11vendors = {}
12guests = {}
13tokens = {}
14id_counters = {'users': 0, 'weddings': 0, 'tasks': 0, 'vendors': 0, 'guests': 0}
15 
16def get_current_user(authorization: Optional[str] = Header(None)):
17    if not authorization:
18        raise HTTPException(status_code=401, detail="Missing auth header")
19    token = authorization.replace("Bearer ", "")
20    user_id = tokens.get(token)
21    if not user_id:
22        raise HTTPException(status_code=401, detail="Invalid token")
23    return user_id
24 
25@app.post("/signup")
26def signup(username: str, password: str):
27    for uid, u in users.items():
28        if u['username'] == username:
29            raise HTTPException(status_code=400, detail="User exists")
30    id_counters['users'] += 1
31    user_id = id_counters['users']
32    users[user_id] = {'id': user_id, 'username': username, 'password': password}
33    return {'user_id': user_id, 'message': 'User created'}
34 
35@app.post("/login")
36def login(username: str, password: str):
37    for uid, u in users.items():
38        if u['username'] == username and u['password'] == password:
39            token = str(uuid.uuid4())
40            tokens[token] = uid
41            return {'token': token}
42    raise HTTPException(status_code=401, detail="Invalid credentials")
43 
44@app.get("/weddings/{wedding_id}")
45def get_wedding(wedding_id: int, authorization: Optional[str] = Header(None)):
46    get_current_user(authorization)
47    wedding = weddings.get(wedding_id)
48    if not wedding:
49        raise HTTPException(status_code=404, detail="Wedding not found")
50    return wedding
51 
52@app.post("/weddings")
53def create_wedding(name: str, date: str, authorization: Optional[str] = Header(None)):
54    user_id = get_current_user(authorization)
55    id_counters['weddings'] += 1
56    wedding_id = id_counters['weddings']
57    weddings[wedding_id] = {'id': wedding_id, 'name': name, 'date': date, 'user_id': user_id}
58    return weddings[wedding_id]
59 
60@app.get("/vendors/{vendor_id}")
61def get_vendor(vendor_id: int, authorization: Optional[str] = Header(None)):
62    get_current_user(authorization)
63    vendor = vendors.get(vendor_id)
64    if not vendor:
65        raise HTTPException(status_code=404, detail="Vendor not found")
66    return vendor
67 
68@app.post("/vendors")
69def create_vendor(name: str, category: str, wedding_id: int, authorization: Optional[str] = Header(None)):
70    get_current_user(authorization)
71    if wedding_id not in weddings:
72        raise HTTPException(status_code=404, detail="Wedding not found")
73    id_counters['vendors'] += 1
74    vendor_id = id_counters['vendors']
75    vendors[vendor_id] = {'id': vendor_id, 'name': name, 'category': category, 'wedding_id': wedding_id}
76    return vendors[vendor_id]
77 
78@app.get("/guests/{guest_id}")
79def get_guest(guest_id: int, authorization: Optional[str] = Header(None)):
80    get_current_user(authorization)
81    guest = guests.get(guest_id)
82    if not guest:
83        raise HTTPException(status_code=404, detail="Guest not found")
84    return guest
85 
86@app.post("/guests")
87def create_guest(name: str, email: str, wedding_id: int, authorization: Optional[str] = Header(None)):
88    get_current_user(authorization)
89    if wedding_id not in weddings:
90        raise HTTPException(status_code=404, detail="Wedding not found")
91    id_counters['guests'] += 1
92    guest_id = id_counters['guests']
93    guests[guest_id] = {'id': guest_id, 'name': name, 'email': email, 'wedding_id': wedding_id}
94    return guests[guest_id]
95 
96@app.get("/tasks/{task_id}")
97def get_task(task_id: int, authorization: Optional[str] = Header(None)):
98    get_current_user(authorization)
99    task = tasks.get(task_id)
100    if not task:
101        raise HTTPException(status_code=404, detail="Task not found")
102    return task
103 
104@app.post("/tasks")
105def create_task(title: str, wedding_id: int, assigned_vendor_id: Optional[int] = None, budget: Optional[float] = None, authorization: Optional[str] = Header(None)):
106    get_current_user(authorization)
107    if wedding_id not in weddings:
108        raise HTTPException(status_code=404, detail="Wedding not found")
109    id_counters['tasks'] += 1
110    task_id = id_counters['tasks']
111    tasks[task_id] = {
112        'id': task_id,
113        'title': title,
114        'wedding_id': wedding_id,
115        'assigned_vendor_id': assigned_vendor_id,
116        'budget': budget,
117        'completed': False
118    }
119    return tasks[task_id]
120 
121@app.get("/weddings/{wedding_id}/tasks")
122def get_wedding_tasks(wedding_id: int, authorization: Optional[str] = Header(None)):
123    get_current_user(authorization)
124    if wedding_id not in weddings:
125        raise HTTPException(status_code=404, detail="Wedding not found")
126    wedding_tasks = []
127    for t in tasks.values():
128        if t['wedding_id'] == wedding_id:
129            vendor_name = None
130            if t['assigned_vendor_id'] and t['assigned_vendor_id'] in vendors:
131                vendor_name = vendors[t['assigned_vendor_id']]['name']
132            wedding_tasks.append({
133                'id': t['id'],
134                'title': t['title'],
135                'completed': t['completed'],
136                'assigned_vendor': vendor_name,
137                'budget': t['budget']
138            })
139    return wedding_tasks
140 
141@app.post("/tasks/update")
142def update_task(task_id: int, completed: bool, authorization: Optional[str] = Header(None)):
143    user_id = get_current_user(authorization)
144    task = tasks.get(task_id)
145    if not task:
146        raise HTTPException(status_code=404, detail="Task not found")
147    task['completed'] = completed
148    
149    if completed:
150        wedding = weddings.get(task['wedding_id'])
151        if wedding:
152            notified_guests = []
153            for g in guests.values():
154                if g['wedding_id'] == task['wedding_id']:
155                    notified_guests.append({'name': g['name'], 'email': g['email']})
156            return {
157                'message': f"Task '{task['title']}' marked complete. Notified {len(notified_guests)} guests.",
158                'notified_guests': notified_guests
159            }
160    return {'message': f"Task '{task['title']}' updated"}

1fastapi
2uvicorn