65 lines
1.7 KiB
Python
65 lines
1.7 KiB
Python
# crud.py
|
|
from typing import List
|
|
from sqlalchemy.orm import Session
|
|
from exceptions import AlbumAlreadyExistError, AlbumNotFoundError
|
|
from models import Album
|
|
from schemas import CreateAndUpdateAlbum
|
|
|
|
|
|
# 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 a 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
|