136 lines
No EOL
4.5 KiB
Python
136 lines
No EOL
4.5 KiB
Python
from sqlmodel import create_engine, SQLModel, Session, select
|
|
from src.config import definitions
|
|
from typing import Union
|
|
|
|
from src.classes import boardgame_classes, play_classes
|
|
|
|
sqlite_url = definitions.SQLITE_URL
|
|
|
|
|
|
connect_args = {"check_same_thread": False}
|
|
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
|
|
|
|
def get_session():
|
|
with Session(engine) as session:
|
|
return session
|
|
|
|
def add_boardgame(session: Session, boardgame: Union[
|
|
boardgame_classes.BoardGame, boardgame_classes.BoardGameExpansion,
|
|
boardgame_classes.OwnedBoardGame, boardgame_classes.OwnedBoardGameExpansion,
|
|
boardgame_classes.WishlistBoardGame, boardgame_classes.WishlistBoardGameExpansion]):
|
|
|
|
is_boardgame_present = len(session.exec(
|
|
select(boardgame.__class__).where(boardgame.__class__.id == boardgame.id)
|
|
).all()) != 0
|
|
|
|
if not is_boardgame_present:
|
|
session.add(boardgame)
|
|
|
|
session.commit()
|
|
|
|
def add_multiple_boardgames(session: Session, boardgame_list: list[Union[
|
|
boardgame_classes.BoardGame, boardgame_classes.BoardGameExpansion,
|
|
boardgame_classes.OwnedBoardGame, boardgame_classes.OwnedBoardGameExpansion,
|
|
boardgame_classes.WishlistBoardGame, boardgame_classes.WishlistBoardGameExpansion]]):
|
|
|
|
for boardgame in boardgame_list:
|
|
is_boardgame_present = len(session.exec(
|
|
select(boardgame.__class__).where(boardgame.__class__.id == boardgame.id)
|
|
).all()) != 0
|
|
|
|
if not is_boardgame_present:
|
|
session.add(boardgame)
|
|
|
|
session.commit()
|
|
|
|
def get_boardgame(session: Session, boardgame_id: int) -> Union[
|
|
boardgame_classes.BoardGame, boardgame_classes.BoardGameExpansion]:
|
|
|
|
statement = select(boardgame_classes.BoardGame).where(boardgame_classes.BoardGame.id == boardgame_id)
|
|
|
|
base_boardgames = session.exec(statement).all()
|
|
|
|
statement = select(boardgame_classes.BoardGameExpansion).where(boardgame_classes.BoardGameExpansion.id == boardgame_id)
|
|
|
|
expansion_boardgames = session.exec(statement).all()
|
|
|
|
returned_boardgames = base_boardgames + expansion_boardgames
|
|
|
|
if len(returned_boardgames) == 0:
|
|
boardgame = None
|
|
else:
|
|
boardgame = returned_boardgames[0]
|
|
|
|
return boardgame
|
|
|
|
def get_multiple_boardgames(session: Session, boardgame_ids: list[int]) -> tuple[Union[
|
|
list[boardgame_classes.BoardGame], list[boardgame_classes.BoardGameExpansion]], list[int]]:
|
|
|
|
statement = select(boardgame_classes.BoardGame).where(boardgame_classes.BoardGame.id.in_(boardgame_ids))
|
|
results = session.exec(statement)
|
|
|
|
boardgames = results.all()
|
|
|
|
statement = select(boardgame_classes.BoardGameExpansion).where(boardgame_classes.BoardGameExpansion.id.in_(boardgame_ids))
|
|
results = session.exec(statement)
|
|
|
|
expansions = results.all()
|
|
|
|
boardgames += expansions
|
|
|
|
missing_boardgame_ids = list(filter(lambda x: x not in [boardgame.id for boardgame in boardgames], boardgame_ids))
|
|
|
|
return boardgames, missing_boardgame_ids
|
|
|
|
def get_all_boardgames(session: Session, boardgame_type: SQLModel) -> Union[
|
|
list[boardgame_classes.BoardGame], list[boardgame_classes.BoardGameExpansion],
|
|
list[boardgame_classes.OwnedBoardGame], list[boardgame_classes.OwnedBoardGameExpansion],
|
|
list[boardgame_classes.WishlistBoardGame], list[boardgame_classes.WishlistBoardGameExpansion]]:
|
|
|
|
statement = select(boardgame_type)
|
|
|
|
results = session.exec(statement)
|
|
|
|
boardgame_list = results.all()
|
|
|
|
return boardgame_list
|
|
|
|
def add_play(session: Session, play: play_classes.Play):
|
|
|
|
session.add(play)
|
|
|
|
session.commit()
|
|
session.refresh(play)
|
|
|
|
def add_multiple_plays(session: Session, play_list: list[play_classes.Play]):
|
|
|
|
for play in play_list:
|
|
is_play_present = len(session.exec(select(play_classes.Play).where(play_classes.Play.id == play.id)).all()) != 0
|
|
|
|
if not is_play_present:
|
|
session.add(play)
|
|
|
|
session.commit()
|
|
|
|
def get_plays(session: Session) -> list[play_classes.Play]:
|
|
statement = select(play_classes.Play)
|
|
results = session.exec(statement)
|
|
|
|
play_list = results.all()
|
|
|
|
return play_list
|
|
|
|
def get_players_from_play(session: Session, play_id: int) -> list[play_classes.PlayPlayer]:
|
|
statement = select(play_classes.PlayPlayer).where(play_classes.PlayPlayer.play_id == play_id)
|
|
results = session.exec(statement)
|
|
|
|
player_list = results.all()
|
|
|
|
return player_list
|
|
|
|
def delete_database():
|
|
SQLModel.metadata.drop_all(engine)
|
|
|
|
|
|
def create_db_and_tables():
|
|
SQLModel.metadata.create_all(engine) |