from typing import List, Optional, Dict, Any from uuid import UUID from sqlalchemy.orm import Session from sqlalchemy import desc from models.fruit import Fruit from schemas.fruit import FruitCreate, FruitUpdate def get_all_fruits( db: Session, skip: int = 0, limit: int = 100, sort_by: str = "created_at", sort_desc: bool = True ) -> List[Fruit]: """ Retrieves all fruits from the database with pagination and sorting. Args: db (Session): The database session skip (int): Number of records to skip limit (int): Maximum number of records to return sort_by (str): Field to sort by sort_desc (bool): Sort in descending order if True Returns: List[Fruit]: List of fruit objects """ query = db.query(Fruit) if hasattr(Fruit, sort_by): order_criteria = desc(getattr(Fruit, sort_by)) if sort_desc else getattr(Fruit, sort_by) query = query.order_by(order_criteria) return query.offset(skip).limit(limit).all() def get_fruit_by_id(db: Session, fruit_id: UUID) -> Optional[Fruit]: """ Retrieves a single fruit by its ID. Args: db (Session): The database session fruit_id (UUID): The ID of the fruit to retrieve Returns: Optional[Fruit]: The fruit object if found, otherwise None """ return db.query(Fruit).filter(Fruit.id == fruit_id).first() def create_fruit(db: Session, fruit_data: FruitCreate) -> Fruit: """ Creates a new fruit in the database. Args: db (Session): The database session fruit_data (FruitCreate): The data for the fruit to create Returns: Fruit: The newly created fruit object """ db_fruit = Fruit(**fruit_data.dict()) db.add(db_fruit) db.commit() db.refresh(db_fruit) return db_fruit def update_fruit(db: Session, fruit_id: UUID, fruit_data: FruitUpdate) -> Optional[Fruit]: """ Updates an existing fruit in the database. Args: db (Session): The database session fruit_id (UUID): The ID of the fruit to update fruit_data (FruitUpdate): The update data Returns: Optional[Fruit]: The updated fruit object if found, otherwise None """ db_fruit = get_fruit_by_id(db, fruit_id) if not db_fruit: return None update_data = fruit_data.dict(exclude_unset=True) for field, value in update_data.items(): setattr(db_fruit, field, value) db.commit() db.refresh(db_fruit) return db_fruit def delete_fruit(db: Session, fruit_id: UUID) -> bool: """ Deletes a fruit from the database. Args: db (Session): The database session fruit_id (UUID): The ID of the fruit to delete Returns: bool: True if deletion was successful, False if fruit wasn't found """ db_fruit = get_fruit_by_id(db, fruit_id) if not db_fruit: return False db.delete(db_fruit) db.commit() return True def search_fruits_by_name(db: Session, name: str, limit: int = 10) -> List[Fruit]: """ Searches for fruits by name (case-insensitive partial match). Args: db (Session): The database session name (str): The name to search for limit (int): Maximum number of results to return Returns: List[Fruit]: List of matching fruit objects """ return db.query(Fruit).filter( Fruit.name.ilike(f"%{name}%") ).limit(limit).all() def validate_fruit_data(data: Dict[str, Any]) -> bool: """ Validates fruit input data. Args: data (Dict[str, Any]): The input data to validate Returns: bool: True if the data is valid, False otherwise """ if not data: return False if "name" not in data or not isinstance(data["name"], str) or len(data["name"]) < 1: return False if "description" in data and data["description"] is not None: if not isinstance(data["description"], str): return False return True