import uuid from typing import List, Optional from sqlalchemy.orm import Session from sqlalchemy import or_ from models.route import Route from schemas.route import RouteCreate, RouteUpdate def get_route_by_id(db: Session, route_id: uuid.UUID) -> Optional[Route]: """ Retrieves a single route by its ID. Args: db (Session): The database session. route_id (UUID): The ID of the route to retrieve. Returns: Optional[Route]: The route object if found, otherwise None. """ return db.query(Route).filter(Route.id == route_id).first() def get_routes(db: Session, name: Optional[str] = None, path: Optional[str] = None) -> List[Route]: """ Retrieves a list of routes based on optional filters. Args: db (Session): The database session. name (Optional[str]): The name of the route to filter by. path (Optional[str]): The path of the route to filter by. Returns: List[Route]: A list of route objects matching the filters. """ query = db.query(Route) if name: query = query.filter(Route.name.ilike(f"%{name}%")) if path: query = query.filter(Route.path.ilike(f"%{path}%")) return query.all() def create_route(db: Session, route_data: RouteCreate) -> Route: """ Creates a new route in the database. Args: db (Session): The database session. route_data (RouteCreate): The data for the route to create. Returns: Route: The newly created route object. """ db_route = Route(**route_data.dict()) db.add(db_route) db.commit() db.refresh(db_route) return db_route def update_route(db: Session, route_id: uuid.UUID, route_data: RouteUpdate) -> Optional[Route]: """ Updates an existing route in the database. Args: db (Session): The database session. route_id (UUID): The ID of the route to update. route_data (RouteUpdate): The updated data for the route. Returns: Optional[Route]: The updated route object if found, otherwise None. """ db_route = get_route_by_id(db, route_id) if not db_route: return None for field, value in route_data.dict(exclude_unset=True).items(): setattr(db_route, field, value) db.commit() db.refresh(db_route) return db_route def delete_route(db: Session, route_id: uuid.UUID) -> bool: """ Deletes a route from the database. Args: db (Session): The database session. route_id (UUID): The ID of the route to delete. Returns: bool: True if the route was deleted, False otherwise. """ db_route = get_route_by_id(db, route_id) if not db_route: return False db.delete(db_route) db.commit() return True def validate_unique_route(db: Session, route_data: RouteCreate) -> bool: """ Validates if a route with the given name or path already exists. Args: db (Session): The database session. route_data (RouteCreate): The data for the new route. Returns: bool: True if the route name and path are unique, False otherwise. """ existing_route = db.query(Route).filter( or_(Route.name == route_data.name, Route.path == route_data.path) ).first() return existing_route is None