Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import secrets
5 
6app = FastAPI()
7 
8users = {}
9agents = {}
10tokens = {}
11user_id_counter = 1
12agent_id_counter = 1
13 
14class SignupRequest(BaseModel):
15    username: str
16    password: str
17 
18class LoginRequest(BaseModel):
19    username: str
20    password: str
21 
22class AgentCreate(BaseModel):
23    name: str
24    license_number: str
25    service_area: str
26    commission_rate: float
27 
28class AgentUpdate(BaseModel):
29    name: Optional[str] = None
30    license_number: Optional[str] = None
31    service_area: Optional[str] = None
32    commission_rate: Optional[float] = None
33 
34def get_current_user(authorization: str = Header(...)):
35    if not authorization.startswith("Bearer "):
36        raise HTTPException(status_code=401, detail="Invalid auth header")
37    token = authorization[7:]
38    if token not in tokens:
39        raise HTTPException(status_code=401, detail="Invalid token")
40    return tokens[token]
41 
42@app.post("/signup")
43def signup(req: SignupRequest):
44    global user_id_counter
45    if req.username in users:
46        raise HTTPException(status_code=400, detail="Username already exists")
47    user_id = user_id_counter
48    user_id_counter += 1
49    users[req.username] = {"id": user_id, "username": req.username, "password": req.password}
50    return {"id": user_id, "username": req.username}
51 
52@app.post("/login")
53def login(req: LoginRequest):
54    if req.username not in users or users[req.username]["password"] != req.password:
55        raise HTTPException(status_code=401, detail="Invalid credentials")
56    token = secrets.token_hex(16)
57    tokens[token] = req.username
58    return {"token": token}
59 
60@app.post("/agents")
61def create_agent(req: AgentCreate, authorization: str = Header(...)):
62    global agent_id_counter
63    get_current_user(authorization)
64    agent_id = agent_id_counter
65    agent_id_counter += 1
66    agents[agent_id] = {
67        "id": agent_id,
68        "name": req.name,
69        "license_number": req.license_number,
70        "service_area": req.service_area,
71        "commission_rate": req.commission_rate
72    }
73    return agents[agent_id]
74 
75@app.get("/agents/{agent_id}")
76def get_agent(agent_id: int, authorization: str = Header(...)):
77    get_current_user(authorization)
78    if agent_id not in agents:
79        raise HTTPException(status_code=404, detail="Agent not found")
80    return agents[agent_id]
81 
82@app.patch("/agents/{agent_id}")
83def update_agent(agent_id: int, req: AgentUpdate, authorization: str = Header(...)):
84    get_current_user(authorization)
85    if agent_id not in agents:
86        raise HTTPException(status_code=404, detail="Agent not found")
87    agent = agents[agent_id]
88    if req.name is not None:
89        agent["name"] = req.name
90    if req.license_number is not None:
91        agent["license_number"] = req.license_number
92    if req.service_area is not None:
93        agent["service_area"] = req.service_area
94    if req.commission_rate is not None:
95        agent["commission_rate"] = req.commission_rate
96    return agent

1fastapi
2uvicorn