from typing import List, Optional from sqlalchemy.orm import Session from app.models.location import Location from app.schemas.location import LocationCreate, LocationUpdate def get(db: Session, location_id: int) -> Optional[Location]: """ Get a location by ID. Args: db: Database session location_id: Location ID Returns: Location object if found, None otherwise """ return db.query(Location).filter(Location.id == location_id).first() def get_user_locations(db: Session, user_id: int) -> List[Location]: """ Get all locations for a user. Args: db: Database session user_id: User ID Returns: List of location objects """ return db.query(Location).filter(Location.user_id == user_id).all() def get_user_default_location(db: Session, user_id: int) -> Optional[Location]: """ Get the default location for a user. Args: db: Database session user_id: User ID Returns: Default location object if found, None otherwise """ return db.query(Location).filter( Location.user_id == user_id, Location.is_default ).first() def create(db: Session, obj_in: LocationCreate, user_id: int) -> Location: """ Create a new location. Args: db: Database session obj_in: Location creation data user_id: User ID Returns: Created location object """ # If this is set as default, unset any existing default if obj_in.is_default: existing_default = get_user_default_location(db, user_id) if existing_default: existing_default.is_default = False db.add(existing_default) db_obj = Location( **obj_in.dict(), user_id=user_id ) db.add(db_obj) db.commit() db.refresh(db_obj) return db_obj def update(db: Session, db_obj: Location, obj_in: LocationUpdate) -> Location: """ Update a location. Args: db: Database session db_obj: Location object to update obj_in: Location update data Returns: Updated location object """ update_data = obj_in.dict(exclude_unset=True) # If this is set as default, unset any existing default if update_data.get("is_default"): existing_default = get_user_default_location(db, db_obj.user_id) if existing_default and existing_default.id != db_obj.id: existing_default.is_default = False db.add(existing_default) for field, value in update_data.items(): setattr(db_obj, field, value) db.add(db_obj) db.commit() db.refresh(db_obj) return db_obj def delete(db: Session, location_id: int) -> Optional[Location]: """ Delete a location. Args: db: Database session location_id: Location ID Returns: Deleted location object if found, None otherwise """ location = db.query(Location).filter(Location.id == location_id).first() if location: db.delete(location) db.commit() return location