Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, Dict, Any
4import secrets
5 
6app = FastAPI()
7 
8users = {}
9tokens = {}
10listings = {}
11listing_id_counter = 1
12 
13class UserCreate(BaseModel):
14    email: str
15    password: str
16 
17class UserLogin(BaseModel):
18    email: str
19    password: str
20 
21class ListingCreate(BaseModel):
22    name: str
23    price: float
24    status: str
25    agent_name: str
26    agent_phone: Optional[str] = None
27    extra: Optional[Dict[str, Any]] = {}
28 
29class ListingUpdate(BaseModel):
30    name: Optional[str] = None
31    price: Optional[float] = None
32    status: Optional[str] = None
33    agent_name: Optional[str] = None
34    agent_phone: Optional[str] = None
35    extra: Optional[Dict[str, Any]] = {}
36 
37def get_current_user(authorization: str = Header(None)):
38    if not authorization:
39        raise HTTPException(status_code=401, detail="Missing auth header")
40    token = authorization.replace("Bearer ", "")
41    if token not in tokens:
42        raise HTTPException(status_code=401, detail="Invalid token")
43    return tokens[token]
44 
45@app.post("/signup")
46def signup(user: UserCreate):
47    if user.email in users:
48        raise HTTPException(status_code=400, detail="User already exists")
49    users[user.email] = {"email": user.email, "password": user.password}
50    return {"message": "User created"}
51 
52@app.post("/login")
53def login(user: UserLogin):
54    if user.email not in users or users[user.email]["password"] != user.password:
55        raise HTTPException(status_code=401, detail="Invalid credentials")
56    token = secrets.token_hex(16)
57    tokens[token] = user.email
58    return {"token": token}
59 
60@app.post("/listings")
61def create_listing(listing: ListingCreate, authorization: str = Header(None)):
62    get_current_user(authorization)
63    global listing_id_counter
64    listing_data = {
65        "id": listing_id_counter,
66        "name": listing.name,
67        "price": listing.price,
68        "status": listing.status,
69        "agent_name": listing.agent_name,
70        "agent_phone": listing.agent_phone,
71        "version": 1,
72        **listing.extra
73    }
74    listings[listing_id_counter] = listing_data
75    listing_id_counter += 1
76    return listing_data
77 
78@app.get("/listings/{listing_id}")
79def get_listing(listing_id: int, authorization: str = Header(None)):
80    get_current_user(authorization)
81    if listing_id not in listings:
82        raise HTTPException(status_code=404, detail="Listing not found")
83    return listings[listing_id]
84 
85@app.patch("/listings/{listing_id}")
86def update_listing(listing_id: int, update: ListingUpdate, authorization: str = Header(None)):
87    get_current_user(authorization)
88    if listing_id not in listings:
89        raise HTTPException(status_code=404, detail="Listing not found")
90    listing = listings[listing_id]
91    if update.name is not None:
92        listing["name"] = update.name
93    if update.price is not None:
94        listing["price"] = update.price
95    if update.status is not None:
96        listing["status"] = update.status
97    if update.agent_name is not None:
98        listing["agent_name"] = update.agent_name
99    if update.agent_phone is not None:
100        listing["agent_phone"] = update.agent_phone
101    if update.extra:
102        for key, value in update.extra.items():
103            listing[key] = value
104    listing["version"] += 1
105    return listing

1fastapi
2uvicorn