bgg_api/src/modules/data_connection.py

from typing import Union

from threading import Lock
critical_function_lock = Lock()

from src.modules import bgg_connection, db_connection
from src.classes import boardgame_classes, play_classes

def get_boardgame(boardgame_id: int) -> Union[boardgame_classes.BoardGame, boardgame_classes.BoardGameExpansion]:
    #Will check if it already exists in db, then it will get it from there
    
    boardgame_in_db = db_connection.get_boardgame(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(to_return_boardgame)
        to_return_boardgame = db_connection.get_boardgame(boardgame_id)

    return to_return_boardgame


def get_multiple_boardgames(boardgame_ids: list[int]) -> list[Union[boardgame_classes.BoardGame, boardgame_classes.BoardGameExpansion]]:
    boardgames_in_db, boardgame_ids_missing = db_connection.get_multiple_boardgames(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(missing_boardgames)

        boardgames_in_db, boardgame_ids_missing = db_connection.get_multiple_boardgames(boardgame_ids=boardgame_ids)

    return boardgames_in_db

def get_user_collection() -> list[Union[boardgame_classes.BoardGame, boardgame_classes.OwnedBoardGame]]:
    boardgames_from_db: list[boardgame_classes.BoardGame] = db_connection.get_all_boardgames(boardgame_classes.BoardGame)
    boardgame_expansions_from_db: list[boardgame_classes.BoardGameExpansion] = db_connection.get_all_boardgames(boardgame_classes.BoardGameExpansion)

    if len(boardgames_from_db) == 0 and len(boardgame_expansions_from_db) == 0:
        boardgames = bgg_connection.get_user_collection()
        db_connection.add_multiple_boardgames(boardgames)

        boardgames_from_db: list[boardgame_classes.BoardGame] = db_connection.get_all_boardgames(boardgame_classes.BoardGame)
        boardgame_expansions_from_db: list[boardgame_classes.BoardGameExpansion] = db_connection.get_all_boardgames(boardgame_classes.BoardGameExpansion)

    return boardgames_from_db + boardgame_expansions_from_db

def get_user_owned_collection() -> list[Union[boardgame_classes.OwnedBoardGame, boardgame_classes.OwnedBoardGameExpansion]]:
    
    owned_boardgames_from_db: list[boardgame_classes.OwnedBoardGame] = db_connection.get_all_boardgames(boardgame_classes.OwnedBoardGame)
    owned_boardgame_expanions_from_db: list[boardgame_classes.OwnedBoardGameExpansion] = db_connection.get_all_boardgames(boardgame_classes.OwnedBoardGameExpansion)

    if len(owned_boardgames_from_db) == 0 and len(owned_boardgame_expanions_from_db) == 0:
        owned_boardgames = bgg_connection.get_user_owned_collection()
        db_connection.add_multiple_boardgames(owned_boardgames)

        owned_boardgames_from_db: list[boardgame_classes.OwnedBoardGame] = db_connection.get_all_boardgames(boardgame_classes.OwnedBoardGame)
        owned_boardgame_expanions_from_db: list[boardgame_classes.OwnedBoardGameExpansion] = db_connection.get_all_boardgames(boardgame_classes.OwnedBoardGameExpansion)


    return owned_boardgames_from_db + owned_boardgame_expanions_from_db
        


def get_user_wishlist_collection(wishlist_priority: int = 0) -> Union[list[boardgame_classes.WishlistBoardGame], list[boardgame_classes.WishlistBoardGameExpansion]]:

    wishlisted_boardgames_from_db = db_connection.get_all_boardgames(boardgame_classes.WishlistBoardGame)
    wishlisted_boardgame_expansions_from_db = db_connection.get_all_boardgames(boardgame_classes.WishlistBoardGameExpansion)

    if len(wishlisted_boardgames_from_db) == 0 and len(wishlisted_boardgame_expansions_from_db) == 0:
        wishlisted_boardgames = bgg_connection.get_user_wishlist_collection()
        db_connection.add_multiple_boardgames(wishlisted_boardgames)

        wishlisted_boardgames_from_db = db_connection.get_all_boardgames(boardgame_classes.WishlistBoardGame)
        wishlisted_boardgame_expansions_from_db = db_connection.get_all_boardgames(boardgame_classes.WishlistBoardGameExpansion)

    to_return_boardgames = wishlisted_boardgames_from_db + wishlisted_boardgame_expansions_from_db

    if wishlist_priority != 0:
        to_return_boardgames = filter(lambda game: game.wishlist_priority == wishlist_priority, to_return_boardgames)

    return to_return_boardgames


def get_plays() -> list[play_classes.Play]:

    plays_from_db = db_connection.get_plays()

    if len(plays_from_db) == 0:
        all_plays = bgg_connection.get_plays()

        db_connection.add_multiple_plays(all_plays)

        plays_from_db = db_connection.get_plays()

    #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(played_boardgame_ids + played_expansion_ids)

    return plays_from_db

def get_players_from_play(play_id: int) -> list[play_classes.PlayPlayer]:
    players_from_db = db_connection.get_players_from_play(play_id)

    if len(players_from_db) == 0:
        all_plays = bgg_connection.get_plays()

        db_connection.add_multiple_plays(all_plays)

        players_from_db = db_connection.get_players_from_play(play_id)

    return players_from_db

def delete_database():
    db_connection.delete_database()

def create_db_and_tables():
    db_connection.create_db_and_tables()