diff --git a/helpers/user_helpers.py b/helpers/user_helpers.py new file mode 100644 index 0000000..c3d514b --- /dev/null +++ b/helpers/user_helpers.py @@ -0,0 +1,86 @@ +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 \ No newline at end of file