python-webservice/models/album.py

from sqlalchemy.schema import Column
from sqlalchemy.types import String, Integer, Float
from database import Base
from typing import List
from sqlalchemy.orm import Session
from schemas import CreateAndUpdateAlbum


class Album(Base):
    __tablename__ = "albums"

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


class AlbumException(Exception):
    ...


class AlbumNotFoundError(AlbumException):
    def __init__(self):
        self.status_code = 404
        self.detail = "Album  Not Found"


class AlbumAlreadyExistError(AlbumException):
    def __init__(self):
        self.status_code = 409
        self.detail = "Album  Already Exists"


# Function to get list of album info
def get_all_albums(session: Session, limit: int, offset: int) -> List[Album]:
    return session.query(Album).offset(offset).limit(limit).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: CreateAndUpdateAlbum) -> 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: CreateAndUpdateAlbum) -> 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