Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List
4import uuid
5import time
6import math
7 
8app = FastAPI()
9 
10users = {}
11tokens = {}
12listings = {}
13listing_id_counter = 1
14 
15class SignupRequest(BaseModel):
16    username: str
17    password: str
18 
19class LoginRequest(BaseModel):
20    username: str
21    password: str
22 
23class ListingCreate(BaseModel):
24    title: str
25    price: float
26    bedrooms: int
27    sqft: float
28    days_on_market: int = 0
29 
30class ListingBulk(BaseModel):
31    listings: List[ListingCreate]
32 
33class ListingFilter(BaseModel):
34    min_price: Optional[float] = None
35    max_price: Optional[float] = None
36    min_bedrooms: Optional[int] = None
37    max_bedrooms: Optional[int] = None
38    min_sqft: Optional[float] = None
39    max_sqft: Optional[float] = None
40 
41def get_user_id_from_token(authorization: str = Header(...)):
42    if not authorization.startswith("Bearer "):
43        raise HTTPException(status_code=401, detail="Invalid auth header")
44    token = authorization[7:]
45    if token not in tokens:
46        raise HTTPException(status_code=401, detail="Invalid token")
47    return tokens[token]
48 
49@app.post("/signup")
50def signup(req: SignupRequest):
51    if req.username in users:
52        raise HTTPException(status_code=400, detail="User already exists")
53    users[req.username] = {"password": req.password}
54    return {"message": "User created"}
55 
56@app.post("/login")
57def login(req: LoginRequest):
58    if req.username not in users or users[req.username]["password"] != req.password:
59        raise HTTPException(status_code=401, detail="Invalid credentials")
60    token = str(uuid.uuid4())
61    tokens[token] = req.username
62    return {"token": token}
63 
64@app.post("/listings")
65def create_listing(listing: ListingCreate, authorization: str = Header(...)):
66    user_id = get_user_id_from_token(authorization)
67    global listing_id_counter
68    listing_id = listing_id_counter
69    listing_id_counter += 1
70    listings[listing_id] = {
71        "id": listing_id,
72        "title": listing.title,
73        "price": listing.price,
74        "bedrooms": listing.bedrooms,
75        "sqft": listing.sqft,
76        "days_on_market": listing.days_on_market,
77        "price_per_sqft": round(listing.price / listing.sqft, 2) if listing.sqft > 0 else 0,
78        "created_by": user_id
79    }
80    return listings[listing_id]
81 
82@app.post("/listings/bulk")
83def bulk_upload(bulk: ListingBulk, authorization: str = Header(...)):
84    user_id = get_user_id_from_token(authorization)
85    global listing_id_counter
86    created = []
87    for listing in bulk.listings:
88        listing_id = listing_id_counter
89        listing_id_counter += 1
90        listings[listing_id] = {
91            "id": listing_id,
92            "title": listing.title,
93            "price": listing.price,
94            "bedrooms": listing.bedrooms,
95            "sqft": listing.sqft,
96            "days_on_market": listing.days_on_market,
97            "price_per_sqft": round(listing.price / listing.sqft, 2) if listing.sqft > 0 else 0,
98            "created_by": user_id
99        }
100        created.append(listings[listing_id])
101    return created
102 
103@app.get("/listings/{listing_id}")
104def get_listing(listing_id: int, authorization: str = Header(...)):
105    user_id = get_user_id_from_token(authorization)
106    if listing_id not in listings:
107        raise HTTPException(status_code=404, detail="Listing not found")
108    return listings[listing_id]
109 
110@app.get("/listings")
111def search_listings(
112    min_price: Optional[float] = None,
113    max_price: Optional[float] = None,
114    bedrooms: Optional[int] = None,
115    min_sqft: Optional[float] = None,
116    max_sqft: Optional[float] = None,
117    authorization: str = Header(...)
118):
119    user_id = get_user_id_from_token(authorization)
120    results = []
121    for lid, listing in listings.items():
122        if min_price is not None and listing["price"] < min_price:
123            continue
124        if max_price is not None and listing["price"] > max_price:
125            continue
126        if bedrooms is not None and listing["bedrooms"] != bedrooms:
127            continue
128        if min_sqft is not None and listing["sqft"] < min_sqft:
129            continue
130        if max_sqft is not None and listing["sqft"] > max_sqft:
131            continue
132        results.append(listing)
133    results.sort(key=lambda x: (x["price_per_sqft"], x["days_on_market"]))
134    return results

1fastapi
2uvicorn