bgg_api/src/main.py

from typing import Union
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
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:
        data_connection.get_user_collection(session)
        data_connection.get_user_owned_collection(session)
        data_connection.get_user_wishlist_collection(session)
        data_connection.get_plays(session)

    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_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("/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[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/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/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