92 lines
2.6 KiB
Python
92 lines
2.6 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from models.apple import Apple
|
|
from schemas.apple import AppleSchema, AppleCreate, AppleUpdate
|
|
from uuid import UUID
|
|
|
|
def get_all_apples(db: Session) -> List[AppleSchema]:
|
|
"""
|
|
Retrieves all apples from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[AppleSchema]: A list of all apple objects.
|
|
"""
|
|
apples = db.query(Apple).all()
|
|
return [AppleSchema.from_orm(apple) for apple in apples]
|
|
|
|
def get_apple_by_name(db: Session, name: str) -> Optional[AppleSchema]:
|
|
"""
|
|
Retrieves an apple from the database by its name.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
name (str): The name of the apple to retrieve.
|
|
|
|
Returns:
|
|
Optional[AppleSchema]: The apple object if found, otherwise None.
|
|
"""
|
|
apple = db.query(Apple).filter(Apple.name == name).first()
|
|
return AppleSchema.from_orm(apple) if apple else None
|
|
|
|
def create_apple(db: Session, apple_data: AppleCreate) -> AppleSchema:
|
|
"""
|
|
Creates a new apple in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
apple_data (AppleCreate): The data for the apple to create.
|
|
|
|
Returns:
|
|
AppleSchema: The newly created apple object.
|
|
"""
|
|
db_apple = Apple(**apple_data.dict())
|
|
db.add(db_apple)
|
|
db.commit()
|
|
db.refresh(db_apple)
|
|
return AppleSchema.from_orm(db_apple)
|
|
|
|
def update_apple(db: Session, apple_id: UUID, apple_data: AppleUpdate) -> Optional[AppleSchema]:
|
|
"""
|
|
Updates an existing apple in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
apple_id (UUID): The ID of the apple to update.
|
|
apple_data (AppleUpdate): The updated data for the apple.
|
|
|
|
Returns:
|
|
Optional[AppleSchema]: The updated apple object if found, otherwise None.
|
|
"""
|
|
db_apple = db.query(Apple).filter(Apple.id == apple_id).first()
|
|
if not db_apple:
|
|
return None
|
|
|
|
update_data = apple_data.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_apple, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_apple)
|
|
return AppleSchema.from_orm(db_apple)
|
|
|
|
def delete_apple(db: Session, apple_id: UUID) -> bool:
|
|
"""
|
|
Deletes an apple from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
apple_id (UUID): The ID of the apple to delete.
|
|
|
|
Returns:
|
|
bool: True if the apple was successfully deleted, False otherwise.
|
|
"""
|
|
db_apple = db.query(Apple).filter(Apple.id == apple_id).first()
|
|
if not db_apple:
|
|
return False
|
|
|
|
db.delete(db_apple)
|
|
db.commit()
|
|
return True |