131 lines
No EOL
4.9 KiB
Python
131 lines
No EOL
4.9 KiB
Python
from typing import Union
|
|
from sqlmodel import Session
|
|
|
|
from threading import Lock
|
|
critical_function_lock = Lock()
|
|
|
|
from src.modules import bgg_connection, db_connection
|
|
from src.classes import boardgame_classes, play_classes, people_classes
|
|
|
|
def get_db_engine():
|
|
return db_connection.get_engine()
|
|
|
|
def get_boardgame(session: Session, boardgame_id: int) -> boardgame_classes.BoardGame:
|
|
#Will check if it already exists in db, then it will get it from there
|
|
|
|
boardgame_in_db = db_connection.get_boardgame(session, boardgame_id=boardgame_id)
|
|
|
|
to_return_boardgame = None
|
|
|
|
if boardgame_in_db != None:
|
|
to_return_boardgame = boardgame_in_db
|
|
else:
|
|
to_return_boardgame = bgg_connection.get_boardgame(boardgame_id)
|
|
db_connection.add_boardgame(session, to_return_boardgame)
|
|
to_return_boardgame = db_connection.get_boardgame(session, boardgame_id)
|
|
|
|
|
|
return to_return_boardgame
|
|
|
|
|
|
def get_multiple_boardgames(session: Session, boardgame_ids: list[int]) -> list[boardgame_classes.BoardGame]:
|
|
boardgames_in_db, boardgame_ids_missing = db_connection.get_multiple_boardgames(session, boardgame_ids=boardgame_ids)
|
|
|
|
if len(boardgame_ids_missing) != 0:
|
|
missing_boardgames = bgg_connection.get_multiple_boardgames(boardgame_ids_missing)
|
|
db_connection.add_multiple_boardgames(session, missing_boardgames)
|
|
|
|
boardgames_in_db, boardgame_ids_missing = db_connection.get_multiple_boardgames(session, boardgame_ids=boardgame_ids)
|
|
|
|
return boardgames_in_db
|
|
|
|
def get_user_collection(session: Session) -> list[boardgame_classes.BoardGame]:
|
|
boardgames_from_db: list[boardgame_classes.BoardGame] = db_connection.get_all_boardgames(session)
|
|
|
|
if len(boardgames_from_db) == 0:
|
|
boardgames = bgg_connection.get_user_collection()
|
|
|
|
db_connection.add_multiple_boardgames(session, boardgames)
|
|
|
|
boardgames_from_db: list[boardgame_classes.BoardGame] = db_connection.get_all_boardgames(session)
|
|
|
|
return boardgames_from_db
|
|
|
|
def get_user_owned_collection(session: Session) -> list[boardgame_classes.BoardGame]:
|
|
|
|
owned_boardgames_from_db = db_connection.get_owned_boardgames(session)
|
|
|
|
if len(owned_boardgames_from_db) == 0:
|
|
owned_boardgames = bgg_connection.get_user_owned_collection()
|
|
db_connection.add_multiple_boardgames(session, owned_boardgames)
|
|
|
|
owned_boardgames_from_db: list[boardgame_classes.BoardGame] = db_connection.get_owned_boardgames(session)
|
|
|
|
return owned_boardgames_from_db
|
|
|
|
|
|
def get_user_wishlist_collection(session: Session, wishlist_priority: int = 0) -> list[boardgame_classes.BoardGame]:
|
|
|
|
wishlisted_boardgames_from_db = db_connection.get_wishlist_boardgames(session)
|
|
|
|
if len(wishlisted_boardgames_from_db) == 0:
|
|
wishlisted_boardgames = bgg_connection.get_user_wishlist_collection()
|
|
db_connection.add_multiple_boardgames(session, wishlisted_boardgames)
|
|
|
|
wishlisted_boardgames_from_db = db_connection.get_wishlist_boardgames(session)
|
|
|
|
to_return_boardgames = wishlisted_boardgames_from_db
|
|
|
|
if wishlist_priority != 0:
|
|
to_return_boardgames = list(filter(lambda game: game.wishlist_info.wishlist_priority == wishlist_priority, to_return_boardgames))
|
|
|
|
return to_return_boardgames
|
|
|
|
|
|
def get_plays(session: Session) -> list[play_classes.Play]:
|
|
|
|
plays_from_db = db_connection.get_plays(session)
|
|
|
|
if len(plays_from_db) == 0:
|
|
all_plays = bgg_connection.get_plays()
|
|
|
|
db_connection.add_multiple_plays(session, all_plays)
|
|
|
|
plays_from_db = db_connection.get_plays(session)
|
|
|
|
#Making sure all played board games are in table 'boardgames'
|
|
#list + set to remove duplicates
|
|
played_boardgame_ids = list(set([play.boardgame_id for play in plays_from_db]))
|
|
played_expansion_ids = list(set([play.expansion_id for play in plays_from_db]))
|
|
|
|
#Remove None's (played board games don't have expansion id and vice versa)
|
|
played_boardgame_ids = list(filter(lambda x: x != None, played_boardgame_ids))
|
|
played_expansion_ids = list(filter(lambda x: x != None, played_expansion_ids))
|
|
|
|
assert len(list(filter(lambda x: x == None, played_boardgame_ids))) == 0, plays_from_db
|
|
|
|
#get_multiple_boardgames(session, played_boardgame_ids + played_expansion_ids)
|
|
|
|
return plays_from_db
|
|
|
|
def get_players_from_play(session: Session, play_id: int) -> list[play_classes.PlayPlayer]:
|
|
players_from_db = db_connection.get_players_from_play(session, play_id)
|
|
|
|
if len(players_from_db) == 0:
|
|
all_plays = bgg_connection.get_plays()
|
|
|
|
db_connection.add_multiple_plays(session, all_plays)
|
|
|
|
players_from_db = db_connection.get_players_from_play(session, play_id)
|
|
|
|
return players_from_db
|
|
|
|
def get_designers(session: Session) -> list[people_classes.Designer]:
|
|
designers_from_db = db_connection.get_all_designers(session)
|
|
return designers_from_db
|
|
|
|
def delete_database():
|
|
db_connection.delete_database()
|
|
|
|
def create_db_and_tables():
|
|
db_connection.create_db_and_tables() |