Yarninator/bgg_api
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Made it so that BGG request can no longer spam

Browse source
This commit is contained in:
Yarne Coppens 2024-09-04 22:46:18 +02:00
parent 0529819da8
commit 18fb6a38b3
2 changed files with 81 additions and 72 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
src/modules/bgg_connection.py
View file

@ -175,7 +175,8 @@ def convert_collection_xml_to_wishlist_boardgame(boardgame_extra_info: boardgame
wishlist_priority = collection_boardgame_xml.find('status').get('wishlistpriority') wishlist_priority = collection_boardgame_xml.find('status').get('wishlistpriority')
wishlist_boardgame_dict = { wishlist_boardgame_dict = {
"wishlist_priority" : wishlist_priority, "boardgame_id" : boardgame_extra_info.id,
"wishlist_priority" : wishlist_priority
} }
wishlist_info = boardgame_classes.WishlistInfo.model_validate(wishlist_boardgame_dict) wishlist_info = boardgame_classes.WishlistInfo.model_validate(wishlist_boardgame_dict)

146
src/modules/data_connection.py
View file

@ -13,122 +13,130 @@ def get_db_engine():
def get_boardgame(session: Session, boardgame_id: int) -> boardgame_classes.BoardGame: 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 #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 with critical_function_lock:
boardgame_in_db = db_connection.get_boardgame(session, boardgame_id=boardgame_id)
if boardgame_in_db != None: to_return_boardgame = None
to_return_boardgame = boardgame_in_db
else: if boardgame_in_db != None:
to_return_boardgame = bgg_connection.get_boardgame(boardgame_id) to_return_boardgame = boardgame_in_db
db_connection.add_boardgame(session, to_return_boardgame) else:
to_return_boardgame = db_connection.get_boardgame(session, boardgame_id) 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 return to_return_boardgame
def get_multiple_boardgames(session: Session, boardgame_ids: list[int]) -> list[boardgame_classes.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.upsert_multiple_boardgames(session, missing_boardgames)
with critical_function_lock:
boardgames_in_db, boardgame_ids_missing = db_connection.get_multiple_boardgames(session, boardgame_ids=boardgame_ids) boardgames_in_db, boardgame_ids_missing = db_connection.get_multiple_boardgames(session, boardgame_ids=boardgame_ids)
return boardgames_in_db if len(boardgame_ids_missing) != 0:
missing_boardgames = bgg_connection.get_multiple_boardgames(boardgame_ids_missing)
db_connection.upsert_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]: 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.upsert_multiple_boardgames(session, boardgames)
with critical_function_lock:
boardgames_from_db: list[boardgame_classes.BoardGame] = db_connection.get_all_boardgames(session) boardgames_from_db: list[boardgame_classes.BoardGame] = db_connection.get_all_boardgames(session)
return boardgames_from_db if len(boardgames_from_db) == 0:
boardgames = bgg_connection.get_user_collection()
db_connection.upsert_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]: def get_user_owned_collection(session: Session) -> list[boardgame_classes.BoardGame]:
owned_boardgames_from_db = db_connection.get_owned_boardgames(session) with critical_function_lock:
owned_boardgames_from_db = db_connection.get_owned_boardgames(session)
if len(owned_boardgames_from_db) == 0: if len(owned_boardgames_from_db) == 0:
owned_boardgames = bgg_connection.get_user_owned_collection() owned_boardgames = bgg_connection.get_user_owned_collection()
db_connection.upsert_multiple_boardgames(session, owned_boardgames) db_connection.upsert_multiple_boardgames(session, owned_boardgames)
owned_boardgames_from_db: list[boardgame_classes.BoardGame] = db_connection.get_owned_boardgames(session) owned_boardgames_from_db: list[boardgame_classes.BoardGame] = db_connection.get_owned_boardgames(session)
return owned_boardgames_from_db return owned_boardgames_from_db
def get_user_wishlist_collection(session: Session, wishlist_priority: int = 0) -> list[boardgame_classes.BoardGame]: 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) with critical_function_lock:
if len(wishlisted_boardgames_from_db) == 0:
wishlisted_boardgames = bgg_connection.get_user_wishlist_collection()
db_connection.upsert_multiple_boardgames(session, wishlisted_boardgames)
wishlisted_boardgames_from_db = db_connection.get_wishlist_boardgames(session) wishlisted_boardgames_from_db = db_connection.get_wishlist_boardgames(session)
print(wishlisted_boardgames_from_db) if len(wishlisted_boardgames_from_db) == 0:
if wishlist_priority != 0: wishlisted_boardgames = bgg_connection.get_user_wishlist_collection()
wishlisted_boardgames_from_db = list(filter(lambda game: game.wishlist_info.wishlist_priority == wishlist_priority, wishlisted_boardgames_from_db)) db_connection.upsert_multiple_boardgames(session, wishlisted_boardgames)
return wishlisted_boardgames_from_db wishlisted_boardgames_from_db = db_connection.get_wishlist_boardgames(session)
if wishlist_priority != 0:
wishlisted_boardgames_from_db = list(filter(lambda game: game.wishlist_info.wishlist_priority == wishlist_priority, wishlisted_boardgames_from_db))
return wishlisted_boardgames_from_db
def get_plays(session: Session) -> list[play_classes.Play]: def get_plays(session: Session) -> list[play_classes.Play]:
plays_from_db = db_connection.get_plays(session) with critical_function_lock:
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) plays_from_db = db_connection.get_plays(session)
#Making sure all played board games are in table 'boardgames' if len(plays_from_db) == 0:
#list + set to remove duplicates all_plays = bgg_connection.get_plays()
played_boardgame_ids = list(set([play.boardgame_id for play in plays_from_db]))
#Remove None's (played board games don't have expansion id and vice versa) db_connection.add_multiple_plays(session, all_plays)
played_boardgame_ids = list(filter(lambda x: x != None, played_boardgame_ids))
assert len(list(filter(lambda x: x == None, played_boardgame_ids))) == 0, plays_from_db plays_from_db = db_connection.get_plays(session)
#Make sure to add all board games that are played #Making sure all played board games are in table 'boardgames'
get_multiple_boardgames(session, played_boardgame_ids) #list + set to remove duplicates
played_boardgame_ids = list(set([play.boardgame_id for play in plays_from_db]))
return 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))
assert len(list(filter(lambda x: x == None, played_boardgame_ids))) == 0, plays_from_db
return plays_from_db
def get_players_from_play(session: Session, play_id: int) -> list[play_classes.PlayPlayer]: 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)
with critical_function_lock:
players_from_db = db_connection.get_players_from_play(session, play_id) players_from_db = db_connection.get_players_from_play(session, play_id)
return players_from_db 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_all_players(session: Session) -> list[player_classes.Player]: def get_all_players(session: Session) -> list[player_classes.Player]:
players_from_db = db_connection.get_all_players(session) with critical_function_lock:
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_all_players(session) players_from_db = db_connection.get_all_players(session)
return players_from_db 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_all_players(session)
return players_from_db
def get_player(player_name: str, session: Session) -> player_classes.Player: def get_player(player_name: str, session: Session) -> player_classes.Player:
player_from_db = db_connection.get_player(player_name.title(), session) player_from_db = db_connection.get_player(player_name.title(), session)