232 lines
8.2 KiB
Python
232 lines
8.2 KiB
Python
import os
|
|
import json
|
|
import uuid
|
|
from typing import Optional
|
|
from datetime import datetime
|
|
from fastapi import FastAPI, HTTPException, Path
|
|
from pydantic import BaseModel, Field
|
|
from filelock import FileLock
|
|
import argparse
|
|
import uvicorn
|
|
|
|
app = FastAPI(title="Tour Voting API")
|
|
|
|
# ─── Pydantic MODELS ──────────────────────────────────────────────────────────
|
|
|
|
class Idea(BaseModel):
|
|
id: str
|
|
name: str
|
|
start_time: Optional[datetime] = None
|
|
end_time: Optional[datetime] = None
|
|
description: str
|
|
voters: list[str] = Field(default_factory=list)
|
|
decided: bool = False
|
|
|
|
class Tour(BaseModel):
|
|
id: str
|
|
name: str
|
|
start_date: Optional[datetime] = None
|
|
end_date: Optional[datetime] = None
|
|
description: str
|
|
ideas: list[Idea] = Field(default_factory=list)
|
|
createdAt: datetime
|
|
|
|
class TourCreate(BaseModel):
|
|
name: str
|
|
start_date: Optional[datetime] = None
|
|
end_date: Optional[datetime] = None
|
|
description: str
|
|
|
|
class IdeaCreate(BaseModel):
|
|
name: str
|
|
start_time: Optional[datetime] = None
|
|
end_time: Optional[datetime] = None
|
|
description: str
|
|
|
|
class Vote(BaseModel):
|
|
voterName: str
|
|
|
|
# ─── GLOBAL CONFIG ────────────────────────────────────────────────────────────
|
|
|
|
DATA_DIR = "" # will be overridden via command line
|
|
|
|
def get_tour_filepath(tour_id: str) -> str:
|
|
fname = f"{tour_id}.json"
|
|
full = os.path.abspath(os.path.join(DATA_DIR, fname))
|
|
if not full.startswith(os.path.abspath(DATA_DIR) + os.sep):
|
|
raise HTTPException(status_code=400, detail="Invalid tour ID")
|
|
return full
|
|
|
|
# ─── ENDPOINT: CREATE TOUR ─────────────────────────────────────────────────────
|
|
|
|
@app.post("/tour/v1/tours", response_model=Tour)
|
|
def create_tour(tour: TourCreate):
|
|
"""
|
|
Create a new tour. Generates its own UUID and timestamp, then writes to {tour_id}.json.
|
|
"""
|
|
tour_id = str(uuid.uuid4())
|
|
created_at = datetime.utcnow()
|
|
|
|
new_tour = Tour(
|
|
id=tour_id,
|
|
name=tour.name,
|
|
start_date=tour.start_date,
|
|
end_date=tour.end_date,
|
|
description=tour.description,
|
|
ideas=[],
|
|
createdAt=created_at,
|
|
)
|
|
|
|
filepath = get_tour_filepath(tour_id)
|
|
with open(filepath, "w") as f:
|
|
# Using default=str so that datetime objects become ISO strings
|
|
json.dump(new_tour.model_dump(), f, default=str)
|
|
|
|
return new_tour
|
|
|
|
# ─── ENDPOINT: GET TOUR ────────────────────────────────────────────────────────
|
|
|
|
@app.get("/tour/v1/tours/{tour_id}", response_model=Tour)
|
|
def get_tour(tour_id: str = Path(...)):
|
|
"""
|
|
Retrieve an existing tour by ID. Loads {tour_id}.json and parses back into a Tour.
|
|
"""
|
|
filepath = get_tour_filepath(tour_id)
|
|
if not os.path.exists(filepath):
|
|
raise HTTPException(status_code=404, detail="Tour not found")
|
|
|
|
with open(filepath) as f:
|
|
data = json.load(f)
|
|
|
|
if data.get("start_date"):
|
|
data["start_date"] = datetime.fromisoformat(data["start_date"])
|
|
if data.get("end_date"):
|
|
data["end_date"] = datetime.fromisoformat(data["end_date"])
|
|
data["createdAt"] = datetime.fromisoformat(data["createdAt"])
|
|
|
|
# Convert nested idea timestamps as well
|
|
for idea in data["ideas"]:
|
|
if idea.get("start_time"):
|
|
idea["start_time"] = datetime.fromisoformat(idea["start_time"])
|
|
if idea.get("end_time"):
|
|
idea["end_time"] = datetime.fromisoformat(idea["end_time"])
|
|
return Tour(**data)
|
|
|
|
# ─── ENDPOINT: ADD IDEA ─────────────────────────────────────────────────────────
|
|
|
|
@app.post("/tour/v1/tours/{tour_id}/ideas", response_model=Idea)
|
|
def add_idea(tour_id: str, idea: IdeaCreate):
|
|
"""
|
|
Add a new idea to the given tour. Generates a new UUID for the idea and appends it.
|
|
"""
|
|
filepath = get_tour_filepath(tour_id)
|
|
if not os.path.exists(filepath):
|
|
raise HTTPException(status_code=404, detail="Tour not found")
|
|
|
|
lock = FileLock(filepath + ".lock")
|
|
with lock:
|
|
with open(filepath) as f:
|
|
data = json.load(f)
|
|
|
|
if any(i.get("decided") for i in data["ideas"]):
|
|
raise HTTPException(status_code=403, detail="Idea already decided")
|
|
|
|
idea_id = str(uuid.uuid4())
|
|
new_idea = Idea(
|
|
id=idea_id,
|
|
name=idea.name,
|
|
description=idea.description,
|
|
voters=[],
|
|
start_time=idea.start_time,
|
|
end_time=idea.end_time,
|
|
decided=False,
|
|
)
|
|
|
|
data["ideas"].append(new_idea.model_dump())
|
|
with open(filepath, "w") as f:
|
|
json.dump(data, f, default=str)
|
|
|
|
return new_idea
|
|
|
|
# ─── ENDPOINT: VOTE FOR IDEA ───────────────────────────────────────────────────
|
|
|
|
@app.post("/tour/v1/tours/{tour_id}/ideas/{idea_id}/vote", response_model=Idea)
|
|
def vote_idea(tour_id: str, idea_id: str, vote: Vote):
|
|
"""
|
|
Cast a vote on a specific idea. Adds voterName if not already present.
|
|
"""
|
|
filepath = get_tour_filepath(tour_id)
|
|
if not os.path.exists(filepath):
|
|
raise HTTPException(status_code=404, detail="Tour not found")
|
|
|
|
lock = FileLock(filepath + ".lock")
|
|
with lock:
|
|
with open(filepath) as f:
|
|
data = json.load(f)
|
|
|
|
if any(i.get("decided") for i in data["ideas"]):
|
|
raise HTTPException(status_code=403, detail="Idea already decided")
|
|
for idea in data["ideas"]:
|
|
if idea["id"] == idea_id:
|
|
if vote.voterName not in idea["voters"]:
|
|
idea["voters"].append(vote.voterName)
|
|
# Persist the change
|
|
with open(filepath, "w") as f:
|
|
json.dump(data, f, default=str)
|
|
return Idea(**idea)
|
|
|
|
raise HTTPException(status_code=404, detail="Idea not found")
|
|
|
|
# ─── ENDPOINT: DECIDE IDEA ────────────────────────────────────────────────────
|
|
|
|
@app.post("/tour/v1/tours/{tour_id}/ideas/{idea_id}/decide", response_model=Idea)
|
|
def decide_idea(tour_id: str, idea_id: str):
|
|
"""Mark an idea as decided. Only one idea can be decided per tour."""
|
|
filepath = get_tour_filepath(tour_id)
|
|
if not os.path.exists(filepath):
|
|
raise HTTPException(status_code=404, detail="Tour not found")
|
|
|
|
lock = FileLock(filepath + ".lock")
|
|
with lock:
|
|
with open(filepath) as f:
|
|
data = json.load(f)
|
|
|
|
if any(i.get("decided") for i in data["ideas"]):
|
|
raise HTTPException(status_code=403, detail="Idea already decided")
|
|
|
|
for idea in data["ideas"]:
|
|
if idea["id"] == idea_id:
|
|
idea["decided"] = True
|
|
with open(filepath, "w") as f:
|
|
json.dump(data, f, default=str)
|
|
return Idea(**idea)
|
|
|
|
raise HTTPException(status_code=404, detail="Idea not found")
|
|
|
|
# ─── MAIN: RUN UVIDORN WITH COMMAND-LINE DATA DIR ──────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Run Tour Voting API server")
|
|
parser.add_argument(
|
|
"--data-dir",
|
|
required=True,
|
|
help="Directory in which to store tour JSON files (one file per tour_id)",
|
|
)
|
|
parser.add_argument(
|
|
"--host",
|
|
default="0.0.0.0",
|
|
help="Host to run the server on",
|
|
)
|
|
parser.add_argument(
|
|
"--port",
|
|
type=int,
|
|
default=8000,
|
|
help="Port to run the server on",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
DATA_DIR = args.data_dir
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
|
|
uvicorn.run(app, host=args.host, port=args.port, reload=False)
|