Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional
4import uuid
5import hashlib
6import secrets
7 
8app = FastAPI()
9 
10users = {}
11tokens = {}
12portfolios = {}
13next_user_id = 1
14next_portfolio_id = 1
15 
16class SignupRequest(BaseModel):
17    username: str
18    password: str
19 
20class LoginRequest(BaseModel):
21    username: str
22    password: str
23 
24class PortfolioCreate(BaseModel):
25    name: str
26    risk_tolerance: str = "moderate"
27    rebalancing_frequency: str = "quarterly"
28 
29class PortfolioUpdate(BaseModel):
30    name: Optional[str] = None
31    risk_tolerance: Optional[str] = None
32    rebalancing_frequency: Optional[str] = None
33    subscription_tier: Optional[str] = None
34    role: Optional[str] = None
35 
36def get_current_user(authorization: str = Header(...)):
37    token = authorization.replace("Bearer ", "")
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 next_user_id
45    user_id = next_user_id
46    next_user_id += 1
47    salt = secrets.token_hex(16)
48    hashed = hashlib.sha256((req.password + salt).encode()).hexdigest()
49    users[user_id] = {
50        "id": user_id,
51        "username": req.username,
52        "password_hash": hashed,
53        "salt": salt
54    }
55    return {"user_id": user_id, "username": req.username}
56 
57@app.post("/login")
58def login(req: LoginRequest):
59    for uid, user in users.items():
60        if user["username"] == req.username:
61            hash_check = hashlib.sha256((req.password + user["salt"]).encode()).hexdigest()
62            if hash_check == user["password_hash"]:
63                token = secrets.token_hex(32)
64                tokens[token] = uid
65                return {"token": token}
66    raise HTTPException(status_code=401, detail="Invalid credentials")
67 
68@app.post("/portfolios")
69def create_portfolio(req: PortfolioCreate, authorization: str = Header(...)):
70    global next_portfolio_id
71    user_id = get_current_user(authorization)
72    portfolio_id = next_portfolio_id
73    next_portfolio_id += 1
74    portfolios[portfolio_id] = {
75        "id": portfolio_id,
76        "owner_id": user_id,
77        "name": req.name,
78        "risk_tolerance": req.risk_tolerance,
79        "rebalancing_frequency": req.rebalancing_frequency,
80        "subscription_tier": "basic",
81        "role": "owner"
82    }
83    return portfolios[portfolio_id]
84 
85@app.get("/portfolios/{portfolio_id}")
86def get_portfolio(portfolio_id: int, authorization: str = Header(...)):
87    user_id = get_current_user(authorization)
88    if portfolio_id not in portfolios:
89        raise HTTPException(status_code=404, detail="Portfolio not found")
90    return portfolios[portfolio_id]
91 
92@app.put("/portfolios/{portfolio_id}")
93def update_portfolio(portfolio_id: int, req: PortfolioUpdate, authorization: str = Header(...)):
94    user_id = get_current_user(authorization)
95    if portfolio_id not in portfolios:
96        raise HTTPException(status_code=404, detail="Portfolio not found")
97    portfolio = portfolios[portfolio_id]
98    if req.name is not None:
99        portfolio["name"] = req.name
100    if req.risk_tolerance is not None:
101        portfolio["risk_tolerance"] = req.risk_tolerance
102    if req.rebalancing_frequency is not None:
103        portfolio["rebalancing_frequency"] = req.rebalancing_frequency
104    if req.subscription_tier is not None:
105        portfolio["subscription_tier"] = req.subscription_tier
106    if req.role is not None:
107        portfolio["role"] = req.role
108    return portfolio

1fastapi
2uvicorn