Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5 
6app = FastAPI()
7 
8users = {}
9tokens = {}
10events = {}
11tasks = {}
12vendors = {}
13event_id_counter = 1
14task_id_counter = 1
15vendor_id_counter = 1
16user_id_counter = 1
17 
18def get_current_user(authorization: str = Header(None)):
19    if not authorization:
20        raise HTTPException(status_code=401, detail="No auth token")
21    token = authorization.replace("Bearer ", "")
22    for uid, t in tokens.items():
23        if t == token:
24            return uid
25    raise HTTPException(status_code=401, detail="Invalid token")
26 
27class SignupRequest(BaseModel):
28    username: str
29    password: str
30 
31class LoginRequest(BaseModel):
32    username: str
33    password: str
34 
35class EventCreate(BaseModel):
36    name: str
37    date: str
38    venue: str
39    guest_count: int
40    budget: float
41 
42class EventUpdate(BaseModel):
43    date: Optional[str] = None
44    venue: Optional[str] = None
45    guest_count: Optional[int] = None
46    budget: Optional[float] = None
47 
48class TaskCreate(BaseModel):
49    name: str
50    assigned_to: str
51    deadline: str
52    status: str = "pending"
53 
54class VendorCreate(BaseModel):
55    name: str
56    service: str
57    contact: str
58    price: float
59 
60@app.post("/signup")
61def signup(req: SignupRequest):
62    global user_id_counter
63    for u in users.values():
64        if u["username"] == req.username:
65            raise HTTPException(status_code=400, detail="Username exists")
66    uid = user_id_counter
67    users[uid] = {"id": uid, "username": req.username, "password": req.password}
68    user_id_counter += 1
69    return {"id": uid, "username": req.username}
70 
71@app.post("/login")
72def login(req: LoginRequest):
73    for uid, u in users.items():
74        if u["username"] == req.username and u["password"] == req.password:
75            token = secrets.token_hex(16)
76            tokens[uid] = token
77            return {"token": token}
78    raise HTTPException(status_code=401, detail="Invalid credentials")
79 
80@app.post("/events")
81def create_event(event: EventCreate, authorization: str = Header(None)):
82    get_current_user(authorization)
83    global event_id_counter
84    eid = event_id_counter
85    events[eid] = {
86        "id": eid,
87        "name": event.name,
88        "date": event.date,
89        "venue": event.venue,
90        "guest_count": event.guest_count,
91        "budget": event.budget,
92        "tasks": [],
93        "vendors": []
94    }
95    event_id_counter += 1
96    return events[eid]
97 
98@app.get("/events/{event_id}")
99def get_event(event_id: int, authorization: str = Header(None)):
100    get_current_user(authorization)
101    if event_id not in events:
102        raise HTTPException(status_code=404, detail="Event not found")
103    return events[event_id]
104 
105@app.put("/events/{event_id}")
106def update_event(event_id: int, update: EventUpdate, authorization: str = Header(None)):
107    get_current_user(authorization)
108    if event_id not in events:
109        raise HTTPException(status_code=404, detail="Event not found")
110    e = events[event_id]
111    if update.date is not None:
112        e["date"] = update.date
113    if update.venue is not None:
114        e["venue"] = update.venue
115    if update.guest_count is not None:
116        e["guest_count"] = update.guest_count
117    if update.budget is not None:
118        e["budget"] = update.budget
119    return e
120 
121@app.post("/tasks")
122def create_task(task: TaskCreate, authorization: str = Header(None)):
123    get_current_user(authorization)
124    global task_id_counter
125    tid = task_id_counter
126    tasks[tid] = {
127        "id": tid,
128        "name": task.name,
129        "assigned_to": task.assigned_to,
130        "deadline": task.deadline,
131        "status": task.status
132    }
133    task_id_counter += 1
134    return tasks[tid]
135 
136@app.get("/tasks/{task_id}")
137def get_task(task_id: int, authorization: str = Header(None)):
138    get_current_user(authorization)
139    if task_id not in tasks:
140        raise HTTPException(status_code=404, detail="Task not found")
141    return tasks[task_id]
142 
143@app.post("/vendors")
144def create_vendor(vendor: VendorCreate, authorization: str = Header(None)):
145    get_current_user(authorization)
146    global vendor_id_counter
147    vid = vendor_id_counter
148    vendors[vid] = {
149        "id": vid,
150        "name": vendor.name,
151        "service": vendor.service,
152        "contact": vendor.contact,
153        "price": vendor.price
154    }
155    vendor_id_counter += 1
156    return vendors[vid]
157 
158@app.get("/vendors/{vendor_id}")
159def get_vendor(vendor_id: int, authorization: str = Header(None)):
160    get_current_user(authorization)
161    if vendor_id not in vendors:
162        raise HTTPException(status_code=404, detail="Vendor not found")
163    return vendors[vendor_id]

1fastapi
2uvicorn