Browse the data

1import hashlib
2import os
3from datetime import datetime
4from io import BytesIO
5 
6from fastapi import FastAPI, HTTPException, Header, Query
7from fastapi.responses import StreamingResponse
8from pydantic import BaseModel
9from reportlab.lib.pagesizes import letter
10from reportlab.pdfgen import canvas
11 
12app = FastAPI()
13 
14# In-memory storage
15users = {}
16next_user_id = 1
17tokens = {}
18transactions = {}
19next_txn_id = 1
20 
21# Simple auth helper
22def get_user_from_token(authorization: str = Header(None)):
23    if not authorization or not authorization.startswith("Bearer "):
24        raise HTTPException(status_code=401, detail="Missing or invalid token")
25    token = authorization.split(" ")[1]
26    user_id = tokens.get(token)
27    if not user_id:
28        raise HTTPException(status_code=401, detail="Invalid token")
29    return user_id
30 
31# --- Auth endpoints ---
32 
33class SignupRequest(BaseModel):
34    username: str
35    password: str
36 
37@app.post("/signup")
38def signup(req: SignupRequest):
39    global next_user_id
40    if req.username in users:
41        raise HTTPException(status_code=400, detail="Username already exists")
42    user_id = next_user_id
43    next_user_id += 1
44    users[req.username] = {"id": user_id, "password": req.password}
45    return {"user_id": user_id, "username": req.username}
46 
47class LoginRequest(BaseModel):
48    username: str
49    password: str
50 
51@app.post("/login")
52def login(req: LoginRequest):
53    user = users.get(req.username)
54    if not user or user["password"] != req.password:
55        raise HTTPException(status_code=401, detail="Invalid credentials")
56    token = hashlib.sha256(os.urandom(32)).hexdigest()
57    tokens[token] = user["id"]
58    return {"token": token}
59 
60# --- Transaction endpoints ---
61 
62class TransactionCreate(BaseModel):
63    amount: float
64    date: str  # ISO format e.g. 2025-01-15
65 
66@app.post("/transactions")
67def create_transaction(req: TransactionCreate, authorization: str = Header(None)):
68    user_id = get_user_from_token(authorization)
69    global next_txn_id
70    txn_id = next_txn_id
71    next_txn_id += 1
72    transactions[txn_id] = {
73        "id": txn_id,
74        "user_id": user_id,
75        "amount": req.amount,
76        "date": req.date,
77    }
78    return {"id": txn_id, "amount": req.amount, "date": req.date}
79 
80@app.get("/transactions/{txn_id}")
81def get_transaction(txn_id: int, authorization: str = Header(None)):
82    user_id = get_user_from_token(authorization)
83    txn = transactions.get(txn_id)
84    if not txn:
85        raise HTTPException(status_code=404, detail="Transaction not found")
86    if txn["user_id"] != user_id:
87        raise HTTPException(status_code=403, detail="Access denied")
88    return txn
89 
90# --- Receipt endpoints ---
91 
92@app.get("/receipts/{txn_id}")
93def get_receipt(txn_id: int, authorization: str = Header(None)):
94    user_id = get_user_from_token(authorization)
95    txn = transactions.get(txn_id)
96    if not txn:
97        raise HTTPException(status_code=404, detail="Transaction not found")
98    if txn["user_id"] != user_id:
99        raise HTTPException(status_code=403, detail="Access denied")
100 
101    # Generate PDF
102    buf = BytesIO()
103    c = canvas.Canvas(buf, pagesize=letter)
104    c.drawString(100, 750, f"Receipt for Transaction #{txn_id}")
105    c.drawString(100, 730, f"Date: {txn['date']}")
106    c.drawString(100, 710, f"Amount: ${txn['amount']:.2f}")
107    c.drawString(100, 690, "Thank you for your business!")
108    c.save()
109    buf.seek(0)
110 
111    return StreamingResponse(buf, media_type="application/pdf",
112                             headers={"Content-Disposition": f"attachment; filename=receipt_{txn_id}.pdf"})
113 
114class BatchReceiptRequest(BaseModel):
115    txn_ids: list[int]
116 
117@app.post("/receipts/batch")
118def batch_receipts(req: BatchReceiptRequest, authorization: str = Header(None)):
119    user_id = get_user_from_token(authorization)
120    from PyPDF2 import PdfMerger
121    merger = PdfMerger()
122    for txn_id in req.txn_ids:
123        txn = transactions.get(txn_id)
124        if not txn or txn["user_id"] != user_id:
125            continue
126        buf = BytesIO()
127        c = canvas.Canvas(buf, pagesize=letter)
128        c.drawString(100, 750, f"Receipt for Transaction #{txn_id}")
129        c.drawString(100, 730, f"Date: {txn['date']}")
130        c.drawString(100, 710, f"Amount: ${txn['amount']:.2f}")
131        c.drawString(100, 690, "Thank you for your business!")
132        c.save()
133        buf.seek(0)
134        merger.append(buf)
135    out = BytesIO()
136    merger.write(out)
137    out.seek(0)
138    return StreamingResponse(out, media_type="application/pdf",
139                             headers={"Content-Disposition": "attachment; filename=receipts_batch.pdf"})

1fastapi
2uvicorn
3reportlab
4PyPDF2