51 lines
1.4 KiB
Python
51 lines
1.4 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import exc
|
|
from models.soap import Soap
|
|
from schemas.soap import SoapCreate, SoapSchema
|
|
|
|
def get_all_soaps(db: Session) -> List[SoapSchema]:
|
|
"""
|
|
Retrieves all soaps from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[SoapSchema]: A list of all soap objects.
|
|
"""
|
|
return db.query(Soap).all()
|
|
|
|
def create_soap(db: Session, soap_data: SoapCreate) -> SoapSchema:
|
|
"""
|
|
Creates a new soap in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
soap_data (SoapCreate): The data for the soap to create.
|
|
|
|
Returns:
|
|
SoapSchema: The newly created soap object.
|
|
"""
|
|
try:
|
|
db_soap = Soap(name=soap_data.name, color=soap_data.color)
|
|
db.add(db_soap)
|
|
db.commit()
|
|
db.refresh(db_soap)
|
|
return db_soap
|
|
except exc.IntegrityError:
|
|
db.rollback()
|
|
raise ValueError("Soap name already exists")
|
|
|
|
def get_soap_by_name(db: Session, soap_name: str) -> Optional[SoapSchema]:
|
|
"""
|
|
Retrieves a single soap by its name.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
soap_name (str): The name of the soap to retrieve.
|
|
|
|
Returns:
|
|
Optional[SoapSchema]: The soap object if found, otherwise None.
|
|
"""
|
|
return db.query(Soap).filter(Soap.name == soap_name).first() |