43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import insert
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.location import Location
|
|
from app.schemas.location import LocationRecordRequest
|
|
|
|
|
|
def _parse_optional_float_compat(value: str | None) -> float:
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return 0.0
|
|
|
|
|
|
def _parse_required_float(value: str, field_name: str) -> float:
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError(f"Invalid numeric value for {field_name}") from exc
|
|
|
|
|
|
def _utc_now_rfc3339() -> str:
|
|
now = datetime.now(timezone.utc).replace(microsecond=0)
|
|
return now.isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
def record_location(session: Session, payload: LocationRecordRequest) -> None:
|
|
stmt = (
|
|
insert(Location)
|
|
.prefix_with("OR IGNORE")
|
|
.values(
|
|
person=payload.person,
|
|
datetime=_utc_now_rfc3339(),
|
|
latitude=_parse_required_float(payload.latitude, "latitude"),
|
|
longitude=_parse_required_float(payload.longitude, "longitude"),
|
|
altitude=_parse_optional_float_compat(payload.altitude),
|
|
)
|
|
)
|
|
session.execute(stmt)
|
|
session.commit()
|