Wrote wishlist test

.gitignore

@@ -160,3 +160,4 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
 
+secrets/auth.yaml

README.md

@@ -1,4 +1,2 @@
 # bgg_api
 
-An API implementation that will be used by my own board game website.
-The plan is for this API to be used by a single person who wants their board game collection to be cached so that retrieval is much faster.

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+pythonpath = .

secrets/auth_example.yaml

@@ -0,0 +1,3 @@
+#Rename this file to auth.yaml and provide correct values below to authenticate to bgg
+username: username_example
+password: password_example

src/classes/boardgame_classes.py

@@ -0,0 +1,46 @@
+from pydantic import BaseModel, HttpUrl
+from datetime import date
+from enum import Enum
+
+class BoardgameType(Enum):
+    BOARDGAME = 'boardgame'
+    BOARDGAMEEXPANSION = 'boardgameexpansion'
+    OWNEDBOARDGAME = 'ownedboardgame'
+    OWNEDBOARDGAMEEXPANSION = 'ownedboardgameexpansion'
+    WISHLISTBOARDGAME = 'wishlistboardgame'
+    WISHLISTBOARDGAMEEXPANSION = 'wishlistboardgameexpansion'
+
+
+class BoardGame(BaseModel):
+    id: int
+    name: str
+    description: str
+    image_url : HttpUrl
+    thumbnail_url : HttpUrl
+    year_published: int
+    min_players: int
+    max_players: int
+    min_playing_time: int
+    max_playing_time: int
+    min_age: int
+    all_expansion_ids: list[int]
+    type: BoardgameType = BoardgameType.BOARDGAME
+
+class BoardGameExpansion(BoardGame):
+    type: BoardgameType = BoardgameType.BOARDGAMEEXPANSION
+
+class OwnedBoardGame(BoardGame):
+    price_paid: float
+    acquisition_date: date
+    acquired_from: str
+    type: BoardgameType = BoardgameType.OWNEDBOARDGAME
+
+class OwnedBoardGameExpansion(OwnedBoardGame):
+    type: BoardgameType = BoardgameType.OWNEDBOARDGAMEEXPANSION
+
+class WishlistBoardGame(BoardGame):
+    wishlist_priority: int
+    type: BoardgameType = BoardgameType.WISHLISTBOARDGAME
+
+class WishlistBoardGameExpansion(WishlistBoardGame):
+    type: BoardgameType = BoardgameType.WISHLISTBOARDGAMEEXPANSION

src/classes/play_classes.py

@@ -0,0 +1,18 @@
+from pydantic import BaseModel
+from typing import Union
+from datetime import date
+
+class PlayPlayer(BaseModel):
+    name: str
+    username: str
+    score: Union[float, None]
+    first_play : bool
+    has_won : bool
+
+class Play(BaseModel):
+    boardgame_id: int
+    players: list[PlayPlayer]
+    play_date: date
+    duration: int #In minutes
+    ignore_for_stats : bool
+    location: str

src/config/definitions.py

@@ -0,0 +1,7 @@
+import os
+
+ROOT_PATH = project_root = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
+SECRETS_FILE_PATH = ROOT_PATH + '/secrets/auth.yaml'
+
+BGG_MAX_THING_BOARDGAMES = 20
+BGG_PLAY_PAGE_SIZE = 100

src/main.py

@@ -0,0 +1,35 @@
+from typing import Union
+from fastapi import FastAPI
+
+from src.classes import boardgame_classes, play_classes
+from src.modules import bgg_connection
+
+app = FastAPI()
+
+
+@app.get("/")
+def read_root():
+    return {"Hello": "World"}
+
+@app.get("/boardgame/{boardgame_id}", response_model=boardgame_classes.BoardGame)
+def get_boardgame_by_id(boardgame_id: int):
+    requested_boardgame: boardgame_classes.BoardGame = bgg_connection.get_boardgame(boardgame_id)
+    return requested_boardgame
+
+@app.get("/owned", response_model=list[boardgame_classes.OwnedBoardGame])
+def get_owned_collection():
+    requested_collection: list[boardgame_classes.OwnedBoardGame] = bgg_connection.get_user_owned_collection()
+    return requested_collection
+
+
+@app.get("/wishlist", response_model=list[boardgame_classes.WishlistBoardGame])
+def get_wishlist_collection():
+    requested_collection: list[boardgame_classes.WishlistBoardGame] = bgg_connection.get_user_wishlist_collection()
+    return requested_collection
+
+
+
+@app.get("/plays", response_model=list[play_classes.Play])
+def get_plays():
+    requested_plays: list[play_classes.Play] = bgg_connection.get_plays()
+    return requested_plays

src/modules/auth_manager.py

@@ -0,0 +1,25 @@
+#Can only be imported on bgg_connection.py
+
+import yaml
+
+from src.config import definitions
+
+username: str = None
+password: str = None
+
+def load_username_password_from_secrets():
+    global username
+    global password
+    
+    with open(definitions.SECRETS_FILE_PATH, 'r') as auth_file:
+        auth_object = yaml.safe_load(auth_file)
+
+    username = auth_object['username']
+    password = auth_object['password']
+
+
+def get_username_password():
+    return username, password
+
+
+load_username_password_from_secrets()

src/modules/bgg_connection.py

@@ -0,0 +1,330 @@
+import requests
+import xml.etree.ElementTree as ET
+from pydantic import HttpUrl
+import requests
+from datetime import datetime
+import time
+import math
+
+from src.classes import boardgame_classes, play_classes
+from src.modules import auth_manager
+from src.config import definitions
+
+authenticated_session: requests.Session = requests.Session()
+
+def url_to_xml_object(url: HttpUrl) -> ET.Element:
+    r = authenticated_session.get(url)
+
+    while r.status_code == 202:
+        print('BGG is processing...')
+        time.sleep(10)
+        r = authenticated_session.get(url)
+
+    assert r.status_code == 200, "Got {} status code".format(r.status_code)
+    root = ET.fromstring(r.content)
+    return root
+
+def get_boardgame(boardgame_id: int) -> boardgame_classes.BoardGame:
+    url : str = "https://boardgamegeek.com/xmlapi2/thing?id={}&stats=true".format(boardgame_id)
+    boardgame_xml_object : ET.Element = url_to_xml_object(url)
+
+    requested_boardgame = convert_xml_to_boardgame(boardgame_xml_object.find('item'))
+
+    return requested_boardgame
+
+def get_multiple_boardgames(boardgame_ids: list[int]) -> list[boardgame_classes.BoardGame]:
+
+    def divide_list_in_chunks(list_to_divide: list[int], chunk_size: int = definitions.BGG_MAX_THING_BOARDGAMES): 
+        for i in range(0, len(list_to_divide), chunk_size):  
+            yield list_to_divide[i:i + chunk_size] 
+
+    
+    boardgame_list_to_return: list[boardgame_classes.BoardGame] = []
+
+    #Boardgamegeek only allows chunks of 20 boardgames at a time
+    boardgame_ids_divided = list(divide_list_in_chunks(boardgame_ids))
+
+    for boardgame_id_list_size_20 in boardgame_ids_divided:
+        boardgame_id_list_commas: str = ','.join(boardgame_id_list_size_20)
+        url : str = "https://boardgamegeek.com/xmlapi2/thing?id={}&stats=true".format(boardgame_id_list_commas)
+        boardgames_xml_object : ET.Element = url_to_xml_object(url)
+
+        for boardgame_xml_object in boardgames_xml_object:
+            requested_boardgame = convert_xml_to_boardgame(boardgame_xml_object)
+            boardgame_list_to_return.append(requested_boardgame)
+
+    return boardgame_list_to_return
+
+
+#Requires single boardgame XML 'item' from bgg api on /thing
+def convert_xml_to_boardgame(boardgame_xml: ET.Element) -> boardgame_classes.BoardGame:
+
+    boardgame_type = boardgame_xml.get('type')
+
+    expansion_ids: list[int] = []
+
+    all_links = boardgame_xml.findall('link')
+
+    for link in all_links:
+        if link.get('type') == 'boardgameexpansion':
+            expansion_ids.append(int(link.get('id')))
+
+    boardgame_dict = {
+        "id" : int(boardgame_xml.get('id')),
+        "name" : boardgame_xml.find('name').get('value'),
+        "description" : boardgame_xml.find('description').text,
+        "image_url" : boardgame_xml.find('image').text,
+        "thumbnail_url" : boardgame_xml.find('thumbnail').text,
+        "year_published" : int(boardgame_xml.find('yearpublished').get('value')),
+        "min_players" : int(boardgame_xml.find('minplayers').get('value')),
+        "max_players" : int(boardgame_xml.find('maxplayers').get('value')),
+        "min_playing_time" : int(boardgame_xml.find('minplaytime').get('value')),
+        "max_playing_time" : int(boardgame_xml.find('maxplaytime').get('value')),
+        "min_age" : int(boardgame_xml.find('minage').get('value')),
+        "all_expansion_ids" : expansion_ids
+    }
+
+    match boardgame_type:
+        case "boardgame":
+            boardgame = boardgame_classes.BoardGame(**boardgame_dict)
+        case "boardgameexpansion":
+            boardgame = boardgame_classes.BoardGameExpansion(**boardgame_dict)
+
+    
+
+    return boardgame
+
+def convert_collection_xml_to_owned_boardgame(boardgame_extra_info: boardgame_classes.BoardGame, collection_boardgame_xml: ET.Element) -> boardgame_classes.OwnedBoardGame:
+
+    boardgame_type = collection_boardgame_xml.get('subtype')
+
+    price_paid = collection_boardgame_xml.find('privateinfo').get('pricepaid')
+    if price_paid == '':
+        price_paid = 0.0
+    else:
+        price_paid = float(price_paid)
+
+
+    date_string = collection_boardgame_xml.find('privateinfo').get('acquisitiondate')
+    if date_string == '':
+        date_string = '1970-01-01'
+
+    
+    acquisition_date = datetime.strptime(date_string, '%Y-%m-%d').date()
+        
+    acquired_from = collection_boardgame_xml.find('privateinfo').get('acquiredfrom')
+
+    owned_boardgame_dict = {
+        "price_paid" : price_paid,
+        "acquisition_date" : acquisition_date,
+        "acquired_from" : acquired_from
+    }
+
+    boardgame_dict = {
+        "id" : boardgame_extra_info.id,
+        "name" : boardgame_extra_info.name,
+        "description" : boardgame_extra_info.description,
+        "image_url" : boardgame_extra_info.image_url,
+        "thumbnail_url" : boardgame_extra_info.thumbnail_url,
+        "year_published" : boardgame_extra_info.year_published,
+        "min_players" : boardgame_extra_info.min_players,
+        "max_players" : boardgame_extra_info.max_players,
+        "min_playing_time" : boardgame_extra_info.min_playing_time,
+        "max_playing_time" : boardgame_extra_info.max_playing_time,
+        "min_age" : boardgame_extra_info.min_age,
+        "all_expansion_ids" : boardgame_extra_info.all_expansion_ids,
+        **owned_boardgame_dict
+    }
+
+    match boardgame_type:
+        case "boardgame":
+            boardgame = boardgame_classes.OwnedBoardGame(**boardgame_dict)
+        case "boardgameexpansion":
+            boardgame = boardgame_classes.OwnedBoardGameExpansion(**boardgame_dict)
+
+    
+
+    return boardgame
+
+def convert_collection_xml_to_wishlist_boardgame(boardgame_extra_info: boardgame_classes.BoardGame, collection_boardgame_xml: ET.Element) -> boardgame_classes.WishlistBoardGame:
+
+    boardgame_type = collection_boardgame_xml.get('subtype')
+
+    wishlist_priority = collection_boardgame_xml.find('status').get('wishlistpriority')
+
+    wishlist_boardgame_dict = {
+        "wishlist_priority" : wishlist_priority,
+    }
+
+    boardgame_dict = {
+        "id" : boardgame_extra_info.id,
+        "name" : boardgame_extra_info.name,
+        "description" : boardgame_extra_info.description,
+        "image_url" : boardgame_extra_info.image_url,
+        "thumbnail_url" : boardgame_extra_info.thumbnail_url,
+        "year_published" : boardgame_extra_info.year_published,
+        "min_players" : boardgame_extra_info.min_players,
+        "max_players" : boardgame_extra_info.max_players,
+        "min_playing_time" : boardgame_extra_info.min_playing_time,
+        "max_playing_time" : boardgame_extra_info.max_playing_time,
+        "min_age" : boardgame_extra_info.min_age,
+        "all_expansion_ids" : boardgame_extra_info.all_expansion_ids,
+        **wishlist_boardgame_dict
+    }
+
+    match boardgame_type:
+        case "boardgame":
+            boardgame = boardgame_classes.WishlistBoardGame(**boardgame_dict)
+        case "boardgameexpansion":
+            boardgame = boardgame_classes.WishlistBoardGameExpansion(**boardgame_dict)
+
+    
+
+    return boardgame
+
+def convert_playplayer_xml_to_playplayer(playplayer_xml: ET.Element) -> play_classes.PlayPlayer:
+
+    score = playplayer_xml.get('score')
+    if score == '':
+        score = None
+    else:
+        score = float(score)
+
+    playplayer_dict = {
+        "name" : playplayer_xml.get('name'),
+        "username" : playplayer_xml.get('username'),
+        "score" : score,
+        "first_play" : bool(int(playplayer_xml.get('new'))),
+        "has_won" : bool(int(playplayer_xml.get('win')))
+    }
+
+    playplayer = play_classes.PlayPlayer(**playplayer_dict)
+
+    return playplayer
+
+def convert_play_xml_to_play(play_xml: ET.Element) -> play_classes.Play:
+
+    date_string = play_xml.get('date')
+    play_date = datetime.strptime(date_string, '%Y-%m-%d').date()
+    
+    playplayer_list: list[play_classes.PlayPlayer] = []
+
+    for play_player_xml in play_xml.find('players'):
+        playplayer_list.append(convert_playplayer_xml_to_playplayer(play_player_xml))
+
+    play_dict = {
+        "boardgame_id" : int(play_xml.find('item').get('objectid')),
+        "players" : playplayer_list,
+        "play_date" : play_date,
+        "duration" : int(play_xml.get('length')),
+        "ignore_for_stats" : bool(play_xml.get('nowinstats')),
+        "location" : play_xml.get('location')
+    }
+
+    play = play_classes.Play(**play_dict)
+
+    return play
+
+    
+
+#Creates list of board games from a collection '/collection' URL
+def get_boardgames_from_collection_url(collection_url: str, boardgame_type: boardgame_classes.BoardgameType) -> list[boardgame_classes.BoardGame]:
+    collection_xml = url_to_xml_object(collection_url)
+
+    collection_list: list[boardgame_classes.BoardGame] = []
+
+    collection_id_list: list[int] = []
+
+    #Get all board game ID's from the collection
+    for boardgame_item in collection_xml:
+        collection_id_list.append(boardgame_item.get('objectid'))
+
+    #Request extra info about the ID's
+    boardgame_extras = get_multiple_boardgames(collection_id_list)
+
+    assert len(boardgame_extras) == len(collection_xml)
+
+    current_index = 0
+
+    for boardgame_item in collection_xml:
+        boardgame_extra = boardgame_extras[current_index]
+        match boardgame_type:
+            case boardgame_classes.BoardgameType.OWNEDBOARDGAME:
+                boardgame = convert_collection_xml_to_owned_boardgame(boardgame_extra, boardgame_item)
+            case boardgame_classes.BoardgameType.OWNEDBOARDGAMEEXPANSION:
+                boardgame = convert_collection_xml_to_owned_boardgame(boardgame_extra, boardgame_item)
+            case boardgame_classes.BoardgameType.WISHLISTBOARDGAME:
+                boardgame = convert_collection_xml_to_wishlist_boardgame(boardgame_extra, boardgame_item)
+            case boardgame_classes.BoardgameType.WISHLISTBOARDGAMEEXPANSION:
+                boardgame = convert_collection_xml_to_wishlist_boardgame(boardgame_extra, boardgame_item)
+        collection_list.append(boardgame)
+        current_index += 1
+
+    return collection_list
+
+def get_user_owned_collection() -> list[boardgame_classes.BoardGame]:
+    url_no_expansions = 'https://boardgamegeek.com/xmlapi2/collection?username={}&own=1&stats=1&excludesubtype=boardgameexpansion&showprivate=1&version=1'.format(auth_manager.username)
+    url_only_expansions = 'https://boardgamegeek.com/xmlapi2/collection?username={}&own=1&stats=1&subtype=boardgameexpansion&showprivate=1&version=1'.format(auth_manager.username)
+
+    owned_boardgames = get_boardgames_from_collection_url(url_no_expansions, boardgame_classes.BoardgameType.OWNEDBOARDGAME)
+    owned_boardgame_expansions = get_boardgames_from_collection_url(url_only_expansions, boardgame_classes.BoardgameType.OWNEDBOARDGAMEEXPANSION)
+    
+    owned_boardgames += owned_boardgame_expansions
+
+    return owned_boardgames
+    
+def get_user_wishlist_collection() -> list[boardgame_classes.BoardGame]:
+    url_no_expanions = 'https://boardgamegeek.com/xmlapi2/collection?username={}&wishlist=1&stats=1&excludesubtype=boardgameexpansion&showprivate=1&version=1'.format(auth_manager.username)
+    url_only_expansions = 'https://boardgamegeek.com/xmlapi2/collection?username={}&wishlist=1&stats=1&subtype=boardgameexpansion&showprivate=1&version=1'.format(auth_manager.username)
+
+    wishlisted_boardgames = get_boardgames_from_collection_url(url_no_expanions, boardgame_classes.BoardgameType.WISHLISTBOARDGAME)
+    wishlisted_boardgame_expansions = get_boardgames_from_collection_url(url_only_expansions, boardgame_classes.BoardgameType.WISHLISTBOARDGAMEEXPANSION)
+
+    wishlisted_boardgames += wishlisted_boardgame_expansions
+
+    return wishlisted_boardgames
+
+
+
+def get_plays() -> list[play_classes.Play]:
+    first_page_url = 'https://boardgamegeek.com/xmlapi2/plays?username={}'.format(auth_manager.username)
+    plays_first_page_xml_object = url_to_xml_object(first_page_url)
+
+    amount_of_plays_total = float(plays_first_page_xml_object.get('total'))
+
+    amount_of_pages_needed = math.ceil(amount_of_plays_total/float(definitions.BGG_PLAY_PAGE_SIZE))
+
+    all_plays : list[play_classes.Play] = []
+
+    for page in range(amount_of_pages_needed):
+        url = 'https://boardgamegeek.com/xmlapi2/plays?username={}&page={}'.format(auth_manager.username, page)
+        plays_page_xml_object = url_to_xml_object(url)
+        for play_xml in plays_page_xml_object:
+            new_play = convert_play_xml_to_play(play_xml)
+            all_plays.append(new_play)
+
+    return all_plays
+
+
+
+
+def load_authenticated_bgg_session(username: str, password: str) -> requests.Session:
+    global authenticated_session
+
+    login_url = "https://boardgamegeek.com/login/api/v1"
+
+    post_data = {
+        "credentials":{
+            "username": username,
+            "password": password
+        }
+    }
+
+
+    assert len(authenticated_session.cookies) == 0, 'Session already exists'
+
+    login_response = authenticated_session.post(login_url, json=post_data)
+
+    assert login_response.status_code == 204, "Login failed!"
+   
+load_authenticated_bgg_session(auth_manager.username, auth_manager.password)

tests/test_main.py

@@ -0,0 +1,77 @@
+import validators
+from fastapi.testclient import TestClient
+from datetime import date
+
+from src.main import app
+
+from src.classes import boardgame_classes, play_classes
+
+client = TestClient(app)
+
+def default_boardgame_test(to_test_boardgame: boardgame_classes.BoardGame):
+    assert type(to_test_boardgame.id) == int
+    assert type(to_test_boardgame.name) == str
+    assert type(to_test_boardgame.description) == str
+    assert validators.url(str(to_test_boardgame.image_url))
+    assert validators.url(str(to_test_boardgame.thumbnail_url))
+    assert type(to_test_boardgame.year_published) == int
+    assert type(to_test_boardgame.min_players) == int
+    assert type(to_test_boardgame.max_players) == int
+    assert type(to_test_boardgame.min_playing_time) == int
+    assert type(to_test_boardgame.max_playing_time) == int
+    assert type(to_test_boardgame.min_age) == int
+    assert type(to_test_boardgame.all_expansion_ids) == list
+    assert type(to_test_boardgame.all_expansion_ids[0]) == int
+    assert type(to_test_boardgame.type) == boardgame_classes.BoardgameType
+
+
+
+def test_read_main():
+    response = client.get("/")
+    assert response.status_code == 200
+    assert response.json() == {"Hello": "World"}
+
+def test_retrieve_boardgame():
+    response = client.get("/boardgame/373167")
+    assert response.status_code == 200
+
+    returned_boardgame = boardgame_classes.BoardGame(**response.json())
+
+    default_boardgame_test(returned_boardgame)
+
+def test_retrieve_owned():
+    response = client.get("/owned")
+    assert response.status_code == 200
+
+    returned_boardgame = boardgame_classes.OwnedBoardGame(**response.json()[0])
+
+    default_boardgame_test(returned_boardgame)
+    assert type(returned_boardgame.price_paid) == float
+    assert type(returned_boardgame.acquisition_date) == date
+    assert type(returned_boardgame.acquired_from) == str
+
+
+def test_retrieve_wishlist():
+    response = client.get("/wishlist")
+    assert response.status_code == 200
+
+    returned_boardgame = boardgame_classes.WishlistBoardGame(**response.json()[0])
+
+    default_boardgame_test(returned_boardgame)
+    assert type(returned_boardgame.wishlist_priority) == int
+    assert returned_boardgame.wishlist_priority > 0
+
+
+def test_retrieve_plays():
+    response = client.get("/plays")
+    assert response.status_code == 200
+
+    returned_play = play_classes.Play(**response.json()[0])
+
+    assert type(returned_play.boardgame_id) == int
+    assert type(returned_play.players) == list
+    assert type(returned_play.players[0]) == play_classes.PlayPlayer
+    assert type(returned_play.play_date) == date
+    assert type(returned_play.duration) == int
+    assert type(returned_play.ignore_for_stats) == bool
+    assert type(returned_play.location) == str