Browse the data

1import os
2import zipfile
3import io
4from fastapi import FastAPI, HTTPException, Header, File, UploadFile
5from fastapi.responses import FileResponse, StreamingResponse
6from pydantic import BaseModel
7from typing import Optional
8import uuid
9import shutil
10 
11app = FastAPI()
12 
13users = {}
14tokens = {}
15orders = {}
16shipments = {}
17next_user_id = 1
18next_order_id = 1
19next_shipment_id = 1
20 
21def get_current_user(authorization: Optional[str] = Header(None)):
22    if not authorization:
23        raise HTTPException(status_code=401, detail="Missing auth header")
24    token = authorization.replace("Bearer ", "")
25    if token not in tokens:
26        raise HTTPException(status_code=401, detail="Invalid token")
27    return tokens[token]
28 
29class SignupRequest(BaseModel):
30    username: str
31    password: str
32 
33class LoginRequest(BaseModel):
34    username: str
35    password: str
36 
37@app.post("/signup")
38def signup(req: SignupRequest):
39    global next_user_id
40    for u in users.values():
41        if u["username"] == req.username:
42            raise HTTPException(status_code=400, detail="Username taken")
43    user_id = next_user_id
44    next_user_id += 1
45    users[user_id] = {"id": user_id, "username": req.username, "password": req.password}
46    return {"id": user_id, "username": req.username}
47 
48@app.post("/login")
49def login(req: LoginRequest):
50    for u in users.values():
51        if u["username"] == req.username and u["password"] == req.password:
52            token = str(uuid.uuid4())
53            tokens[token] = u["id"]
54            return {"token": token}
55    raise HTTPException(status_code=401, detail="Invalid credentials")
56 
57class OrderCreate(BaseModel):
58    description: str
59    amount: float
60 
61@app.post("/orders")
62def create_order(order: OrderCreate, authorization: Optional[str] = Header(None)):
63    user_id = get_current_user(authorization)
64    global next_order_id
65    oid = next_order_id
66    next_order_id += 1
67    orders[oid] = {"id": oid, "description": order.description, "amount": order.amount, "user_id": user_id}
68    return orders[oid]
69 
70@app.get("/orders/{order_id}")
71def get_order(order_id: int, authorization: Optional[str] = Header(None)):
72    user_id = get_current_user(authorization)
73    if order_id not in orders:
74        raise HTTPException(status_code=404, detail="Order not found")
75    return orders[order_id]
76 
77class ShipmentCreate(BaseModel):
78    order_id: int
79    address: str
80 
81@app.post("/shipments")
82def create_shipment(shipment: ShipmentCreate, authorization: Optional[str] = Header(None)):
83    user_id = get_current_user(authorization)
84    if shipment.order_id not in orders:
85        raise HTTPException(status_code=404, detail="Order not found")
86    global next_shipment_id
87    sid = next_shipment_id
88    next_shipment_id += 1
89    shipments[sid] = {"id": sid, "order_id": shipment.order_id, "address": shipment.address, "user_id": user_id}
90    os.makedirs(f"/var/shipments/{sid}", exist_ok=True)
91    return shipments[sid]
92 
93@app.get("/shipments/{shipment_id}")
94def get_shipment(shipment_id: int, authorization: Optional[str] = Header(None)):
95    user_id = get_current_user(authorization)
96    if shipment_id not in shipments:
97        raise HTTPException(status_code=404, detail="Shipment not found")
98    return shipments[shipment_id]
99 
100@app.post("/shipments/{shipment_id}/docs")
101def upload_doc(shipment_id: int, file: UploadFile = File(...), authorization: Optional[str] = Header(None)):
102    user_id = get_current_user(authorization)
103    if shipment_id not in shipments:
104        raise HTTPException(status_code=404, detail="Shipment not found")
105    os.makedirs(f"/var/shipments/{shipment_id}", exist_ok=True)
106    file_path = f"/var/shipments/{shipment_id}/{file.filename}"
107    with open(file_path, "wb") as f:
108        shutil.copyfileobj(file.file, f)
109    return {"filename": file.filename}
110 
111@app.get("/shipment/{shipment_id}/docs/{file_name}")
112def get_doc(shipment_id: int, file_name: str, authorization: Optional[str] = Header(None)):
113    user_id = get_current_user(authorization)
114    if shipment_id not in shipments:
115        raise HTTPException(status_code=404, detail="Shipment not found")
116    file_path = f"/var/shipments/{shipment_id}/{file_name}"
117    if not os.path.exists(file_path):
118        raise HTTPException(status_code=404, detail="File not found")
119    return FileResponse(file_path, media_type="application/pdf")
120 
121@app.get("/shipment/{shipment_id}/docs/zip")
122def download_all_docs(shipment_id: int, authorization: Optional[str] = Header(None)):
123    user_id = get_current_user(authorization)
124    if shipment_id not in shipments:
125        raise HTTPException(status_code=404, detail="Shipment not found")
126    dir_path = f"/var/shipments/{shipment_id}"
127    if not os.path.exists(dir_path):
128        raise HTTPException(status_code=404, detail="No docs found")
129    zip_buffer = io.BytesIO()
130    with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf:
131        for root, dirs, files in os.walk(dir_path):
132            for file in files:
133                file_path = os.path.join(root, file)
134                zf.write(file_path, arcname=file)
135    zip_buffer.seek(0)
136    return StreamingResponse(zip_buffer, media_type="application/zip", headers={"Content-Disposition": f"attachment; filename=shipment_{shipment_id}_docs.zip"})

1fastapi
2uvicorn
3python-multipart