Browse the data

1from fastapi import FastAPI, HTTPException, Header, UploadFile, File
2from fastapi.responses import FileResponse
3import os
4import shutil
5import uuid
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11listings = {}
12next_user_id = 1
13next_listing_id = 1
14 
15UPLOAD_DIR = "marketplace_uploads"
16 
17if not os.path.exists(UPLOAD_DIR):
18    os.makedirs(UPLOAD_DIR)
19 
20@app.post("/signup")
21def signup(username: str, password: str):
22    global next_user_id
23    for u in users.values():
24        if u["username"] == username:
25            raise HTTPException(400, "Username taken")
26    user_id = next_user_id
27    next_user_id += 1
28    users[user_id] = {"id": user_id, "username": username, "password": password, "role": "seller"}
29    return {"user_id": user_id, "username": username}
30 
31@app.post("/login")
32def login(username: str, password: str):
33    for u in users.values():
34        if u["username"] == username and u["password"] == password:
35            token = str(uuid.uuid4())
36            tokens[token] = u["id"]
37            return {"token": token}
38    raise HTTPException(401, "Invalid credentials")
39 
40def get_current_user(authorization: str = Header(None)):
41    if not authorization:
42        raise HTTPException(401, "Missing auth header")
43    token = authorization.replace("Bearer ", "")
44    user_id = tokens.get(token)
45    if not user_id:
46        raise HTTPException(401, "Invalid token")
47    return users[user_id]
48 
49@app.post("/listings")
50def create_listing(title: str, price: float, authorization: str = Header(None)):
51    user = get_current_user(authorization)
52    global next_listing_id
53    listing_id = next_listing_id
54    next_listing_id += 1
55    listing_dir = os.path.join(UPLOAD_DIR, str(listing_id))
56    os.makedirs(listing_dir, exist_ok=True)
57    listings[listing_id] = {"id": listing_id, "title": title, "price": price, "seller_id": user["id"]}
58    return {"listing_id": listing_id, "title": title, "price": price}
59 
60@app.get("/listings/{listing_id}")
61def get_listing(listing_id: int):
62    listing = listings.get(listing_id)
63    if not listing:
64        raise HTTPException(404, "Listing not found")
65    return listing
66 
67@app.post("/listings/{listing_id}/media")
68async def upload_media(listing_id: int, file: UploadFile = File(...), authorization: str = Header(None)):
69    user = get_current_user(authorization)
70    if user["role"] != "seller":
71        raise HTTPException(403, "Only sellers can upload")
72    listing = listings.get(listing_id)
73    if not listing:
74        raise HTTPException(404, "Listing not found")
75    if listing["seller_id"] != user["id"]:
76        raise HTTPException(403, "Not your listing")
77    listing_dir = os.path.join(UPLOAD_DIR, str(listing_id))
78    os.makedirs(listing_dir, exist_ok=True)
79    file_path = os.path.join(listing_dir, file.filename)
80    with open(file_path, "wb") as f:
81        shutil.copyfileobj(file.file, f)
82    return {"file_name": file.filename}
83 
84@app.get("/listings/{listing_id}/media/{file_name}")
85def get_media(listing_id: int, file_name: str, authorization: str = Header(None)):
86    user = get_current_user(authorization)
87    if user["role"] not in ["buyer", "seller"]:
88        raise HTTPException(403, "Access denied")
89    file_path = os.path.join(UPLOAD_DIR, str(listing_id), file_name)
90    if not os.path.exists(file_path):
91        raise HTTPException(404, "File not found")
92    return FileResponse(file_path)

1fastapi
2uvicorn
3python-multipart