Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import random
5import string
6import time
7 
8app = FastAPI()
9 
10# In-memory stores
11users = {}
12tokens = {}
13listings = {}
14schedules = {}
15neighborhood_stats = {}
16id_counter = {"users": 0, "listings": 0, "schedules": 0, "neighborhood_stats": 0}
17 
18# Seed some data
19listings[1] = {"id": 1, "price": 450000, "sqft": 1500, "walk_score": 85}
20listings[2] = {"id": 2, "price": 320000, "sqft": 1100, "walk_score": 72}
21listings[3] = {"id": 3, "price": 680000, "sqft": 2200, "walk_score": 90}
22id_counter["listings"] = 3
23 
24neighborhood_stats[1] = {"id": 1, "name": "Downtown", "crime_rate": 4.2, "school_rating": 7.5, "avg_commute": 22}
25neighborhood_stats[2] = {"id": 2, "name": "Suburbia", "crime_rate": 1.8, "school_rating": 8.9, "avg_commute": 35}
26neighborhood_stats[3] = {"id": 3, "name": "Uptown", "crime_rate": 3.1, "school_rating": 8.2, "avg_commute": 28}
27id_counter["neighborhood_stats"] = 3
28 
29# Models
30class UserCreate(BaseModel):
31    username: str
32    password: str
33 
34class LoginRequest(BaseModel):
35    username: str
36    password: str
37 
38class ListingCreate(BaseModel):
39    price: float
40    sqft: float
41    walk_score: int
42 
43class ScheduleCreate(BaseModel):
44    listing_id: int
45    time_slot: str
46 
47class NeighborhoodStatsCreate(BaseModel):
48    name: str
49    crime_rate: float
50    school_rating: float
51    avg_commute: float
52 
53def generate_token():
54    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
55 
56def get_current_user(authorization: Optional[str] = Header(None)):
57    if not authorization or not authorization.startswith("Bearer "):
58        raise HTTPException(status_code=401, detail="Invalid auth header")
59    token = authorization.split(" ")[1]
60    for uid, t in tokens.items():
61        if t == token:
62            return users[uid]
63    raise HTTPException(status_code=401, detail="Invalid token")
64 
65# Auth endpoints
66@app.post("/signup")
67def signup(user: UserCreate):
68    global id_counter
69    if any(u["username"] == user.username for u in users.values()):
70        raise HTTPException(status_code=400, detail="Username already exists")
71    id_counter["users"] += 1
72    uid = id_counter["users"]
73    users[uid] = {"id": uid, "username": user.username, "password": user.password}
74    token = generate_token()
75    tokens[uid] = token
76    return {"user_id": uid, "token": token}
77 
78@app.post("/login")
79def login(req: LoginRequest):
80    for uid, u in users.items():
81        if u["username"] == req.username and u["password"] == req.password:
82            token = generate_token()
83            tokens[uid] = token
84            return {"user_id": uid, "token": token}
85    raise HTTPException(status_code=401, detail="Invalid credentials")
86 
87# Listing endpoints
88@app.get("/listings")
89def get_listings():
90    return list(listings.values())
91 
92@app.get("/listings/{listing_id}")
93def get_listing(listing_id: int):
94    if listing_id not in listings:
95        raise HTTPException(status_code=404, detail="Listing not found")
96    return listings[listing_id]
97 
98@app.post("/listings")
99def create_listing(listing: ListingCreate, authorization: Optional[str] = Header(None)):
100    get_current_user(authorization)
101    global id_counter
102    id_counter["listings"] += 1
103    lid = id_counter["listings"]
104    listings[lid] = {"id": lid, "price": listing.price, "sqft": listing.sqft, "walk_score": listing.walk_score}
105    return listings[lid]
106 
107# Schedule viewing endpoints
108@app.post("/schedule-viewing")
109def schedule_viewing(schedule: ScheduleCreate, authorization: Optional[str] = Header(None)):
110    user = get_current_user(authorization)
111    if schedule.listing_id not in listings:
112        raise HTTPException(status_code=404, detail="Listing not found")
113    global id_counter
114    id_counter["schedules"] += 1
115    sid = id_counter["schedules"]
116    schedules[sid] = {"id": sid, "user_id": user["id"], "listing_id": schedule.listing_id, "time_slot": schedule.time_slot}
117    # Send confirmation (simulated)
118    print(f"Confirmation: Viewing scheduled for listing {schedule.listing_id} at {schedule.time_slot}")
119    return schedules[sid]
120 
121@app.get("/schedule-viewing/{schedule_id}")
122def get_schedule(schedule_id: int, authorization: Optional[str] = Header(None)):
123    get_current_user(authorization)
124    if schedule_id not in schedules:
125        raise HTTPException(status_code=404, detail="Schedule not found")
126    return schedules[schedule_id]
127 
128# Neighborhood stats endpoints
129@app.get("/neighborhood-stats")
130def get_neighborhood_stats():
131    return list(neighborhood_stats.values())
132 
133@app.get("/neighborhood-stats/{stat_id}")
134def get_neighborhood_stat(stat_id: int):
135    if stat_id not in neighborhood_stats:
136        raise HTTPException(status_code=404, detail="Neighborhood stat not found")
137    return neighborhood_stats[stat_id]
138 
139@app.post("/neighborhood-stats")
140def create_neighborhood_stat(stat: NeighborhoodStatsCreate, authorization: Optional[str] = Header(None)):
141    get_current_user(authorization)
142    global id_counter
143    id_counter["neighborhood_stats"] += 1
144    nid = id_counter["neighborhood_stats"]
145    neighborhood_stats[nid] = {"id": nid, "name": stat.name, "crime_rate": stat.crime_rate, "school_rating": stat.school_rating, "avg_commute": stat.avg_commute}
146    return neighborhood_stats[nid]

1fastapi
2uvicorn