Browse the data

1import os
2import secrets
3import hashlib
4from fastapi import FastAPI, HTTPException, Header
5from fastapi.responses import FileResponse
6from pydantic import BaseModel
7import uvicorn
8 
9app = FastAPI()
10 
11users = {}
12tokens = {}
13products_store = {}
14next_user_id = 1
15next_product_id = 1
16 
17class SignupRequest(BaseModel):
18    username: str
19    password: str
20 
21class LoginRequest(BaseModel):
22    username: str
23    password: str
24 
25class ProductCreate(BaseModel):
26    name: str
27    price: float
28 
29def hash_password(password: str) -> str:
30    return hashlib.sha256(password.encode()).hexdigest()
31 
32def get_current_user(authorization: str = Header(None)):
33    if not authorization or not authorization.startswith("Bearer "):
34        raise HTTPException(status_code=401, detail="Invalid auth header")
35    token = authorization.split(" ")[1]
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="User already exists")
45    user_id = next_user_id
46    next_user_id += 1
47    users[req.username] = {"id": user_id, "username": req.username, "password": hash_password(req.password)}
48    token = secrets.token_hex(32)
49    tokens[token] = req.username
50    return {"user_id": user_id, "token": token}
51 
52@app.post("/login")
53def login(req: LoginRequest):
54    if req.username not in users:
55        raise HTTPException(status_code=401, detail="Invalid credentials")
56    if users[req.username]["password"] != hash_password(req.password):
57        raise HTTPException(status_code=401, detail="Invalid credentials")
58    token = secrets.token_hex(32)
59    tokens[token] = req.username
60    return {"token": token}
61 
62@app.post("/product")
63def create_product(req: ProductCreate, authorization: str = Header(None)):
64    current_user = get_current_user(authorization)
65    global next_product_id
66    product_id = next_product_id
67    next_product_id += 1
68    products_store[product_id] = {"id": product_id, "name": req.name, "price": req.price, "owner": current_user}
69    return products_store[product_id]
70 
71@app.get("/product/{product_id}")
72def get_product(product_id: int, authorization: str = Header(None)):
73    current_user = get_current_user(authorization)
74    if product_id not in products_store:
75        raise HTTPException(status_code=404, detail="Product not found")
76    return products_store[product_id]
77 
78@app.get("/product/{sku}/images/{file}")
79def get_product_image(sku: str, file: str, authorization: str = Header(None)):
80    current_user = get_current_user(authorization)
81    image_path = f"/var/products/{sku}/{file}"
82    if not os.path.isfile(image_path):
83        raise HTTPException(status_code=404, detail="Image not found")
84    return FileResponse(image_path)
85 
86@app.get("/product/{sku}/images")
87def list_product_images(sku: str, authorization: str = Header(None)):
88    current_user = get_current_user(authorization)
89    image_dir = f"/var/products/{sku}"
90    if not os.path.isdir(image_dir):
91        raise HTTPException(status_code=404, detail="Product directory not found")
92    try:
93        files = os.listdir(image_dir)
94    except PermissionError:
95        raise HTTPException(status_code=403, detail="Cannot access product directory")
96    return {"sku": sku, "images": files}
97 
98if __name__ == "__main__":
99    uvicorn.run(app, host="127.0.0.1", port=8000)

1fastapi
2uvicorn