python-webservice/models/album.py

from sqlalchemy.schema import Column
from sqlalchemy.types import String, Integer, Float
from typing import List
from sqlalchemy.orm import Session

from models import schemas

from database import Base
from models.exceptions import AlbumNotFoundError, AlbumAlreadyExistError


class Album(Base):
    __tablename__ = "albums"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    artist = Column(String)
    price = Column(Float)


# Function to get list of album info
def get_all_albums(session: Session) -> List[Album]:
    return session.query(Album).all()


# Function to  get info of a particular album
def get_album_by_id(session: Session, _id: int) -> Album:
    album = session.query(Album).get(_id)

    if album is None:
        raise AlbumNotFoundError

    return album


# Function to add a new album info to the database
def create_album(session: Session, album: schemas.AlbumBase) -> Album:
    album_details = session.query(Album).filter(Album.title == album.title, Album.artist == album.artist).first()
    if album_details is not None:
        raise AlbumAlreadyExistError

    new_album = Album(**album.dict())
    session.add(new_album)
    session.commit()
    session.refresh(new_album)
    return new_album


# Function to update details of the album
def update_album(session: Session, _id: int, info_update: schemas.AlbumBase) -> Album:
    album = get_album_by_id(session, _id)

    if album is None:
        raise AlbumNotFoundError

    album.title = info_update.title
    album.artist = info_update.artist
    album.price = info_update.price

    session.commit()
    session.refresh(album)

    return album


# Function to delete an album info from the db
def delete_album(session: Session, _id: int):
    album = get_album_by_id(session, _id)

    if album is None:
        raise AlbumNotFoundError

    session.delete(album)
    session.commit()

    return