Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from fastapi.responses import Response
3import csv
4import io
5import os
6import random
7import string
8 
9app = FastAPI()
10 
11users = {}
12tokens = {}
13holdings = {}
14holdings_id_counter = 1
15 
16def generate_token():
17    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
18 
19def get_user_from_token(authorization: str = Header(None)):
20    if not authorization:
21        raise HTTPException(status_code=401, detail="Missing Authorization header")
22    token = authorization.replace("Bearer ", "")
23    if token not in tokens:
24        raise HTTPException(status_code=401, detail="Invalid token")
25    return tokens[token]
26 
27@app.post("/signup")
28def signup(username: str, password: str):
29    if username in users:
30        raise HTTPException(status_code=400, detail="User already exists")
31    user_id = len(users) + 1
32    users[username] = {"id": user_id, "password": password}
33    return {"user_id": user_id, "username": username}
34 
35@app.post("/login")
36def login(username: str, password: str):
37    if username not in users or users[username]["password"] != password:
38        raise HTTPException(status_code=401, detail="Invalid credentials")
39    token = generate_token()
40    tokens[token] = users[username]["id"]
41    return {"token": token}
42 
43@app.get("/holdings/{holding_id}")
44def get_holding(holding_id: int, authorization: str = Header(None)):
45    user_id = get_user_from_token(authorization)
46    if holding_id not in holdings:
47        raise HTTPException(status_code=404, detail="Holding not found")
48    return holdings[holding_id]
49 
50@app.post("/holdings")
51def create_holding(symbol: str, quantity: int, price: float, authorization: str = Header(None)):
52    global holdings_id_counter
53    user_id = get_user_from_token(authorization)
54    holding = {
55        "id": holdings_id_counter,
56        "user_id": user_id,
57        "symbol": symbol,
58        "quantity": quantity,
59        "price": price
60    }
61    holdings[holdings_id_counter] = holding
62    holdings_id_counter += 1
63    return holding
64 
65@app.get("/holdings/{user_id}/export")
66def export_holdings(user_id: int, format: str = "csv", authorization: str = Header(None)):
67    current_user = get_user_from_token(authorization)
68    if current_user != user_id:
69        raise HTTPException(status_code=403, detail="Not authorized")
70    
71    csv_path = f"/user_data/{user_id}/portfolio.csv"
72    if not os.path.exists(csv_path):
73        raise HTTPException(status_code=404, detail="Portfolio file not found")
74    
75    with open(csv_path, "r") as f:
76        content = f.read()
77    
78    return Response(content=content, media_type="text/csv", headers={"Content-Disposition": f"attachment; filename=portfolio.csv"})

1fastapi
2uvicorn