266 lines
No EOL
10 KiB
Python
266 lines
No EOL
10 KiB
Python
from typing import Union, Dict
|
|
from pydantic import BaseModel
|
|
from sqlmodel import Session
|
|
from threading import Thread
|
|
|
|
|
|
from fastapi import FastAPI, Depends
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from contextlib import asynccontextmanager
|
|
|
|
from src.classes import boardgame_classes, play_classes, statistic_classes, people_classes, player_classes
|
|
from src.modules import data_connection, statistic_creator
|
|
from src.filters import boardgame_filters, play_filters
|
|
|
|
is_refreshing = False
|
|
|
|
async def get_session():
|
|
with Session(data_connection.get_db_engine()) as session:
|
|
yield session
|
|
|
|
def refresh_data():
|
|
global is_refreshing
|
|
|
|
is_refreshing = True
|
|
data_connection.delete_database()
|
|
data_connection.create_db_and_tables()
|
|
with Session(data_connection.get_db_engine()) as session:
|
|
statistic_creator.clear_cache()
|
|
data_connection.get_user_owned_collection(session)
|
|
data_connection.get_user_wishlist_collection(session)
|
|
data_connection.get_user_preordered_collection(session)
|
|
data_connection.get_user_collection(session)
|
|
data_connection.get_plays(session)
|
|
|
|
print('DONE')
|
|
|
|
is_refreshing = False
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
# Startup
|
|
#data_connection.delete_database()
|
|
data_connection.create_db_and_tables()
|
|
|
|
#refresh_data()
|
|
yield
|
|
# Shutdown
|
|
|
|
app = FastAPI(lifespan=lifespan)
|
|
|
|
origins = [
|
|
"http://127.0.0.1:8080",
|
|
"http://0.0.0.0:8080" #Will become something like 'bordspellen2.yarnecoppens.com'
|
|
]
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
#expansion filtering parameters
|
|
class BoardgameFilterParams(BaseModel):
|
|
filter_expansions_out: bool = False
|
|
only_expansions: bool = False
|
|
|
|
def do_filtering(self,boardgame_list) -> list[boardgame_classes.BoardGame]:
|
|
if self.filter_expansions_out:
|
|
boardgame_list = boardgame_filters.filter_expansions_out(boardgame_list)
|
|
|
|
if self.only_expansions:
|
|
boardgame_list = boardgame_filters.filter_non_expansions_out(boardgame_list)
|
|
|
|
return boardgame_list
|
|
|
|
class PlayFilterParams(BaseModel):
|
|
filter_expansions_out: bool = False
|
|
only_expansions: bool = False
|
|
|
|
def do_filtering(self, play_list) -> list[play_classes.Play]:
|
|
if self.filter_expansions_out:
|
|
play_list = play_filters.filter_expansions_out(play_list)
|
|
|
|
if self.only_expansions:
|
|
play_list = play_filters.filter_non_expansions_out(play_list)
|
|
|
|
return play_list
|
|
|
|
|
|
@app.get("/")
|
|
def read_root():
|
|
return {"Hello": "World"}
|
|
|
|
@app.get('/refresh')
|
|
def refresh():
|
|
if not is_refreshing:
|
|
Thread(target=refresh_data).start()
|
|
return {"Status": "Started refresh"}
|
|
else:
|
|
return {"Status": "Already refreshing"}
|
|
|
|
|
|
@app.get("/boardgame", response_model=boardgame_classes.BoardGamePublic)
|
|
def get_boardgame_by_id(id: int, session: Session = Depends(get_session)):
|
|
requested_boardgame = data_connection.get_boardgame(session, id)
|
|
return requested_boardgame
|
|
|
|
@app.get("/owned", response_model=list[boardgame_classes.BoardGamePublicNoPlays])
|
|
def get_owned_collection(query: BoardgameFilterParams = Depends(), session: Session = Depends(get_session)):
|
|
to_return_boardgames = data_connection.get_user_owned_collection(session)
|
|
|
|
to_return_boardgames = query.do_filtering(to_return_boardgames)
|
|
|
|
to_return_boardgames.sort(key=lambda x: x.name)
|
|
|
|
return to_return_boardgames
|
|
|
|
@app.get('/collection', response_model=list[boardgame_classes.BoardGamePublicNoPlays])
|
|
def get_collection(query: BoardgameFilterParams = Depends(), session: Session = Depends(get_session)):
|
|
to_return_boardgames = data_connection.get_user_collection(session)
|
|
|
|
to_return_boardgames = query.do_filtering(to_return_boardgames)
|
|
|
|
to_return_boardgames.sort(key=lambda x: x.name)
|
|
|
|
return to_return_boardgames
|
|
|
|
@app.get('/designers', response_model=list[people_classes.DesignerPublic])
|
|
def get_designers(session: Session = Depends(get_session)):
|
|
to_return_designers = data_connection.get_all_designers(session)
|
|
return to_return_designers
|
|
|
|
@app.get("/wishlist", response_model=list[boardgame_classes.BoardGamePublicNoPlays])
|
|
def get_wishlist_collection(priority: int = 0, query: BoardgameFilterParams = Depends(), session: Session = Depends(get_session)):
|
|
|
|
to_return_boardgames = data_connection.get_user_wishlist_collection(session, priority)
|
|
|
|
to_return_boardgames = query.do_filtering(to_return_boardgames)
|
|
|
|
to_return_boardgames.sort(key=lambda x: x.name)
|
|
|
|
return to_return_boardgames
|
|
|
|
@app.get("/preordered", response_model=list[boardgame_classes.BoardGamePublicNoPlays])
|
|
def get_preordered_collection(query: BoardgameFilterParams = Depends(), session: Session = Depends(get_session)):
|
|
|
|
to_return_boardgames = data_connection.get_user_preordered_collection(session)
|
|
|
|
to_return_boardgames = query.do_filtering(to_return_boardgames)
|
|
|
|
to_return_boardgames.sort(key=lambda x: x.name)
|
|
|
|
return to_return_boardgames
|
|
|
|
|
|
@app.get("/plays", response_model=list[play_classes.PlayPublic])
|
|
def get_plays(query: PlayFilterParams = Depends(), boardgame_id: int = -1, session: Session = Depends(get_session)):
|
|
|
|
requested_plays = data_connection.get_plays(session)
|
|
|
|
requested_plays = query.do_filtering(requested_plays)
|
|
|
|
if boardgame_id > -1:
|
|
requested_plays = play_filters.filter_only_specific_boardgame(boardgame_id, requested_plays)
|
|
|
|
return requested_plays
|
|
|
|
@app.get('/players', response_model=list[player_classes.PlayerPublicNoPlayPlayers])
|
|
def get_players(session: Session = Depends(get_session)):
|
|
requested_players = data_connection.get_all_players(session)
|
|
|
|
return requested_players
|
|
|
|
@app.get('/player', response_model=player_classes.PlayerPublic)
|
|
def get_player(player_name: str, session: Session = Depends(get_session)):
|
|
requested_players = data_connection.get_player(player_name, session)
|
|
|
|
return requested_players
|
|
|
|
@app.get('/players_from_play', response_model=list[play_classes.PlayPlayerPublic])
|
|
def get_players_from_play(play_id: int, session: Session = Depends(get_session)):
|
|
requested_players = data_connection.get_players_from_play(session, play_id)
|
|
|
|
return requested_players
|
|
|
|
@app.get('/statistics/h_index', response_model=statistic_classes.NumberStatistic)
|
|
def get_h_index(query: BoardgameFilterParams = Depends(), session: Session = Depends(get_session)):
|
|
statistic_to_return = statistic_creator.get_h_index(session, query)
|
|
return statistic_to_return
|
|
|
|
@app.get('/statistics/amount_of_games', response_model=statistic_classes.NumberStatistic)
|
|
def get_amount_of_games(query: BoardgameFilterParams = Depends(), session: Session = Depends(get_session)):
|
|
|
|
statistic_to_return = statistic_creator.get_total_owned_games(session, query)
|
|
|
|
return statistic_to_return
|
|
|
|
@app.get('/statistics/total_collection_cost', response_model=statistic_classes.NumberStatistic)
|
|
def get_total_collection_cost(query: BoardgameFilterParams = Depends(), session: Session = Depends(get_session)):
|
|
|
|
statistic_to_return = statistic_creator.get_total_owned_collection_cost(session, query)
|
|
|
|
return statistic_to_return
|
|
|
|
@app.get('/statistics/amount_of_games_over_time', response_model=statistic_classes.TimeLineStatistic)
|
|
def get_amount_of_games_over_time(query: BoardgameFilterParams = Depends(), day_step: int = 1, session: Session = Depends(get_session)):
|
|
|
|
statistic_to_return = statistic_creator.get_amount_of_games_over_time(session, query, day_step)
|
|
|
|
return statistic_to_return
|
|
|
|
@app.get('/statistics/games_played_per_year', response_model=statistic_classes.TimeLineStatistic)
|
|
def get_amount_of_games_played_per_year(query: PlayFilterParams = Depends(), session: Session = Depends(get_session)):
|
|
|
|
statistic_to_return = statistic_creator.get_amount_of_games_played_per_year(session, query)
|
|
|
|
return statistic_to_return
|
|
|
|
|
|
@app.get('/statistics/most_expensive_games', response_model=statistic_classes.GamesStatistic)
|
|
def get_most_expensive_games(query: BoardgameFilterParams = Depends(), top_amount: int = 10, session: Session = Depends(get_session)):
|
|
|
|
statistic_to_return = statistic_creator.get_most_expensive_games(session, query, top_amount)
|
|
|
|
return statistic_to_return
|
|
|
|
@app.get('/statistics/cheapest_per_play', response_model=statistic_classes.GamesStatistic)
|
|
def get_cheapest_per_play_games(query: BoardgameFilterParams = Depends(), top_amount: int = 10, session: Session = Depends(get_session)):
|
|
|
|
statistic_to_return = statistic_creator.get_cheapest_per_play_games(session, query, top_amount)
|
|
|
|
return statistic_to_return
|
|
|
|
@app.get('/statistics/shelf_of_shame', response_model=statistic_classes.GamesStatistic)
|
|
def get_shelf_of_shame(query: BoardgameFilterParams = Depends(), session: Session = Depends(get_session)):
|
|
|
|
statistic_to_return = statistic_creator.get_shelf_of_shame(session, query)
|
|
|
|
return statistic_to_return
|
|
|
|
@app.get('/statistics/winrate', response_model=statistic_classes.PlayerStatistic)
|
|
def get_winrate(player_name: str | None = None, session: Session = Depends(get_session)):
|
|
statistic_to_return = statistic_creator.get_winrate(session, player_name)
|
|
|
|
return statistic_to_return
|
|
|
|
@app.get('/statistics/winrate_over_time', response_model=Union[statistic_classes.TimeLineStatistic, Dict[str,statistic_classes.TimeLineStatistic]])
|
|
def get_winrate_over_time(player_name: str | None = None, day_step: int = 1, session: Session=Depends(get_session)):
|
|
statistic_to_return = statistic_creator.get_winrate_over_time(session, player_name, day_step)
|
|
|
|
return statistic_to_return
|
|
|
|
|
|
@app.get('/statistics/most_bought_designer', response_model=statistic_classes.GamesStatistic)
|
|
def get_most_bought_from_designer(query: BoardgameFilterParams = Depends(), session: Session=Depends(get_session)):
|
|
statistic_to_return = statistic_creator.get_most_bought_designer(session, query)
|
|
|
|
return statistic_to_return
|
|
|
|
@app.get('/statistics/most_bought_artist', response_model=statistic_classes.GamesStatistic)
|
|
def get_most_bought_from_artist(query: BoardgameFilterParams = Depends(), session: Session=Depends(get_session)):
|
|
statistic_to_return = statistic_creator.get_most_bought_artist(session, query)
|
|
|
|
return statistic_to_return |