Compare commits
3 commits
new_boardg
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
30ea0d8fec | ||
|
|
7279d9f93c | ||
|
|
7d1b3458e7 |
7 changed files with 2 additions and 309 deletions
1
.gitignore
vendored
1
.gitignore
vendored
|
|
@ -160,4 +160,3 @@ cython_debug/
|
||||||
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
|
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
|
||||||
#.idea/
|
#.idea/
|
||||||
|
|
||||||
secrets/auth.yaml
|
|
||||||
|
|
@ -1,2 +1,4 @@
|
||||||
# bgg_api
|
# 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.
|
||||||
|
|
|
||||||
|
|
@ -1,25 +0,0 @@
|
||||||
#Can only be imported on main.py
|
|
||||||
|
|
||||||
import yaml
|
|
||||||
|
|
||||||
username: str = None
|
|
||||||
password: str = None
|
|
||||||
|
|
||||||
auth_secret_file_location = './secrets/auth.yaml'
|
|
||||||
|
|
||||||
def load_username_password_from_secrets():
|
|
||||||
global username
|
|
||||||
global password
|
|
||||||
|
|
||||||
with open(auth_secret_file_location, '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()
|
|
||||||
|
|
@ -1,205 +0,0 @@
|
||||||
import requests
|
|
||||||
import xml.etree.ElementTree as ET
|
|
||||||
from pydantic import HttpUrl
|
|
||||||
import requests
|
|
||||||
from datetime import datetime
|
|
||||||
|
|
||||||
from classes.boardgame import BoardGame, BoardGameExpansion, CollectionBoardGame, CollectionBoardGameExpansion, WishlistBoardGame, BoardgameType
|
|
||||||
import auth_manager
|
|
||||||
|
|
||||||
authenticated_session: requests.Session = requests.Session()
|
|
||||||
|
|
||||||
def url_to_xml_object(url: HttpUrl) -> ET.Element:
|
|
||||||
r = authenticated_session.get(url)
|
|
||||||
assert r.status_code == 200 or r.status_code == 202, "Got {} status code".format(r.status_code)
|
|
||||||
root = ET.fromstring(r.content)
|
|
||||||
return root
|
|
||||||
|
|
||||||
def get_boardgame(boardgame_id: int) -> 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]:
|
|
||||||
|
|
||||||
def divide_list_in_chunks(list_to_divide: list[int], chunk_size: int = 20):
|
|
||||||
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] = []
|
|
||||||
|
|
||||||
#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:
|
|
||||||
|
|
||||||
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(**boardgame_dict)
|
|
||||||
case "boardgameexpansion":
|
|
||||||
boardgame = BoardGameExpansion(**boardgame_dict)
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
return boardgame
|
|
||||||
|
|
||||||
def convert_collection_xml_to_boardgame(boardgame_extra_info: BoardGame, collection_boardgame_xml: ET.Element) -> BoardGame:
|
|
||||||
|
|
||||||
boardgame_type = collection_boardgame_xml.get('subtype')
|
|
||||||
|
|
||||||
price_paid = collection_boardgame_xml.find('privateinfo').get('pricepaid')
|
|
||||||
if price_paid == '':
|
|
||||||
print(boardgame_extra_info.name)
|
|
||||||
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')
|
|
||||||
|
|
||||||
collection_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,
|
|
||||||
**collection_boardgame_dict
|
|
||||||
}
|
|
||||||
|
|
||||||
match boardgame_type:
|
|
||||||
case "boardgame":
|
|
||||||
boardgame = CollectionBoardGame(**boardgame_dict)
|
|
||||||
case "boardgameexpansion":
|
|
||||||
boardgame = CollectionBoardGameExpansion(**boardgame_dict)
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
return boardgame
|
|
||||||
|
|
||||||
#Creates list of board games from a collection '/collection' URL
|
|
||||||
def get_boardgames_from_collection_url(collection_url: str) -> list[BoardGame]:
|
|
||||||
collection_xml = url_to_xml_object(collection_url)
|
|
||||||
|
|
||||||
collection_list: list[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]
|
|
||||||
boardgame = convert_collection_xml_to_boardgame(boardgame_extra, boardgame_item)
|
|
||||||
collection_list.append(boardgame)
|
|
||||||
current_index += 1
|
|
||||||
|
|
||||||
return collection_list
|
|
||||||
|
|
||||||
def get_user_owned_collection() -> list[BoardGame]:
|
|
||||||
url_no_expansions = 'https://boardgamegeek.com/xmlapi2/collection?username={}&own=1&stats=1&excludesubtype=boardgameexpansion&showprivate=1&version=1'.format(auth_manager.username)
|
|
||||||
owned_boardgames = get_boardgames_from_collection_url(url_no_expansions)
|
|
||||||
|
|
||||||
url_only_expansions = 'https://boardgamegeek.com/xmlapi2/collection?username={}&own=1&stats=1&subtype=boardgameexpansion&showprivate=1&version=1'.format(auth_manager.username)
|
|
||||||
owned_boardgame_expansions = get_boardgames_from_collection_url(url_only_expansions)
|
|
||||||
|
|
||||||
owned_boardgames += owned_boardgame_expansions
|
|
||||||
|
|
||||||
return owned_boardgames
|
|
||||||
|
|
||||||
def get_user_wishlist_collection() -> list[BoardGame]:
|
|
||||||
url = 'https://boardgamegeek.com/xmlapi2/collection?username={}&wishlist=1&stats=1&showprivate=1&version=1'.format(auth_manager.username)
|
|
||||||
wishlisted_boardgames = get_boardgames_from_collection_url(url)
|
|
||||||
|
|
||||||
return wishlisted_boardgames
|
|
||||||
|
|
||||||
|
|
||||||
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)
|
|
||||||
|
|
@ -1,42 +0,0 @@
|
||||||
from pydantic import BaseModel, HttpUrl
|
|
||||||
from datetime import date
|
|
||||||
from enum import Enum
|
|
||||||
|
|
||||||
class BoardgameType(Enum):
|
|
||||||
BOARDGAME = 'boardgame'
|
|
||||||
BOARDGAMEEXPANSION = 'boardgameexpansion'
|
|
||||||
COLLECTIONBOARDGAME = 'collectionboardgame'
|
|
||||||
COLLECTIONBOARDGAMEEXPANSION = 'collectionboardgameexpansion'
|
|
||||||
WISHLISTBOARDGAME = 'wishlistboardgame'
|
|
||||||
|
|
||||||
|
|
||||||
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 CollectionBoardGame(BoardGame):
|
|
||||||
price_paid: float
|
|
||||||
acquisition_date: date
|
|
||||||
acquired_from: str
|
|
||||||
type: BoardgameType = BoardgameType.COLLECTIONBOARDGAME
|
|
||||||
|
|
||||||
class CollectionBoardGameExpansion(CollectionBoardGame):
|
|
||||||
type: BoardgameType = BoardgameType.COLLECTIONBOARDGAMEEXPANSION
|
|
||||||
|
|
||||||
class WishlistBoardGame(BoardGame):
|
|
||||||
wishlist_priority: int
|
|
||||||
type: BoardgameType = BoardgameType.WISHLISTBOARDGAME
|
|
||||||
33
main.py
33
main.py
|
|
@ -1,33 +0,0 @@
|
||||||
from typing import Union
|
|
||||||
from fastapi import FastAPI
|
|
||||||
|
|
||||||
from classes.boardgame import BoardGame, BoardGameExpansion, CollectionBoardGame, CollectionBoardGameExpansion, WishlistBoardGame
|
|
||||||
from bgg_connection import get_boardgame, get_user_owned_collection, get_user_wishlist_collection
|
|
||||||
|
|
||||||
app = FastAPI()
|
|
||||||
|
|
||||||
|
|
||||||
@app.get("/")
|
|
||||||
def read_root():
|
|
||||||
return {"Hello": "World"}
|
|
||||||
|
|
||||||
|
|
||||||
@app.get("/items/{item_id}")
|
|
||||||
def read_item(item_id: int, q: Union[str, None] = None):
|
|
||||||
return {"item_id": item_id, "q": q}
|
|
||||||
|
|
||||||
@app.get("/boardgame/{boardgame_id}", response_model=BoardGame)
|
|
||||||
def get_boardgame_by_id(boardgame_id: int):
|
|
||||||
requested_boardgame: BoardGame = get_boardgame(boardgame_id)
|
|
||||||
return requested_boardgame
|
|
||||||
|
|
||||||
@app.get("/collection", response_model=list[CollectionBoardGame])
|
|
||||||
def get_owned_collection():
|
|
||||||
requested_collection: list[CollectionBoardGame] = get_user_owned_collection()
|
|
||||||
return requested_collection
|
|
||||||
|
|
||||||
|
|
||||||
@app.get("/wishlist", response_model=list[WishlistBoardGame])
|
|
||||||
def get_wishlist_collection():
|
|
||||||
requested_collection: list[WishlistBoardGame] = get_user_wishlist_collection()
|
|
||||||
return requested_collection
|
|
||||||
|
|
@ -1,3 +0,0 @@
|
||||||
#Rename this file to auth.yaml and provide correct values below to authenticate to bgg
|
|
||||||
username: username_example
|
|
||||||
password: password_example
|
|
||||||
Loading…
Reference in a new issue