"""
Validated orchestration helpers for coordinate persistence and retrieval.
This module provides a thin, validated wrapper around coordinate seeding,
insertion, and listing. It centralizes argument checks, enforces geographic
bounds, and adds structured logging around database calls. The goal is to give
higher-level services a safe, predictable API that delegates the actual
persistence and generation logic to ``app.coordinates_manager``.
See Also
--------
app.coordinates_manager.seed_coordinates_if_needed :
app.coordinates_manager.get_coordinates :
app.coordinates_manager.set_coordinate :
app.coordinates_utils.calculate_destination_point :
app.schemas.CoordinateSchema :
Notes
-----
- Primary role: validate inputs (types and ranges) and orchestrate calls to
lower-level coordinate operations (seed, list, insert).
- Key dependencies: a live SQLAlchemy ``Session`` and the helpers in
``app.coordinates_manager`` for persistence and generation.
- Invariants: callers own the ``Session`` lifecycle. Latitude must
be within ``[-90, 90]`` and longitude within ``[-180, 180]``.
Examples
--------
>>> # Minimal validation example
>>> from app.coordinates_setup import validate_range
>>> validate_range(0.0, "x", -1.0, 1.0)
"""
import logging
from typing import Optional
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from .coordinates_manager import (
get_coordinates as _get_coordinates,
seed_coordinates_if_needed as _seed_coordinates_if_needed,
set_coordinate as _set_coordinate,
)
from .schemas import CoordinateSchema
logger = logging.getLogger(__name__)
# Constants for valid latitude and longitude ranges
MIN_LATITUDE: float = -90.0
MAX_LATITUDE: float = 90.0
MIN_LONGITUDE: float = -180.0
MAX_LONGITUDE: float = 180.0
[docs]
def validate_range(value: float, name: str, min_value: float, max_value: float) -> None:
"""Validate that a numeric value lies within an inclusive range.
Parameters
----------
value : float
Numeric value to validate.
name : str
Descriptive name of the value, used in error messages.
min_value : float
Minimum allowed value (inclusive).
max_value : float
Maximum allowed value (inclusive).
Raises
------
AssertionError
If ``value`` is outside ``[min_value, max_value]``.
Examples
--------
>>> validate_range(0.0, "x", -1.0, 1.0)
>>> validate_range(2.0, "x", -1.0, 1.0) # doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
...
AssertionError: x must be between -1.0 and 1.0, but was 2.0.
See Also
--------
app.coordinates_setup.seed_coordinates_if_needed
app.coordinates_setup.add_coordinate
app.coordinates_setup.retrieve_all_coordinates
"""
assert (
min_value <= value <= max_value
), f"{name} must be between {min_value} and {max_value}, but was {value}."
def _execute_seed(
session: Session, central_latitude: float, central_longitude: float
) -> None:
"""Seed coordinates via manager and log the operation.
This is a thin wrapper that delegates to
``app.coordinates_manager.seed_coordinates_if_needed`` and logs a concise
message upon completion.
Parameters
----------
session : sqlalchemy.orm.Session
Database session controlling the transaction.
central_latitude : float
Latitude of the central coordinate (decimal degrees).
central_longitude : float
Longitude of the central coordinate (decimal degrees).
Raises
------
RuntimeError
If the stored central coordinate mismatches the provided values.
sqlalchemy.exc.SQLAlchemyError
If a database error occurs during persistence or queries.
See Also
--------
app.coordinates_manager.seed_coordinates_if_needed
"""
_seed_coordinates_if_needed(session, central_latitude, central_longitude)
logger.info(
"Seeded coordinates if needed with central point (%f, %f).",
central_latitude,
central_longitude,
)
def _execute_add_coordinate(
session: Session,
latitude: float,
longitude: float,
label: Optional[str],
is_central: bool,
) -> None:
"""Insert a coordinate via manager and log the operation.
Parameters
----------
session : sqlalchemy.orm.Session
Database session controlling the transaction.
latitude : float
Latitude of the coordinate (decimal degrees).
longitude : float
Longitude of the coordinate (decimal degrees).
label : str | None
Optional label for the coordinate.
is_central : bool
Whether the coordinate should be marked as central.
Raises
------
sqlalchemy.exc.SQLAlchemyError
If a database error occurs during insert or commit.
See Also
--------
app.coordinates_manager.set_coordinate
"""
_set_coordinate(
session=session,
latitude=latitude,
longitude=longitude,
label=label,
is_central=is_central,
)
logger.info(
"Added coordinate (%f, %f), label=%s, is_central=%s",
latitude,
longitude,
label,
is_central,
)
[docs]
def seed_coordinates_if_needed(
session: Session, central_latitude: float, central_longitude: float
) -> None:
"""Seed or validate central and surrounding coordinates.
Seeds the table with a central point plus surrounding points if empty; when
not empty, validates that the persisted central point matches the provided
values.
Parameters
----------
session : sqlalchemy.orm.Session
Database session controlling the transaction.
central_latitude : float
Latitude of the central coordinate (decimal degrees).
central_longitude : float
Longitude of the central coordinate (decimal degrees).
Raises
------
RuntimeError
If the stored central coordinate mismatches the provided values.
SQLAlchemyError
If a database error occurs during persistence or queries.
See Also
--------
app.coordinates_manager.seed_coordinates_if_needed
app.coordinates_manager.get_coordinates
app.schemas.CoordinateSchema
Examples
--------
>>> # Typical application code (requires a real DB) # doctest: +SKIP
>>> # from app.database import SessionLocal # doctest: +SKIP
>>> # session = SessionLocal() # doctest: +SKIP
>>> # seed_coordinates_if_needed(session, 59.3, 18.06) # doctest: +SKIP
"""
assert isinstance(
session, Session
), f"session must be a Session instance, got {type(session)}."
validate_range(central_latitude, "central_latitude", MIN_LATITUDE, MAX_LATITUDE)
validate_range(central_longitude, "central_longitude", MIN_LONGITUDE, MAX_LONGITUDE)
try:
_execute_seed(session, central_latitude, central_longitude)
except RuntimeError as error:
logger.error(
"Central coordinate mismatch or missing for (%f, %f): %s",
central_latitude,
central_longitude,
error,
exc_info=True,
)
raise
except SQLAlchemyError as error:
logger.error(
"Database error during coordinate seeding: %s", error, exc_info=True
)
raise
[docs]
def retrieve_all_coordinates(session: Session) -> list[CoordinateSchema]:
"""Retrieve all coordinates from the database.
Parameters
----------
session : sqlalchemy.orm.Session
Open database session.
Returns
-------
list[app.schemas.CoordinateSchema]
All coordinates represented as Pydantic schemas.
Raises
------
SQLAlchemyError
If a database error occurs when querying.
See Also
--------
app.coordinates_manager.get_coordinates
app.schemas.CoordinateSchema
Examples
--------
>>> # Using a real DB session in application code # doctest: +SKIP
>>> # coords = retrieve_all_coordinates(session) # doctest: +SKIP
>>> # isinstance(coords, list) # doctest: +SKIP
True
"""
assert isinstance(
session, Session
), f"session must be a Session instance, got {type(session)}."
try:
coordinates = _get_coordinates(session)
assert isinstance(
coordinates, list
), f"Expected list of coordinates, got {type(coordinates)}."
logger.debug("Retrieved %d coordinates from database.", len(coordinates))
return coordinates
except SQLAlchemyError as error:
logger.error("Failed to retrieve coordinates: %s", error, exc_info=True)
raise
[docs]
def add_coordinate(
session: Session,
latitude: float,
longitude: float,
label: Optional[str] = None,
is_central: bool = False,
) -> None:
"""Add a new coordinate to the database.
Parameters
----------
session : sqlalchemy.orm.Session
Database session controlling the transaction.
latitude : float
Latitude of the coordinate (decimal degrees).
longitude : float
Longitude of the coordinate (decimal degrees).
label : str | None, optional
Optional label for the coordinate.
is_central : bool, optional
Whether the coordinate should be marked as central.
Raises
------
SQLAlchemyError
If a database error occurs while inserting or committing.
See Also
--------
app.coordinates_manager.set_coordinate
app.schemas.CoordinateSchema
"""
assert isinstance(
session, Session
), f"session must be a Session instance, got {type(session)}."
validate_range(latitude, "latitude", MIN_LATITUDE, MAX_LATITUDE)
validate_range(longitude, "longitude", MIN_LONGITUDE, MAX_LONGITUDE)
try:
_execute_add_coordinate(session, latitude, longitude, label, is_central)
except SQLAlchemyError as error:
logger.error("Failed to add coordinate: %s", error, exc_info=True)
raise