163 lines
4.7 KiB
Python
163 lines
4.7 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import desc, asc
|
|
from uuid import UUID
|
|
|
|
# Import the model and schema
|
|
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: Optional[str] = None,
|
|
sort_order: str = "asc",
|
|
filter_seasonal: Optional[bool] = None
|
|
) -> List[Fruit]:
|
|
"""
|
|
Retrieves all fruits from the database with optional pagination, sorting, and filtering.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
skip (int, optional): Number of records to skip. Defaults to 0.
|
|
limit (int, optional): Maximum number of records to return. Defaults to 100.
|
|
sort_by (Optional[str], optional): Field to sort by. Defaults to None.
|
|
sort_order (str, optional): Sort direction ('asc' or 'desc'). Defaults to "asc".
|
|
filter_seasonal (Optional[bool], optional): Filter by seasonal status. Defaults to None.
|
|
|
|
Returns:
|
|
List[Fruit]: A list of fruit objects.
|
|
"""
|
|
query = db.query(Fruit)
|
|
|
|
# Apply seasonal filter if specified
|
|
if filter_seasonal is not None:
|
|
query = query.filter(Fruit.is_seasonal == filter_seasonal)
|
|
|
|
# Apply sorting if specified
|
|
if sort_by and hasattr(Fruit, sort_by):
|
|
sort_column = getattr(Fruit, sort_by)
|
|
if sort_order.lower() == "desc":
|
|
query = query.order_by(desc(sort_column))
|
|
else:
|
|
query = query.order_by(asc(sort_column))
|
|
|
|
# Apply pagination
|
|
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 get_fruits_by_color(db: Session, color: str) -> List[Fruit]:
|
|
"""
|
|
Retrieves all fruits matching a specific color.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
color (str): The color to filter by.
|
|
|
|
Returns:
|
|
List[Fruit]: A list of fruit objects with the specified color.
|
|
"""
|
|
return db.query(Fruit).filter(Fruit.color == color).all()
|
|
|
|
def get_seasonal_fruits(db: Session) -> List[Fruit]:
|
|
"""
|
|
Retrieves all seasonal fruits from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[Fruit]: A list of seasonal fruit objects.
|
|
"""
|
|
return db.query(Fruit).filter(Fruit.is_seasonal).all()
|
|
|
|
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 data to update the fruit with.
|
|
|
|
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 only the fields that are provided
|
|
update_data = fruit_data.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_fruit, key, 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 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 search_fruits(db: Session, search_term: str) -> List[Fruit]:
|
|
"""
|
|
Searches for fruits by name or description.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
search_term (str): The term to search for.
|
|
|
|
Returns:
|
|
List[Fruit]: A list of fruit objects matching the search term.
|
|
"""
|
|
search_pattern = f"%{search_term}%"
|
|
return db.query(Fruit).filter(
|
|
(Fruit.name.ilike(search_pattern)) |
|
|
(Fruit.description.ilike(search_pattern))
|
|
).all() |