Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from fastapi.responses import JSONResponse
3from pydantic import BaseModel
4from typing import Optional
5import hashlib
6import secrets
7 
8app = FastAPI()
9 
10users = {}
11vendors = {}
12tokens = {}
13next_user_id = 1
14next_vendor_id = 1
15 
16def get_current_user(authorization: str = Header(None)):
17    if not authorization:
18        raise HTTPException(status_code=401, detail="No 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 
25class SignupRequest(BaseModel):
26    username: str
27    password: str
28 
29class LoginRequest(BaseModel):
30    username: str
31    password: str
32 
33class VendorCreate(BaseModel):
34    store_name: str
35    description: str
36    shipping_zones: list
37    commission_tier: str
38    admin_notes: str
39 
40class VendorUpdate(BaseModel):
41    store_name: Optional[str] = None
42    description: Optional[str] = None
43    shipping_zones: Optional[list] = None
44    commission_tier: Optional[str] = None
45    admin_notes: Optional[str] = None
46 
47@app.post("/signup")
48def signup(req: SignupRequest):
49    global next_user_id
50    for u in users.values():
51        if u["username"] == req.username:
52            raise HTTPException(status_code=400, detail="User exists")
53    user_id = next_user_id
54    next_user_id += 1
55    users[user_id] = {"id": user_id, "username": req.username, "password": hashlib.sha256(req.password.encode()).hexdigest()}
56    token = secrets.token_hex(32)
57    tokens[token] = user_id
58    return {"user_id": user_id, "token": token}
59 
60@app.post("/login")
61def login(req: LoginRequest):
62    for u in users.values():
63        if u["username"] == req.username and u["password"] == hashlib.sha256(req.password.encode()).hexdigest():
64            token = secrets.token_hex(32)
65            tokens[token] = u["id"]
66            return {"token": token}
67    raise HTTPException(status_code=401, detail="Invalid credentials")
68 
69@app.get("/vendors/{vendor_id}")
70def get_vendor(vendor_id: int, authorization: str = Header(None)):
71    current_user = get_current_user(authorization)
72    vendor = vendors.get(vendor_id)
73    if not vendor:
74        raise HTTPException(status_code=404, detail="Vendor not found")
75    return vendor
76 
77@app.post("/vendors")
78def create_vendor(req: VendorCreate, authorization: str = Header(None)):
79    global next_vendor_id
80    current_user = get_current_user(authorization)
81    vendor_id = next_vendor_id
82    next_vendor_id += 1
83    vendors[vendor_id] = {
84        "id": vendor_id,
85        "owner_id": current_user,
86        "store_name": req.store_name,
87        "description": req.description,
88        "shipping_zones": req.shipping_zones,
89        "commission_tier": req.commission_tier,
90        "admin_notes": req.admin_notes
91    }
92    return vendors[vendor_id]
93 
94@app.put("/vendors/{vendor_id}")
95def update_vendor(vendor_id: int, req: VendorUpdate, authorization: str = Header(None)):
96    current_user = get_current_user(authorization)
97    vendor = vendors.get(vendor_id)
98    if not vendor:
99        raise HTTPException(status_code=404, detail="Vendor not found")
100    if vendor["owner_id"] != current_user:
101        raise HTTPException(status_code=403, detail="Not your vendor")
102    if req.store_name is not None:
103        vendor["store_name"] = req.store_name
104    if req.description is not None:
105        vendor["description"] = req.description
106    if req.shipping_zones is not None:
107        vendor["shipping_zones"] = req.shipping_zones
108    if req.commission_tier is not None:
109        vendor["commission_tier"] = req.commission_tier
110    if req.admin_notes is not None:
111        vendor["admin_notes"] = req.admin_notes
112    return vendor

1fastapi
2uvicorn