53 lines
1.6 KiB
Python
53 lines
1.6 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.exc import IntegrityError
|
|
from models.anime import Anime
|
|
from schemas.anime import AnimeCreate, AnimeSchema
|
|
|
|
def get_all_animes(db: Session) -> List[AnimeSchema]:
|
|
"""
|
|
Retrieves all animes from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[AnimeSchema]: A list of all anime objects.
|
|
"""
|
|
animes = db.query(Anime).all()
|
|
return [AnimeSchema.from_orm(anime) for anime in animes]
|
|
|
|
def get_anime_by_name(db: Session, name: str) -> Optional[AnimeSchema]:
|
|
"""
|
|
Retrieves an anime by its name.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
name (str): The name of the anime to retrieve.
|
|
|
|
Returns:
|
|
Optional[AnimeSchema]: The anime object if found, otherwise None.
|
|
"""
|
|
anime = db.query(Anime).filter(Anime.name == name).first()
|
|
return AnimeSchema.from_orm(anime) if anime else None
|
|
|
|
def create_anime(db: Session, anime_data: AnimeCreate) -> AnimeSchema:
|
|
"""
|
|
Creates a new anime in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
anime_data (AnimeCreate): The data for the anime to create.
|
|
|
|
Returns:
|
|
AnimeSchema: The newly created anime object.
|
|
"""
|
|
try:
|
|
db_anime = Anime(**anime_data.dict())
|
|
db.add(db_anime)
|
|
db.commit()
|
|
db.refresh(db_anime)
|
|
return AnimeSchema.from_orm(db_anime)
|
|
except IntegrityError:
|
|
db.rollback()
|
|
raise ValueError(f"Anime with name '{anime_data.name}' already exists.") |