116 lines
3.2 KiB
Python
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 |