Browse the data

1import io
2import requests
3from fastapi import FastAPI, HTTPException, Header
4from fastapi.responses import StreamingResponse
5from pydantic import BaseModel
6from reportlab.lib.pagesizes import letter
7from reportlab.pdfgen import canvas
8from reportlab.lib.utils import ImageReader
9 
10app = FastAPI()
11 
12users = {}
13orders = {}
14products = {}
15invoices = {}
16tokens = {}
17next_user_id = 1
18next_order_id = 1
19next_product_id = 1
20next_invoice_id = 1
21next_token_id = 1
22 
23class SignupRequest(BaseModel):
24    username: str
25    password: str
26 
27class LoginRequest(BaseModel):
28    username: str
29    password: str
30 
31class OrderCreate(BaseModel):
32    product_id: int
33    quantity: int
34    image_url: str = None
35 
36class ProductCreate(BaseModel):
37    name: str
38    price: float
39 
40def get_user_from_token(authorization: str = Header(None)):
41    if not authorization:
42        raise HTTPException(status_code=401, detail="No auth header")
43    token = authorization.replace("Bearer ", "")
44    user_id = tokens.get(token)
45    if not user_id:
46        raise HTTPException(status_code=401, detail="Invalid token")
47    return user_id
48 
49@app.post("/signup")
50def signup(req: SignupRequest):
51    global next_user_id
52    user_id = next_user_id
53    next_user_id += 1
54    users[user_id] = {"username": req.username, "password": req.password}
55    return {"user_id": user_id}
56 
57@app.post("/login")
58def login(req: LoginRequest):
59    for uid, u in users.items():
60        if u["username"] == req.username and u["password"] == req.password:
61            global next_token_id
62            token = f"tok_{next_token_id}"
63            next_token_id += 1
64            tokens[token] = uid
65            return {"token": token}
66    raise HTTPException(status_code=401, detail="Invalid credentials")
67 
68@app.post("/products")
69def create_product(req: ProductCreate, authorization: str = Header(None)):
70    get_user_from_token(authorization)
71    global next_product_id
72    pid = next_product_id
73    next_product_id += 1
74    products[pid] = {"name": req.name, "price": req.price}
75    return {"product_id": pid}
76 
77@app.get("/products/{product_id}")
78def get_product(product_id: int, authorization: str = Header(None)):
79    get_user_from_token(authorization)
80    p = products.get(product_id)
81    if not p:
82        raise HTTPException(status_code=404, detail="Product not found")
83    return p
84 
85@app.post("/orders")
86def create_order(req: OrderCreate, authorization: str = Header(None)):
87    user_id = get_user_from_token(authorization)
88    global next_order_id
89    oid = next_order_id
90    next_order_id += 1
91    orders[oid] = {"user_id": user_id, "product_id": req.product_id, "quantity": req.quantity, "image_url": req.image_url}
92    return {"order_id": oid}
93 
94@app.get("/orders/{order_id}")
95def get_order(order_id: int, authorization: str = Header(None)):
96    get_user_from_token(authorization)
97    o = orders.get(order_id)
98    if not o:
99        raise HTTPException(status_code=404, detail="Order not found")
100    return o
101 
102@app.post("/invoice/generate")
103def generate_invoice(order_id: int, authorization: str = Header(None)):
104    user_id = get_user_from_token(authorization)
105    order = orders.get(order_id)
106    if not order:
107        raise HTTPException(status_code=404, detail="Order not found")
108    if order["user_id"] != user_id:
109        raise HTTPException(status_code=403, detail="Not your order")
110 
111    product = products.get(order["product_id"])
112    buf = io.BytesIO()
113    c = canvas.Canvas(buf, pagesize=letter)
114    c.drawString(100, 750, f"Invoice for Order #{order_id}")
115    c.drawString(100, 730, f"Product: {product['name'] if product else 'Unknown'}")
116    c.drawString(100, 710, f"Quantity: {order['quantity']}")
117    c.drawString(100, 690, f"Price: ${product['price'] if product else 0}")
118 
119    if order.get("image_url"):
120        try:
121            resp = requests.get(order["image_url"], timeout=5)
122            if resp.status_code == 200:
123                img_data = io.BytesIO(resp.content)
124                img = ImageReader(img_data)
125                c.drawImage(img, 100, 500, width=200, height=200)
126        except:
127            pass
128 
129    c.showPage()
130    c.save()
131    buf.seek(0)
132 
133    global next_invoice_id
134    inv_id = next_invoice_id
135    next_invoice_id += 1
136    invoices[inv_id] = {"order_id": order_id, "user_id": user_id}
137 
138    return StreamingResponse(buf, media_type="application/pdf", headers={"Content-Disposition": f"attachment; filename=invoice_{order_id}.pdf"})

1fastapi
2uvicorn
3reportlab
4requests