Browse the data

1import os
2from datetime import datetime
3from pathlib import Path
4from typing import Optional
5 
6from fastapi import FastAPI, HTTPException, Header
7from pydantic import BaseModel
8import uvicorn
9 
10app = FastAPI()
11 
12# In-memory user storage
13users = {}
14user_id_counter = 1
15# In-memory token storage: token -> user_id
16tokens = {}
17 
18# Simple auth helper
19def get_current_user(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 
27class SignupRequest(BaseModel):
28    username: str
29    password: str
30 
31class LoginRequest(BaseModel):
32    username: str
33    password: str
34 
35@app.post("/signup")
36def signup(req: SignupRequest):
37    global user_id_counter
38    for u in users.values():
39        if u["username"] == req.username:
40            raise HTTPException(status_code=400, detail="Username already exists")
41    uid = user_id_counter
42    user_id_counter += 1
43    users[uid] = {"id": uid, "username": req.username, "password": req.password}
44    return {"id": uid, "username": req.username}
45 
46@app.post("/login")
47def login(req: LoginRequest):
48    for uid, u in users.items():
49        if u["username"] == req.username and u["password"] == req.password:
50            token = f"tok_{uid}_{datetime.utcnow().timestamp()}"
51            tokens[token] = uid
52            return {"token": token}
53    raise HTTPException(status_code=401, detail="Invalid credentials")
54 
55@app.get("/shipments/{tracking_id}/documents/{doc_name}")
56def get_document(tracking_id: str, doc_name: str, authorization: str = Header(None)):
57    get_current_user(authorization)
58    base_path = Path(f"/logistics/{tracking_id}")
59    file_path = base_path / doc_name
60    if not file_path.exists() or not file_path.is_file():
61        raise HTTPException(status_code=404, detail="Document not found")
62    return FileResponse(file_path)
63 
64@app.get("/shipments/{tracking_id}/documents")
65def list_documents(tracking_id: str, page: int = 1, per_page: int = 10, authorization: str = Header(None)):
66    get_current_user(authorization)
67    base_path = Path(f"/logistics/{tracking_id}")
68    if not base_path.exists() or not base_path.is_dir():
69        raise HTTPException(status_code=404, detail="Tracking ID not found")
70    all_files = []
71    for f in base_path.iterdir():
72        if f.is_file():
73            stat = f.stat()
74            all_files.append({
75                "name": f.name,
76                "size": stat.st_size,
77                "date": datetime.fromtimestamp(stat.st_mtime).isoformat()
78            })
79    total = len(all_files)
80    start = (page - 1) * per_page
81    end = start + per_page
82    return {
83        "page": page,
84        "per_page": per_page,
85        "total": total,
86        "documents": all_files[start:end]
87    }
88 
89# Dummy resources for demo
90orders = {}
91order_id_counter = 1
92 
93class OrderCreate(BaseModel):
94    description: str
95 
96@app.post("/orders")
97def create_order(req: OrderCreate, authorization: str = Header(None)):
98    get_current_user(authorization)
99    global order_id_counter
100    oid = order_id_counter
101    order_id_counter += 1
102    orders[oid] = {"id": oid, "description": req.description}
103    return orders[oid]
104 
105@app.get("/orders/{order_id}")
106def get_order(order_id: int, authorization: str = Header(None)):
107    get_current_user(authorization)
108    if order_id not in orders:
109        raise HTTPException(status_code=404, detail="Order not found")
110    return orders[order_id]
111 
112if __name__ == "__main__":
113    uvicorn.run(app, host="127.0.0.1", port=8000)

1fastapi
2uvicorn