dhit-b5680e/helpers/flower_helpers.py

116 lines
3.2 KiB
Python

from typing import List, Optional
from sqlalchemy.orm import Session
from sqlalchemy import and_
from models.flower import Flower
from schemas.flower import FlowerCreate, FlowerUpdate
import uuid
def get_all_flowers(db: Session) -> List[Flower]:
"""
Retrieves all flowers from the database.
Args:
db (Session): The database session.
Returns:
List[Flower]: A list of all flower objects.
"""
return db.query(Flower).all()
def get_flower_by_id(db: Session, flower_id: uuid.UUID) -> Optional[Flower]:
"""
Retrieves a single flower by its ID.
Args:
db (Session): The database session.
flower_id (UUID): The ID of the flower to retrieve.
Returns:
Optional[Flower]: The flower object if found, otherwise None.
"""
return db.query(Flower).filter(Flower.id == flower_id).first()
def get_flowers_by_name_and_species(db: Session, name: Optional[str] = None, species: Optional[str] = None) -> List[Flower]:
"""
Retrieves flowers based on name and/or species.
Args:
db (Session): The database session.
name (Optional[str]): The name of the flower to filter by.
species (Optional[str]): The species of the flower to filter by.
Returns:
List[Flower]: A list of flower objects matching the filters.
"""
filters = []
if name:
filters.append(Flower.name == name)
if species:
filters.append(Flower.species == species)
query = db.query(Flower)
if filters:
query = query.filter(and_(*filters))
return query.all()
def create_flower(db: Session, flower_data: FlowerCreate) -> Flower:
"""
Creates a new flower in the database.
Args:
db (Session): The database session.
flower_data (FlowerCreate): The data for the flower to create.
Returns:
Flower: The newly created flower object.
"""
db_flower = Flower(**flower_data.dict())
db.add(db_flower)
db.commit()
db.refresh(db_flower)
return db_flower
def update_flower(db: Session, flower_id: uuid.UUID, flower_data: FlowerUpdate) -> Optional[Flower]:
"""
Updates an existing flower in the database.
Args:
db (Session): The database session.
flower_id (UUID): The ID of the flower to update.
flower_data (FlowerUpdate): The updated data for the flower.
Returns:
Optional[Flower]: The updated flower object if found, otherwise None.
"""
db_flower = db.query(Flower).filter(Flower.id == flower_id).first()
if not db_flower:
return None
update_data = flower_data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(db_flower, key, value)
db.add(db_flower)
db.commit()
db.refresh(db_flower)
return db_flower
def delete_flower(db: Session, flower_id: uuid.UUID) -> bool:
"""
Deletes a flower from the database.
Args:
db (Session): The database session.
flower_id (UUID): The ID of the flower to delete.
Returns:
bool: True if the flower was deleted, False otherwise.
"""
db_flower = db.query(Flower).filter(Flower.id == flower_id).first()
if not db_flower:
return False
db.delete(db_flower)
db.commit()
return True