mirror of
https://github.com/Dvorinka/swingmusic-extended.git
synced 2026-06-04 20:43:04 +00:00
Move MongoDB classes into a seperate db module (#62)
* move album class into mongodb dir * inherit AlbumMethods at DB Initialization level * move album -> db/mongodb * move mongodb classes into separate package
This commit is contained in:
Mungai Geoffrey
@@ -0,0 +1,80 @@
|
||||
"""
|
||||
This module creates and initiliazes a MongoDB instance. It also contains the
|
||||
`convert_one()` and `conver_many()` methods for converting MongoDB cursors to Python dicts.
|
||||
"""
|
||||
import json
|
||||
|
||||
import pymongo
|
||||
from app.db import AlbumMethods
|
||||
from app.db import ArtistMethods
|
||||
from app.db import PlaylistMethods
|
||||
from app.db import TrackMethods
|
||||
from bson import json_util
|
||||
|
||||
|
||||
class Mongo:
|
||||
"""
|
||||
The base class for all mongodb classes.
|
||||
"""
|
||||
|
||||
def __init__(self, database):
|
||||
mongo_uri = pymongo.MongoClient()
|
||||
self.db = mongo_uri[database]
|
||||
|
||||
|
||||
class MongoAlbums(Mongo, AlbumMethods):
|
||||
|
||||
def __init__(self):
|
||||
super(MongoAlbums, self).__init__("ALICE_ALBUMS")
|
||||
self.collection = self.db["ALL_ALBUMS"]
|
||||
|
||||
|
||||
class MongoArtists(Mongo, ArtistMethods):
|
||||
|
||||
def __init__(self):
|
||||
super(MongoArtists, self).__init__("ALICE_ARTISTS")
|
||||
self.collection = self.db["ALL_ARTISTS"]
|
||||
|
||||
|
||||
class MongoPlaylists(Mongo, PlaylistMethods):
|
||||
|
||||
def __init__(self):
|
||||
super(MongoPlaylists, self).__init__("ALICE_PLAYLISTS")
|
||||
self.collection = self.db["ALL_PLAYLISTS"]
|
||||
|
||||
|
||||
class MongoTracks(Mongo, TrackMethods):
|
||||
|
||||
def __init__(self):
|
||||
super(MongoTracks, self).__init__("ALICE_MUSIC_TRACKS")
|
||||
self.collection = self.db["ALL_TRACKS"]
|
||||
|
||||
|
||||
# ====================================================================== #
|
||||
# cursor convertion methods
|
||||
|
||||
|
||||
def convert_one(song):
|
||||
"""
|
||||
Converts a single mongodb cursor to a json object.
|
||||
"""
|
||||
json_song = json.dumps(song, default=json_util.default)
|
||||
loaded_song = json.loads(json_song)
|
||||
|
||||
return loaded_song
|
||||
|
||||
|
||||
def convert_many(array):
|
||||
"""
|
||||
Converts a list of mongodb cursors to a list of json objects.
|
||||
"""
|
||||
|
||||
songs = []
|
||||
|
||||
for song in array:
|
||||
json_song = json.dumps(song, default=json_util.default)
|
||||
loaded_song = json.loads(json_song)
|
||||
|
||||
songs.append(loaded_song)
|
||||
|
||||
return songs
|
||||
@@ -0,0 +1,60 @@
|
||||
"""
|
||||
This file contains the Album class for interacting with
|
||||
album documents in MongoDB.
|
||||
"""
|
||||
from app import db
|
||||
from app.db.mongodb import convert_many
|
||||
from app.db.mongodb import convert_one
|
||||
from app.db.mongodb import MongoAlbums
|
||||
from app.models import Album
|
||||
from bson import ObjectId
|
||||
|
||||
|
||||
class Albums(MongoAlbums):
|
||||
"""
|
||||
The class for all album-related database operations.
|
||||
"""
|
||||
|
||||
def insert_album(self, album: Album) -> None:
|
||||
"""
|
||||
Inserts a new album object into the database.
|
||||
"""
|
||||
album = album.__dict__
|
||||
return self.collection.update_one(
|
||||
{
|
||||
"album": album["title"],
|
||||
"artist": album["artist"]
|
||||
},
|
||||
{
|
||||
"$set": album
|
||||
},
|
||||
upsert=True,
|
||||
).upserted_id
|
||||
|
||||
def get_all_albums(self) -> list:
|
||||
"""
|
||||
Returns all the albums in the database.
|
||||
"""
|
||||
albums = self.collection.find()
|
||||
return convert_many(albums)
|
||||
|
||||
def get_album_by_id(self, id: str) -> dict:
|
||||
"""
|
||||
Returns a single album matching the id in the query params.
|
||||
"""
|
||||
album = self.collection.find_one({"_id": ObjectId(id)})
|
||||
return convert_one(album)
|
||||
|
||||
def get_album_by_name(self, name: str, artist: str) -> dict:
|
||||
"""
|
||||
Returns a single album matching the name in the query params.
|
||||
"""
|
||||
album = self.collection.find_one({"album": name, "artist": artist})
|
||||
return convert_one(album)
|
||||
|
||||
def get_album_by_artist(self, name: str) -> dict:
|
||||
"""
|
||||
Returns a single album matching the artist in the query params.
|
||||
"""
|
||||
album = self.collection.find_one({"albumartist": name})
|
||||
return convert_one(album)
|
||||
@@ -0,0 +1,38 @@
|
||||
"""
|
||||
This file contains the Artists class for interacting with artist documents in MongoDB.
|
||||
"""
|
||||
from app.db.mongodb import MongoArtists
|
||||
from bson import ObjectId
|
||||
|
||||
|
||||
class Artists(MongoArtists):
|
||||
"""
|
||||
The artist class for all artist related database operations.
|
||||
"""
|
||||
|
||||
def insert_artist(self, artist_obj: dict) -> None:
|
||||
"""
|
||||
Inserts an artist into the database.
|
||||
"""
|
||||
self.collection.update_one(artist_obj, {
|
||||
"$set": artist_obj
|
||||
},
|
||||
upsert=True).upserted_id
|
||||
|
||||
def get_all_artists(self) -> list:
|
||||
"""
|
||||
Returns a list of all artists in the database.
|
||||
"""
|
||||
return self.collection.find()
|
||||
|
||||
def get_artist_by_id(self, artist_id: str) -> dict:
|
||||
"""
|
||||
Returns an artist matching the mongo Id.
|
||||
"""
|
||||
return self.collection.find_one({"_id": ObjectId(artist_id)})
|
||||
|
||||
def get_artists_by_name(self, query: str):
|
||||
"""
|
||||
Returns all the artists matching the query.
|
||||
"""
|
||||
return self.collection.find({"name": query}).limit(20)
|
||||
@@ -0,0 +1,78 @@
|
||||
"""
|
||||
This file contains the Playlists class for interacting with the playlist documents in MongoDB.
|
||||
"""
|
||||
from app.db.mongodb import convert_many
|
||||
from app.db.mongodb import convert_one
|
||||
from app.db.mongodb import MongoPlaylists
|
||||
from app.helpers import create_new_date
|
||||
from bson import ObjectId
|
||||
|
||||
|
||||
class Playlists(MongoPlaylists):
|
||||
"""
|
||||
The class for all playlist-related database operations.
|
||||
"""
|
||||
|
||||
def insert_playlist(self, playlist: dict) -> None:
|
||||
"""
|
||||
Inserts a new playlist object into the database.
|
||||
"""
|
||||
return self.collection.update_one(
|
||||
{
|
||||
"name": playlist["name"]
|
||||
},
|
||||
{
|
||||
"$set": playlist
|
||||
},
|
||||
upsert=True,
|
||||
).upserted_id
|
||||
|
||||
def get_all_playlists(self) -> list:
|
||||
"""
|
||||
Returns all the playlists in the database.
|
||||
"""
|
||||
playlists = self.collection.find()
|
||||
return convert_many(playlists)
|
||||
|
||||
def get_playlist_by_id(self, id: str) -> dict:
|
||||
"""
|
||||
Returns a single playlist matching the id in the query params.
|
||||
"""
|
||||
playlist = self.collection.find_one({"_id": ObjectId(id)})
|
||||
return convert_one(playlist)
|
||||
|
||||
def add_track_to_playlist(self, playlistid: str, track: dict) -> None:
|
||||
"""
|
||||
Adds a track to a playlist.
|
||||
"""
|
||||
date = create_new_date()
|
||||
|
||||
return self.collection.update_one(
|
||||
{
|
||||
"_id": ObjectId(playlistid),
|
||||
},
|
||||
{
|
||||
"$push": {
|
||||
"pre_tracks": track
|
||||
},
|
||||
"$set": {
|
||||
"lastUpdated": date
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
def get_playlist_by_name(self, name: str) -> dict:
|
||||
"""
|
||||
Returns a single playlist matching the name in the query params.
|
||||
"""
|
||||
playlist = self.collection.find_one({"name": name})
|
||||
return convert_one(playlist)
|
||||
|
||||
def update_playlist(self, playlistid: str, playlist: dict) -> None:
|
||||
"""
|
||||
Updates a playlist.
|
||||
"""
|
||||
return self.collection.update_one(
|
||||
{"_id": ObjectId(playlistid)},
|
||||
{"$set": playlist},
|
||||
)
|
||||
@@ -0,0 +1,30 @@
|
||||
# """
|
||||
# This file contains the TrackColors class for interacting with Track colors documents in MongoDB.
|
||||
# """
|
||||
# from app import db
|
||||
# class TrackColors(db.Mongo):
|
||||
# """
|
||||
# The class for all track-related database operations.
|
||||
# """
|
||||
# def __init__(self):
|
||||
# super(TrackColors, self).__init__("ALICE_TRACK_COLORS")
|
||||
# self.collection = self.db["TRACK_COLORS"]
|
||||
# def insert_track_color(self, track_color: dict) -> None:
|
||||
# """
|
||||
# Inserts a new track object into the database.
|
||||
# """
|
||||
# return self.collection.update_one(
|
||||
# {
|
||||
# "filepath": track_color["filepath"]
|
||||
# },
|
||||
# {
|
||||
# "$set": track_color
|
||||
# },
|
||||
# upsert=True,
|
||||
# ).upserted_id
|
||||
# def get_track_color_by_track(self, filepath: str) -> dict:
|
||||
# """
|
||||
# Returns a track color object by its filepath.
|
||||
# """
|
||||
# track_color = self.collection.find_one({"filepath": filepath})
|
||||
# return db.convert_one(track_color)
|
||||
@@ -0,0 +1,148 @@
|
||||
"""
|
||||
This file contains the AllSongs class for interacting with track documents in MongoDB.
|
||||
"""
|
||||
from app.db.mongodb import convert_many
|
||||
from app.db.mongodb import convert_one
|
||||
from app.db.mongodb import MongoTracks
|
||||
from bson import ObjectId
|
||||
|
||||
|
||||
class Tracks(MongoTracks):
|
||||
"""
|
||||
The class for all track-related database operations.
|
||||
"""
|
||||
|
||||
# def drop_db(self):
|
||||
# self.collection.drop()
|
||||
|
||||
def insert_song(self, song_obj: dict) -> str:
|
||||
"""
|
||||
Inserts a new track object into the database.
|
||||
"""
|
||||
return self.collection.update_one({
|
||||
"filepath": song_obj["filepath"]
|
||||
}, {
|
||||
"$set": song_obj
|
||||
},
|
||||
upsert=True).upserted_id
|
||||
|
||||
def get_all_tracks(self) -> list:
|
||||
"""
|
||||
Returns all tracks in the database.
|
||||
"""
|
||||
return convert_many(self.collection.find())
|
||||
|
||||
def get_song_by_id(self, file_id: str) -> dict:
|
||||
"""
|
||||
Returns a track object by its mongodb id.
|
||||
"""
|
||||
song = self.collection.find_one({"_id": ObjectId(file_id)})
|
||||
return convert_one(song)
|
||||
|
||||
def get_song_by_album(self, name: str, artist: str) -> dict:
|
||||
"""
|
||||
Returns a single track matching the album in the query params.
|
||||
"""
|
||||
song = self.collection.find_one({"album": name, "albumartist": artist})
|
||||
return convert_one(song)
|
||||
|
||||
def search_songs_by_album(self, query: str) -> list:
|
||||
"""
|
||||
Returns all the songs matching the albums in the query params (using regex).
|
||||
"""
|
||||
songs = self.collection.find(
|
||||
{"album": {
|
||||
"$regex": query,
|
||||
"$options": "i"
|
||||
}})
|
||||
return convert_many(songs)
|
||||
|
||||
def search_songs_by_artist(self, query: str) -> list:
|
||||
"""
|
||||
Returns all the songs matching the artists in the query params.
|
||||
"""
|
||||
songs = self.collection.find(
|
||||
{"artists": {
|
||||
"$regex": query,
|
||||
"$options": "i"
|
||||
}})
|
||||
return convert_many(songs)
|
||||
|
||||
def find_song_by_title(self, query: str) -> list:
|
||||
"""
|
||||
Finds all the tracks matching the title in the query params.
|
||||
"""
|
||||
song = self.collection.find(
|
||||
{"title": {
|
||||
"$regex": query,
|
||||
"$options": "i"
|
||||
}})
|
||||
return convert_many(song)
|
||||
|
||||
def find_songs_by_album(self, name: str, artist: str) -> list:
|
||||
"""
|
||||
Returns all the tracks exactly matching the album in the query params.
|
||||
"""
|
||||
songs = self.collection.find({"album": name, "albumartist": artist})
|
||||
return convert_many(songs)
|
||||
|
||||
def find_songs_by_folder(self, query: str) -> list:
|
||||
"""
|
||||
Returns a sorted list of all the tracks exactly matching the folder in the query params
|
||||
"""
|
||||
songs = self.collection.find({
|
||||
"folder": query
|
||||
}).sort("title", db.pymongo.ASCENDING)
|
||||
return convert_many(songs)
|
||||
|
||||
def find_songs_by_folder_og(self, query: str) -> list:
|
||||
"""
|
||||
Returns an unsorted list of all the tracks exactly matching the folder in the query params
|
||||
"""
|
||||
songs = self.collection.find({"folder": query})
|
||||
return convert_many(songs)
|
||||
|
||||
def find_songs_by_artist(self, query: str) -> list:
|
||||
"""
|
||||
Returns a list of all the tracks exactly matching the artists in the query params.
|
||||
"""
|
||||
songs = self.collection.find({"artists": query})
|
||||
return convert_many(songs)
|
||||
|
||||
def find_songs_by_albumartist(self, query: str):
|
||||
"""
|
||||
Returns a list of all the tracks containing the albumartist in the query params.
|
||||
"""
|
||||
songs = self.collection.find(
|
||||
{"albumartist": {
|
||||
"$regex": query,
|
||||
"$options": "i"
|
||||
}})
|
||||
return convert_many(songs)
|
||||
|
||||
def get_song_by_path(self, path: str) -> dict:
|
||||
"""
|
||||
Returns a single track matching the filepath in the query params.
|
||||
"""
|
||||
song = self.collection.find_one({"filepath": path})
|
||||
return convert_one(song)
|
||||
|
||||
def remove_song_by_filepath(self, filepath: str):
|
||||
"""
|
||||
Removes a single track from the database. Returns a boolean indicating success or failure of the operation.
|
||||
"""
|
||||
try:
|
||||
self.collection.delete_one({"filepath": filepath})
|
||||
return True
|
||||
except:
|
||||
return False
|
||||
|
||||
def remove_song_by_id(self, id: str):
|
||||
"""
|
||||
Removes a single track from the database. Returns a boolean indicating success or failure of the operation.
|
||||
"""
|
||||
try:
|
||||
self.collection.delete_one({"_id": ObjectId(id)})
|
||||
return True
|
||||
except:
|
||||
return False
|
||||
Reference in New Issue
Block a user