from typing import List, Optional, Dict, Any from uuid import UUID from sqlalchemy.orm import Session from sqlalchemy.exc import IntegrityError from models.fruit import Fruit from schemas.fruit import FruitCreate, FruitUpdate def get_all_fruits(db: Session) -> List[Fruit]: """ Retrieves all fruits from the database. Args: db (Session): The database session. Returns: List[Fruit]: A list of all fruit objects. """ return db.query(Fruit).order_by(Fruit.name).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 get_fruit_by_name(db: Session, name: str) -> Optional[Fruit]: """ Retrieves a single fruit by its name. Args: db (Session): The database session. name (str): The name of the fruit to retrieve. Returns: Optional[Fruit]: The fruit object if found, otherwise None. """ return db.query(Fruit).filter(Fruit.name.ilike(f"%{name}%")).first() def create_fruit(db: Session, fruit_data: FruitCreate) -> Optional[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: Optional[Fruit]: The newly created fruit object, or None if creation fails. """ try: db_fruit = Fruit(**fruit_data.dict(exclude_unset=True)) db.add(db_fruit) db.commit() db.refresh(db_fruit) return db_fruit except IntegrityError: db.rollback() return None 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 updated fruit data. Returns: Optional[Fruit]: The updated fruit object if found and updated, 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) try: db.commit() db.refresh(db_fruit) return db_fruit except IntegrityError: db.rollback() return None 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 the fruit was deleted, False if not 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 validate_fruit_data(data: Dict[str, Any]) -> bool: """ Validates fruit input data dictionary. 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"]) > 255: return False if "description" in data and not isinstance(data["description"], (str, type(None))): return False return True