114 lines
3.2 KiB
Python
114 lines
3.2 KiB
Python
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 |