90 lines
2.2 KiB
Python
90 lines
2.2 KiB
Python
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 models.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
|