Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, Dict
4import secrets
5 
6app = FastAPI()
7 
8users: Dict[int, dict] = {}
9agents: Dict[int, dict] = {}
10next_user_id = 1
11next_agent_id = 1
12tokens: Dict[str, int] = {}
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_user_id_from_token(authorization: str = Header(...)) -> int:
35    if not authorization.startswith("Bearer "):
36        raise HTTPException(status_code=401, detail="Invalid token")
37    token = authorization.split(" ")[1]
38    user_id = tokens.get(token)
39    if user_id is None:
40        raise HTTPException(status_code=401, detail="Invalid token")
41    return user_id
42 
43@app.post("/signup")
44def signup(req: SignupRequest):
45    global next_user_id
46    user_id = next_user_id
47    next_user_id += 1
48    users[user_id] = {"id": user_id, "username": req.username, "password": req.password}
49    return {"id": user_id, "username": req.username}
50 
51@app.post("/login")
52def login(req: LoginRequest):
53    for u in users.values():
54        if u["username"] == req.username and u["password"] == req.password:
55            token = secrets.token_hex(32)
56            tokens[token] = u["id"]
57            return {"token": token}
58    raise HTTPException(status_code=401, detail="Invalid credentials")
59 
60@app.get("/agents/{agent_id}")
61def get_agent(agent_id: int, authorization: str = Header(...)):
62    get_user_id_from_token(authorization)
63    agent = agents.get(agent_id)
64    if not agent:
65        raise HTTPException(status_code=404, detail="Agent not found")
66    return agent
67 
68@app.post("/agents")
69def create_agent(agent: AgentCreate, authorization: str = Header(...)):
70    global next_agent_id
71    get_user_id_from_token(authorization)
72    agent_id = next_agent_id
73    next_agent_id += 1
74    agents[agent_id] = {
75        "id": agent_id,
76        "name": agent.name,
77        "license_number": agent.license_number,
78        "service_area": agent.service_area,
79        "commission_rate": agent.commission_rate
80    }
81    return agents[agent_id]
82 
83@app.patch("/agents/{agent_id}")
84def update_agent(agent_id: int, update: AgentUpdate, authorization: str = Header(...)):
85    get_user_id_from_token(authorization)
86    agent = agents.get(agent_id)
87    if not agent:
88        raise HTTPException(status_code=404, detail="Agent not found")
89    if update.name is not None:
90        agent["name"] = update.name
91    if update.license_number is not None:
92        agent["license_number"] = update.license_number
93    if update.service_area is not None:
94        agent["service_area"] = update.service_area
95    if update.commission_rate is not None:
96        agent["commission_rate"] = update.commission_rate
97    return agent

1fastapi
2uvicorn