174 lines
4.9 KiB
Python
174 lines
4.9 KiB
Python
from typing import List, Optional, Dict, Any
|
|
from uuid import UUID
|
|
from sqlalchemy.orm import Session
|
|
from datetime import datetime
|
|
|
|
from models.fruit import Fruit
|
|
from schemas.fruit import FruitCreate, FruitUpdate, FruitSchema
|
|
|
|
def get_all_fruits(
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
sort_by: str = "created_at",
|
|
sort_desc: bool = True
|
|
) -> List[FruitSchema]:
|
|
"""
|
|
Returns a static list of fruits with pagination and sorting.
|
|
|
|
Args:
|
|
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[FruitSchema]: List of fruit objects
|
|
"""
|
|
# Static list of fruits
|
|
static_fruits = [
|
|
{
|
|
"id": UUID("f47ac10b-58cc-4372-a567-0e02b2c3d479"),
|
|
"name": "Apple",
|
|
"description": "A sweet and crunchy fruit",
|
|
"created_at": datetime(2023, 1, 1, 12, 0),
|
|
"updated_at": datetime(2023, 1, 1, 12, 0)
|
|
},
|
|
{
|
|
"id": UUID("f47ac10b-58cc-4372-a567-0e02b2c3d480"),
|
|
"name": "Banana",
|
|
"description": "A yellow tropical fruit",
|
|
"created_at": datetime(2023, 1, 2, 12, 0),
|
|
"updated_at": datetime(2023, 1, 2, 12, 0)
|
|
},
|
|
{
|
|
"id": UUID("f47ac10b-58cc-4372-a567-0e02b2c3d481"),
|
|
"name": "Orange",
|
|
"description": "A citrus fruit rich in vitamin C",
|
|
"created_at": datetime(2023, 1, 3, 12, 0),
|
|
"updated_at": datetime(2023, 1, 3, 12, 0)
|
|
}
|
|
]
|
|
|
|
# Convert dictionary items to FruitSchema objects
|
|
fruits = [FruitSchema(**fruit) for fruit in static_fruits]
|
|
|
|
# Sort the fruits
|
|
if sort_by in FruitSchema.__fields__:
|
|
fruits.sort(
|
|
key=lambda x: getattr(x, sort_by),
|
|
reverse=sort_desc
|
|
)
|
|
|
|
# Apply pagination
|
|
return fruits[skip:skip + limit]
|
|
|
|
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 |