from typing import List, Optional from sqlalchemy.orm import Session from schemas.route import RouteCreate, RouteSchema def get_all_routes(db: Session) -> List[RouteSchema]: """ Retrieve all routes from the database. Args: db (Session): SQLAlchemy database session. Returns: List[RouteSchema]: List of all routes. """ routes = db.query(Route).all() return [RouteSchema.from_orm(route) for route in routes] def create_route(db: Session, route: RouteCreate) -> RouteSchema: """ Create a new route in the database. Args: db (Session): SQLAlchemy database session. route (RouteCreate): Route data to create. Returns: RouteSchema: The created route. """ db_route = Route(**route.dict()) db.add(db_route) db.commit() db.refresh(db_route) return RouteSchema.from_orm(db_route) def get_route_by_id(db: Session, route_id: int) -> Optional[RouteSchema]: """ Retrieve a route by its ID from the database. Args: db (Session): SQLAlchemy database session. route_id (int): ID of the route to retrieve. Returns: Optional[RouteSchema]: The route if found, None otherwise. """ route = db.query(Route).filter(Route.id == route_id).first() return RouteSchema.from_orm(route) if route else None def update_route(db: Session, route_id: int, route_data: RouteCreate) -> Optional[RouteSchema]: """ Update a route in the database. Args: db (Session): SQLAlchemy database session. route_id (int): ID of the route to update. route_data (RouteCreate): Updated route data. Returns: Optional[RouteSchema]: The updated route if found, None otherwise. """ route = db.query(Route).filter(Route.id == route_id).first() if route: for key, value in route_data.dict(exclude_unset=True).items(): setattr(route, key, value) db.commit() db.refresh(route) return RouteSchema.from_orm(route) return None def delete_route(db: Session, route_id: int) -> bool: """ Delete a route from the database. Args: db (Session): SQLAlchemy database session. route_id (int): ID of the route to delete. Returns: bool: True if the route was deleted, False otherwise. """ route = db.query(Route).filter(Route.id == route_id).first() if route: db.delete(route) db.commit() return True return False