adii1823
123 lines
3.0 KiB
Python
123 lines
3.0 KiB
Python
# api_code/api/stations.py
|
|
from typing import Optional
|
|
|
|
from fastapi import (
|
|
APIRouter,
|
|
Depends,
|
|
HTTPException,
|
|
Response,
|
|
status,
|
|
)
|
|
from sqlalchemy.orm import Session
|
|
|
|
from . import crud
|
|
from .deps import get_db
|
|
from .schemas import Station, StationCreate, StationUpdate, Train
|
|
|
|
router = APIRouter(prefix="/stations")
|
|
|
|
|
|
@router.get("", response_model=list[Station], tags=["Stations"])
|
|
def get_stations(
|
|
db: Session = Depends(get_db), code: Optional[str] = None
|
|
):
|
|
return crud.get_stations(db=db, code=code)
|
|
|
|
|
|
@router.get(
|
|
"/{station_id}", response_model=Station, tags=["Stations"]
|
|
)
|
|
def get_station(station_id: int, db: Session = Depends(get_db)):
|
|
db_station = crud.get_station(db=db, station_id=station_id)
|
|
if db_station is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Station {station_id} not found.",
|
|
)
|
|
return db_station
|
|
|
|
|
|
@router.get(
|
|
"/{station_id}/departures",
|
|
response_model=list[Train],
|
|
tags=["Trains"],
|
|
)
|
|
def get_station_departures(
|
|
station_id: int, db: Session = Depends(get_db)
|
|
):
|
|
db_station = _get_station(db=db, station_id=station_id)
|
|
return db_station.departures
|
|
|
|
|
|
def _get_station(db: Session, station_id: int):
|
|
db_station = crud.get_station(db=db, station_id=station_id)
|
|
if db_station is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Station {station_id} not found.",
|
|
)
|
|
return db_station
|
|
|
|
|
|
@router.get(
|
|
"/{station_id}/arrivals",
|
|
response_model=list[Train],
|
|
tags=["Trains"],
|
|
)
|
|
def get_station_arrivals(
|
|
station_id: int, db: Session = Depends(get_db)
|
|
):
|
|
db_station = _get_station(db=db, station_id=station_id)
|
|
return db_station.arrivals
|
|
|
|
|
|
@router.post(
|
|
"",
|
|
response_model=Station,
|
|
status_code=status.HTTP_201_CREATED,
|
|
tags=["Stations"],
|
|
)
|
|
def create_station(
|
|
station: StationCreate, db: Session = Depends(get_db)
|
|
):
|
|
db_station = crud.get_station_by_code(
|
|
db=db, code=station.code
|
|
)
|
|
if db_station:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Station {station.code} already exists.",
|
|
)
|
|
return crud.create_station(db=db, station=station)
|
|
|
|
|
|
@router.put("/{station_id}", tags=["Stations"])
|
|
def update_station(
|
|
station_id: int,
|
|
station: StationUpdate,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
db_station = crud.get_station(db=db, station_id=station_id)
|
|
|
|
if db_station is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"Station {station_id} not found.",
|
|
)
|
|
|
|
else:
|
|
crud.update_station(
|
|
db=db, station=station, station_id=station_id
|
|
)
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
|
|
|
|
|
@router.delete("/{station_id}", tags=["Stations"])
|
|
def delete_station(
|
|
station_id: int, db: Session = Depends(get_db)
|
|
):
|
|
row_count = crud.delete_station(db=db, station_id=station_id)
|
|
if row_count:
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
|
return Response(status_code=status.HTTP_404_NOT_FOUND)
|