Add unit tests for station service and enhance documentation
Some checks failed
Backend CI / lint-and-test (push) Failing after 37s
Some checks failed
Backend CI / lint-and-test (push) Failing after 37s
- Introduced unit tests for the station service, covering creation, updating, and archiving of stations. - Added detailed building block view documentation outlining the architecture of the Rail Game system. - Created runtime view documentation illustrating key user interactions and system behavior. - Developed concepts documentation detailing domain models, architectural patterns, and security considerations. - Updated architecture documentation to reference new detailed sections for building block and runtime views.
This commit is contained in:
195
backend/app/services/stations.py
Normal file
195
backend/app/services/stations.py
Normal file
@@ -0,0 +1,195 @@
|
||||
from __future__ import annotations
|
||||
|
||||
"""Application services for station CRUD operations."""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from typing import cast
|
||||
from uuid import UUID
|
||||
|
||||
from geoalchemy2.elements import WKBElement, WKTElement
|
||||
from geoalchemy2.shape import to_shape
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from backend.app.db.models import Station
|
||||
from backend.app.models import StationCreate, StationModel, StationUpdate
|
||||
from backend.app.repositories import StationRepository
|
||||
|
||||
try: # pragma: no cover - optional dependency guard
|
||||
from shapely.geometry import Point # type: ignore
|
||||
except ImportError: # pragma: no cover - allow running without shapely at import time
|
||||
Point = None # type: ignore[assignment]
|
||||
|
||||
|
||||
def list_stations(
|
||||
session: Session, include_inactive: bool = False
|
||||
) -> list[StationModel]:
|
||||
repo = StationRepository(session)
|
||||
if include_inactive:
|
||||
stations = repo.list()
|
||||
else:
|
||||
stations = repo.list_active()
|
||||
return [_to_station_model(station) for station in stations]
|
||||
|
||||
|
||||
def get_station(session: Session, station_id: str) -> StationModel:
|
||||
repo = StationRepository(session)
|
||||
station = _resolve_station(repo, station_id)
|
||||
return _to_station_model(station)
|
||||
|
||||
|
||||
def create_station(session: Session, payload: StationCreate) -> StationModel:
|
||||
name = payload.name.strip()
|
||||
if not name:
|
||||
raise ValueError("Station name must not be empty")
|
||||
_validate_coordinates(payload.latitude, payload.longitude)
|
||||
|
||||
repo = StationRepository(session)
|
||||
station = repo.create(
|
||||
StationCreate(
|
||||
name=name,
|
||||
latitude=payload.latitude,
|
||||
longitude=payload.longitude,
|
||||
osm_id=_normalize_optional(payload.osm_id),
|
||||
code=_normalize_optional(payload.code),
|
||||
elevation_m=payload.elevation_m,
|
||||
is_active=payload.is_active,
|
||||
)
|
||||
)
|
||||
session.flush()
|
||||
session.refresh(station)
|
||||
session.commit()
|
||||
return _to_station_model(station)
|
||||
|
||||
|
||||
def update_station(
|
||||
session: Session, station_id: str, payload: StationUpdate
|
||||
) -> StationModel:
|
||||
repo = StationRepository(session)
|
||||
station = _resolve_station(repo, station_id)
|
||||
|
||||
if payload.name is not None:
|
||||
name = payload.name.strip()
|
||||
if not name:
|
||||
raise ValueError("Station name must not be empty")
|
||||
station.name = name
|
||||
|
||||
if payload.latitude is not None or payload.longitude is not None:
|
||||
if payload.latitude is None or payload.longitude is None:
|
||||
raise ValueError("Both latitude and longitude must be provided together")
|
||||
_validate_coordinates(payload.latitude, payload.longitude)
|
||||
station.location = repo._point(
|
||||
payload.latitude, payload.longitude
|
||||
) # type: ignore[assignment]
|
||||
|
||||
if payload.osm_id is not None:
|
||||
station.osm_id = _normalize_optional(payload.osm_id)
|
||||
|
||||
if payload.code is not None:
|
||||
station.code = _normalize_optional(payload.code)
|
||||
|
||||
if payload.elevation_m is not None:
|
||||
station.elevation_m = payload.elevation_m
|
||||
|
||||
if payload.is_active is not None:
|
||||
station.is_active = payload.is_active
|
||||
|
||||
session.flush()
|
||||
session.refresh(station)
|
||||
session.commit()
|
||||
return _to_station_model(station)
|
||||
|
||||
|
||||
def archive_station(session: Session, station_id: str) -> StationModel:
|
||||
repo = StationRepository(session)
|
||||
station = _resolve_station(repo, station_id)
|
||||
if station.is_active:
|
||||
station.is_active = False
|
||||
session.flush()
|
||||
session.refresh(station)
|
||||
session.commit()
|
||||
return _to_station_model(station)
|
||||
|
||||
|
||||
def _resolve_station(repo: StationRepository, station_id: str) -> Station:
|
||||
identifier = _parse_station_id(station_id)
|
||||
station = repo.get(identifier)
|
||||
if station is None:
|
||||
raise LookupError("Station not found")
|
||||
return station
|
||||
|
||||
|
||||
def _parse_station_id(station_id: str) -> UUID:
|
||||
try:
|
||||
return UUID(station_id)
|
||||
except (ValueError, TypeError) as exc: # pragma: no cover - simple validation
|
||||
raise ValueError("Invalid station identifier") from exc
|
||||
|
||||
|
||||
def _validate_coordinates(latitude: float, longitude: float) -> None:
|
||||
if not (-90.0 <= latitude <= 90.0):
|
||||
raise ValueError("Latitude must be between -90 and 90 degrees")
|
||||
if not (-180.0 <= longitude <= 180.0):
|
||||
raise ValueError("Longitude must be between -180 and 180 degrees")
|
||||
|
||||
|
||||
def _normalize_optional(value: str | None) -> str | None:
|
||||
if value is None:
|
||||
return None
|
||||
normalized = value.strip()
|
||||
return normalized or None
|
||||
|
||||
|
||||
def _to_station_model(station: Station) -> StationModel:
|
||||
latitude, longitude = _extract_coordinates(station.location)
|
||||
created_at = station.created_at or datetime.now(timezone.utc)
|
||||
updated_at = station.updated_at or created_at
|
||||
return StationModel(
|
||||
id=str(station.id),
|
||||
name=station.name,
|
||||
latitude=latitude,
|
||||
longitude=longitude,
|
||||
code=station.code,
|
||||
osm_id=station.osm_id,
|
||||
elevation_m=station.elevation_m,
|
||||
is_active=station.is_active,
|
||||
created_at=cast(datetime, created_at),
|
||||
updated_at=cast(datetime, updated_at),
|
||||
)
|
||||
|
||||
|
||||
def _extract_coordinates(location: object) -> tuple[float, float]:
|
||||
if location is None:
|
||||
raise ValueError("Station location is unavailable")
|
||||
|
||||
# Attempt to leverage GeoAlchemy's shapely integration first.
|
||||
try:
|
||||
geometry = to_shape(cast(WKBElement | WKTElement, location))
|
||||
if Point is not None and isinstance(geometry, Point):
|
||||
return float(geometry.y), float(geometry.x)
|
||||
except Exception: # pragma: no cover - fallback handles parsing
|
||||
pass
|
||||
|
||||
if isinstance(location, WKTElement):
|
||||
return _parse_wkt_point(location.data)
|
||||
|
||||
text = getattr(location, "desc", None)
|
||||
if isinstance(text, str):
|
||||
return _parse_wkt_point(text)
|
||||
|
||||
raise ValueError("Unable to read station geometry")
|
||||
|
||||
|
||||
def _parse_wkt_point(wkt: str) -> tuple[float, float]:
|
||||
marker = "POINT"
|
||||
if not wkt.upper().startswith(marker):
|
||||
raise ValueError("Unsupported geometry format")
|
||||
start = wkt.find("(")
|
||||
end = wkt.find(")", start)
|
||||
if start == -1 or end == -1:
|
||||
raise ValueError("Malformed POINT geometry")
|
||||
coordinates = wkt[start + 1 : end].strip().split()
|
||||
if len(coordinates) != 2:
|
||||
raise ValueError("POINT geometry must contain two coordinates")
|
||||
longitude, latitude = map(float, coordinates)
|
||||
_validate_coordinates(latitude, longitude)
|
||||
return latitude, longitude
|
||||
Reference in New Issue
Block a user