Browse the data

1import csv
2import io
3import zipfile
4from pathlib import Path
5from typing import Optional
6 
7from fastapi import FastAPI, HTTPException, Header
8from fastapi.responses import StreamingResponse
9from pydantic import BaseModel
10import uvicorn
11 
12app = FastAPI()
13 
14# In-memory stores
15users = {}
16next_user_id = 1
17tokens = {}  # token -> user_id
18 
19# Resource stores (users can create arbitrary resources)
20resources = {}
21next_resource_id = 1
22 
23BASE_DIR = Path("/var/finance")
24 
25# ---------- Auth helpers ----------
26def get_current_user(authorization: Optional[str] = Header(None)):
27    if not authorization:
28        raise HTTPException(status_code=401, detail="Missing Authorization header")
29    token = authorization.replace("Bearer ", "")
30    user_id = tokens.get(token)
31    if user_id is None:
32        raise HTTPException(status_code=401, detail="Invalid token")
33    return user_id
34 
35# ---------- Auth endpoints ----------
36class SignupRequest(BaseModel):
37    username: str
38    password: str
39 
40class LoginRequest(BaseModel):
41    username: str
42    password: str
43 
44@app.post("/signup")
45def signup(req: SignupRequest):
46    global next_user_id
47    user_id = next_user_id
48    next_user_id += 1
49    users[user_id] = {"username": req.username, "password": req.password}
50    token = f"tok_{user_id}_{req.username}"
51    tokens[token] = user_id
52    return {"user_id": user_id, "token": token}
53 
54@app.post("/login")
55def login(req: LoginRequest):
56    for uid, u in users.items():
57        if u["username"] == req.username and u["password"] == req.password:
58            token = f"tok_{uid}_{req.username}"
59            tokens[token] = uid
60            return {"token": token}
61    raise HTTPException(status_code=401, detail="Invalid credentials")
62 
63# ---------- Generic resource endpoints ----------
64class CreateResourceRequest(BaseModel):
65    name: str
66    data: dict = {}
67 
68@app.get("/{resource}/{resource_id}")
69def get_resource(resource: str, resource_id: int, authorization: Optional[str] = Header(None)):
70    get_current_user(authorization)
71    key = (resource, resource_id)
72    if key not in resources:
73        raise HTTPException(status_code=404, detail="Resource not found")
74    return resources[key]
75 
76@app.post("/{resource}")
77def create_resource(resource: str, req: CreateResourceRequest, authorization: Optional[str] = Header(None)):
78    get_current_user(authorization)
79    global next_resource_id
80    resource_id = next_resource_id
81    next_resource_id += 1
82    key = (resource, resource_id)
83    resources[key] = {"id": resource_id, "name": req.name, "data": req.data}
84    return {"id": resource_id, "name": req.name, "data": req.data}
85 
86# ---------- Finance-specific endpoints ----------
87@app.get("/statement")
88def get_statement(account: str, file: str, authorization: Optional[str] = Header(None)):
89    get_current_user(authorization)
90    file_path = BASE_DIR / account / file
91    if not file_path.exists() or not file_path.is_file():
92        raise HTTPException(status_code=404, detail="File not found")
93    if file_path.suffix.lower() != ".csv":
94        raise HTTPException(status_code=400, detail="Only CSV files are supported")
95    with open(file_path, "r") as f:
96        content = f.read()
97    return StreamingResponse(iter([content]), media_type="text/csv", headers={"Content-Disposition": f"attachment; filename={file}"})
98 
99@app.get("/bulk")
100def bulk_download(account: str, files: str, authorization: Optional[str] = Header(None)):
101    get_current_user(authorization)
102    file_names = [f.strip() for f in files.split(",") if f.strip()]
103    if not file_names:
104        raise HTTPException(status_code=400, detail="No files specified")
105 
106    zip_buffer = io.BytesIO()
107    with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
108        for fname in file_names:
109            file_path = BASE_DIR / account / fname
110            if not file_path.exists() or not file_path.is_file():
111                continue  # skip missing files
112            if file_path.suffix.lower() != ".csv":
113                continue
114            with open(file_path, "r") as f:
115                zf.writestr(fname, f.read())
116    zip_buffer.seek(0)
117    return StreamingResponse(zip_buffer, media_type="application/zip", headers={"Content-Disposition": f"attachment; filename={account}_bulk.zip"})
118 
119if __name__ == "__main__":
120    uvicorn.run(app, host="127.0.0.1", port=8000)

1fastapi
2uvicorn