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 = {}
10reviews = {}
11products = {}
12next_user_id = 1
13next_review_id = 1
14next_product_id = 1
15 
16class SignupRequest(BaseModel):
17    username: str
18    password: str
19 
20class LoginRequest(BaseModel):
21    username: str
22    password: str
23 
24class ReviewCreate(BaseModel):
25    product_id: int
26    rating: int
27    text: str
28 
29class ProductCreate(BaseModel):
30    name: str
31 
32def get_current_user(authorization: Optional[str] = Header(None)):
33    if not authorization:
34        raise HTTPException(status_code=401, detail="Missing auth header")
35    token = authorization.replace("Bearer ", "")
36    if token not in tokens:
37        raise HTTPException(status_code=401, detail="Invalid token")
38    return tokens[token]
39 
40@app.post("/signup")
41def signup(req: SignupRequest):
42    global next_user_id
43    if req.username in users:
44        raise HTTPException(status_code=400, detail="Username taken")
45    user_id = next_user_id
46    next_user_id += 1
47    users[req.username] = {"id": user_id, "password": req.password}
48    return {"id": user_id, "username": req.username}
49 
50@app.post("/login")
51def login(req: LoginRequest):
52    if req.username not in users or users[req.username]["password"] != req.password:
53        raise HTTPException(status_code=401, detail="Invalid credentials")
54    token = secrets.token_hex(16)
55    tokens[token] = req.username
56    return {"token": token}
57 
58@app.post("/reviews")
59def create_review(review: ReviewCreate, authorization: Optional[str] = Header(None)):
60    current_user = get_current_user(authorization)
61    global next_review_id
62    review_id = next_review_id
63    next_review_id += 1
64    reviews[review_id] = {
65        "id": review_id,
66        "product_id": review.product_id,
67        "rating": review.rating,
68        "text": review.text,
69        "author": current_user
70    }
71    return reviews[review_id]
72 
73@app.get("/reviews/{review_id}")
74def get_review(review_id: int):
75    if review_id not in reviews:
76        raise HTTPException(status_code=404, detail="Review not found")
77    return reviews[review_id]
78 
79@app.post("/products")
80def create_product(product: ProductCreate, authorization: Optional[str] = Header(None)):
81    current_user = get_current_user(authorization)
82    global next_product_id
83    product_id = next_product_id
84    next_product_id += 1
85    products[product_id] = {
86        "id": product_id,
87        "name": product.name,
88        "seller": current_user
89    }
90    return products[product_id]
91 
92@app.get("/products/{product_id}")
93def get_product(product_id: int):
94    if product_id not in products:
95        raise HTTPException(status_code=404, detail="Product not found")
96    return products[product_id]
97 
98@app.get("/products/{product_id}/reviews")
99def get_product_reviews(product_id: int):
100    return [r for r in reviews.values() if r["product_id"] == product_id]
101 
102@app.get("/users/{user_id}")
103def get_user(user_id: int):
104    for u in users.values():
105        if u["id"] == user_id:
106            return u
107    raise HTTPException(status_code=404, detail="User not found")

1fastapi
2uvicorn