Browse the data

1import uuid
2import random
3from fastapi import FastAPI, HTTPException, Header
4from pydantic import BaseModel
5from typing import Optional
6 
7app = FastAPI()
8 
9users = {}
10listings = {}
11swaps = {}
12tokens = {}
13next_user_id = 1
14next_listing_id = 1
15next_swap_id = 1
16 
17class SignupRequest(BaseModel):
18    username: str
19    password: str
20 
21class LoginRequest(BaseModel):
22    username: str
23    password: str
24 
25class ListingCreate(BaseModel):
26    title: str
27    condition: str
28    genre: str
29    owner_id: int
30 
31class SwapInitiate(BaseModel):
32    listing_id: int
33    requester_id: int
34 
35def get_current_user(authorization: str = Header(None)):
36    if not authorization:
37        raise HTTPException(status_code=401, detail="No auth header")
38    token = authorization.replace("Bearer ", "")
39    user_id = tokens.get(token)
40    if user_id is None:
41        raise HTTPException(status_code=401, detail="Invalid token")
42    return user_id
43 
44@app.post("/signup")
45def signup(req: SignupRequest):
46    global next_user_id
47    user_id = next_user_id
48    next_user_id += 1
49    users[user_id] = {
50        "id": user_id,
51        "username": req.username,
52        "password": req.password,
53        "swap_rating": 5.0,
54        "swap_count": 0
55    }
56    return {"user_id": user_id}
57 
58@app.post("/login")
59def login(req: LoginRequest):
60    for uid, u in users.items():
61        if u["username"] == req.username and u["password"] == req.password:
62            token = str(uuid.uuid4())
63            tokens[token] = uid
64            return {"token": token}
65    raise HTTPException(status_code=401, detail="Invalid credentials")
66 
67@app.get("/users/{user_id}")
68def get_user(user_id: int):
69    user = users.get(user_id)
70    if not user:
71        raise HTTPException(status_code=404, detail="User not found")
72    return user
73 
74@app.get("/listings/{listing_id}")
75def get_listing(listing_id: int):
76    listing = listings.get(listing_id)
77    if not listing:
78        raise HTTPException(status_code=404, detail="Listing not found")
79    return listing
80 
81@app.get("/swaps/{swap_id}")
82def get_swap(swap_id: int):
83    swap = swaps.get(swap_id)
84    if not swap:
85        raise HTTPException(status_code=404, detail="Swap not found")
86    return swap
87 
88@app.post("/users")
89def create_user(req: SignupRequest):
90    return signup(req)
91 
92@app.post("/listings")
93def create_listing(req: ListingCreate, authorization: str = Header(None)):
94    current_user = get_current_user(authorization)
95    global next_listing_id
96    lid = next_listing_id
97    next_listing_id += 1
98    listings[lid] = {
99        "id": lid,
100        "title": req.title,
101        "condition": req.condition,
102        "genre": req.genre,
103        "owner_id": req.owner_id,
104        "available": True
105    }
106    return listings[lid]
107 
108@app.get("/listings")
109def get_all_listings():
110    result = []
111    for lid, l in listings.items():
112        if l["available"]:
113            owner = users.get(l["owner_id"])
114            rating = owner["swap_rating"] if owner else 0
115            result.append({
116                "id": l["id"],
117                "title": l["title"],
118                "condition": l["condition"],
119                "genre": l["genre"],
120                "owner_id": l["owner_id"],
121                "owner_swap_rating": rating
122            })
123    return result
124 
125@app.post("/swap")
126def initiate_swap(req: SwapInitiate, authorization: str = Header(None)):
127    current_user = get_current_user(authorization)
128    listing = listings.get(req.listing_id)
129    if not listing:
130        raise HTTPException(status_code=404, detail="Listing not found")
131    if not listing["available"]:
132        raise HTTPException(status_code=400, detail="Listing not available")
133    if listing["owner_id"] == req.requester_id:
134        raise HTTPException(status_code=400, detail="Cannot swap with yourself")
135    owner = users.get(listing["owner_id"])
136    requester = users.get(req.requester_id)
137    if not owner or not requester:
138        raise HTTPException(status_code=404, detail="User not found")
139    global next_swap_id
140    sid = next_swap_id
141    next_swap_id += 1
142    listing["available"] = False
143    tracking = "TRACK" + str(random.randint(100000, 999999))
144    swaps[sid] = {
145        "id": sid,
146        "listing_id": req.listing_id,
147        "owner_id": listing["owner_id"],
148        "requester_id": req.requester_id,
149        "status": "pending",
150        "shipping_label": tracking
151    }
152    return swaps[sid]

1fastapi
2uvicorn