Browse the data

1import os
2import random
3import string
4import shutil
5from fastapi import FastAPI, HTTPException, Header, UploadFile, File
6from fastapi.responses import FileResponse
7from pathlib import Path
8 
9app = FastAPI()
10 
11users = {}
12tokens = {}
13trips = {}
14next_user_id = 1
15next_trip_id = 1
16 
17def generate_token():
18    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
19 
20def get_current_user(authorization: str = Header(None)):
21    if not authorization:
22        raise HTTPException(status_code=401, detail="Missing auth header")
23    token = authorization.replace("Bearer ", "")
24    if token not in tokens:
25        raise HTTPException(status_code=401, detail="Invalid token")
26    return tokens[token]
27 
28@app.post("/signup")
29def signup(username: str, password: str):
30    global next_user_id
31    for user in users.values():
32        if user["username"] == username:
33            raise HTTPException(status_code=400, detail="User exists")
34    user_id = next_user_id
35    users[user_id] = {"id": user_id, "username": username, "password": password}
36    next_user_id += 1
37    return {"id": user_id, "username": username}
38 
39@app.post("/login")
40def login(username: str, password: str):
41    for user in users.values():
42        if user["username"] == username and user["password"] == password:
43            token = generate_token()
44            tokens[token] = user["id"]
45            return {"token": token}
46    raise HTTPException(status_code=401, detail="Invalid credentials")
47 
48@app.post("/trip")
49def create_trip(authorization: str = Header(None)):
50    user_id = get_current_user(authorization)
51    global next_trip_id
52    trip_id = next_trip_id
53    trips[trip_id] = {"id": trip_id, "owner": user_id}
54    next_trip_id += 1
55    trip_dir = f"/var/trips/{trip_id}"
56    os.makedirs(trip_dir, exist_ok=True)
57    return {"id": trip_id}
58 
59@app.get("/trip/{trip_id}")
60def get_trip(trip_id: int, authorization: str = Header(None)):
61    get_current_user(authorization)
62    if trip_id not in trips:
63        raise HTTPException(status_code=404, detail="Trip not found")
64    return trips[trip_id]
65 
66@app.get("/trip/{trip_id}/docs/{file_name}")
67def get_trip_doc(trip_id: int, file_name: str, authorization: str = Header(None)):
68    get_current_user(authorization)
69    if trip_id not in trips:
70        raise HTTPException(status_code=404, detail="Trip not found")
71    file_path = Path(f"/var/trips/{trip_id}/{file_name}")
72    if not file_path.exists() or not file_path.is_file():
73        raise HTTPException(status_code=404, detail="File not found")
74    return FileResponse(str(file_path), media_type="application/pdf")
75 
76@app.post("/trip/{trip_id}/docs/{file_name}")
77async def upload_trip_doc(trip_id: int, file_name: str, file: UploadFile = File(...), authorization: str = Header(None)):
78    get_current_user(authorization)
79    if trip_id not in trips:
80        raise HTTPException(status_code=404, detail="Trip not found")
81    trip_dir = Path(f"/var/trips/{trip_id}")
82    trip_dir.mkdir(parents=True, exist_ok=True)
83    # Remove old files in the directory
84    for old_file in trip_dir.iterdir():
85        if old_file.is_file():
86            old_file.unlink()
87    file_path = trip_dir / file_name
88    with open(file_path, "wb") as f:
89        content = await file.read()
90        f.write(content)
91    return {"message": "File uploaded", "file": file_name}

1fastapi
2uvicorn
3python-multipart