zwitschi
68048ff574
- Introduced CombinedTrackModel, CombinedTrackCreate, and CombinedTrackRepository for managing combined tracks. - Implemented logic to create combined tracks based on existing tracks between two stations. - Added methods to check for existing combined tracks and retrieve constituent track IDs. - Enhanced TrackModel and TrackRepository to support OSM ID and track updates. - Created migration scripts for adding combined tracks table and OSM ID to tracks. - Updated services and API endpoints to handle combined track operations. - Added tests for combined track creation, repository methods, and API interactions.
154 lines
4.4 KiB
Python
154 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.app.api.deps import get_current_user, get_db
|
|
from backend.app.models import (
|
|
CombinedTrackModel,
|
|
TrackCreate,
|
|
TrackUpdate,
|
|
TrackModel,
|
|
UserPublic,
|
|
)
|
|
from backend.app.services.combined_tracks import (
|
|
create_combined_track,
|
|
get_combined_track,
|
|
list_combined_tracks,
|
|
)
|
|
from backend.app.services.tracks import (
|
|
create_track,
|
|
delete_track,
|
|
regenerate_combined_tracks,
|
|
update_track,
|
|
get_track,
|
|
list_tracks,
|
|
)
|
|
|
|
router = APIRouter(prefix="/tracks", tags=["tracks"])
|
|
|
|
|
|
@router.get("", response_model=list[TrackModel])
|
|
def read_combined_tracks(
|
|
_: UserPublic = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> list[TrackModel]:
|
|
"""Return all base tracks."""
|
|
return list_tracks(db)
|
|
|
|
|
|
@router.get("/combined", response_model=list[CombinedTrackModel])
|
|
def read_combined_tracks_combined(
|
|
_: UserPublic = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> list[CombinedTrackModel]:
|
|
return list_combined_tracks(db)
|
|
|
|
|
|
@router.get("/{track_id}", response_model=TrackModel)
|
|
def read_track(
|
|
track_id: str,
|
|
_: UserPublic = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> TrackModel:
|
|
track = get_track(db, track_id)
|
|
if track is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Track {track_id} not found",
|
|
)
|
|
return track
|
|
|
|
|
|
@router.get("/combined/{combined_track_id}", response_model=CombinedTrackModel)
|
|
def read_combined_track(
|
|
combined_track_id: str,
|
|
_: UserPublic = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> CombinedTrackModel:
|
|
combined_track = get_combined_track(db, combined_track_id)
|
|
if combined_track is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Combined track {combined_track_id} not found",
|
|
)
|
|
return combined_track
|
|
|
|
|
|
@router.post("", response_model=TrackModel, status_code=status.HTTP_201_CREATED)
|
|
def create_track_endpoint(
|
|
payload: TrackCreate,
|
|
regenerate: bool = False,
|
|
_: UserPublic = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> TrackModel:
|
|
try:
|
|
track = create_track(db, payload)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)
|
|
) from exc
|
|
|
|
if regenerate:
|
|
regenerate_combined_tracks(
|
|
db, [track.start_station_id, track.end_station_id])
|
|
|
|
return track
|
|
|
|
|
|
@router.post(
|
|
"/combined",
|
|
response_model=CombinedTrackModel,
|
|
status_code=status.HTTP_201_CREATED,
|
|
summary="Create a combined track between two stations using pathfinding",
|
|
)
|
|
def create_combined_track_endpoint(
|
|
start_station_id: str,
|
|
end_station_id: str,
|
|
_: UserPublic = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> CombinedTrackModel:
|
|
combined_track = create_combined_track(
|
|
db, start_station_id, end_station_id)
|
|
if combined_track is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Could not create combined track: no path exists between stations or track already exists",
|
|
)
|
|
return combined_track
|
|
|
|
|
|
@router.put("/{track_id}", response_model=TrackModel)
|
|
def update_track_endpoint(
|
|
track_id: str,
|
|
payload: TrackUpdate,
|
|
regenerate: bool = False,
|
|
_: UserPublic = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> TrackModel:
|
|
track = update_track(db, track_id, payload)
|
|
if track is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Track {track_id} not found",
|
|
)
|
|
if regenerate:
|
|
regenerate_combined_tracks(
|
|
db, [track.start_station_id, track.end_station_id])
|
|
return track
|
|
|
|
|
|
@router.delete("/{track_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_track_endpoint(
|
|
track_id: str,
|
|
regenerate: bool = False,
|
|
_: UserPublic = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> None:
|
|
deleted = delete_track(db, track_id, regenerate=regenerate)
|
|
if not deleted:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Track {track_id} not found",
|
|
)
|