419 lines
16 KiB
Python
419 lines
16 KiB
Python
"""
|
|
Project: Magic Tracker
|
|
Author: Edward Middleton-Smith
|
|
Shuffle & Skirmish
|
|
|
|
Technology: DataStores
|
|
Feature: MTG DataStore
|
|
|
|
Description:
|
|
Datastore for MTG game data
|
|
"""
|
|
|
|
# internal
|
|
import lib.argument_validation as av
|
|
from business_objects.tcg.mtg_deck import MTG_Deck, Parameters_MTG_Deck
|
|
from business_objects.tcg.mtg_deck_commander_bracket import MTG_Deck_Commander_Bracket, Parameters_MTG_Deck_Commander_Bracket
|
|
from business_objects.tcg.mtg_game import MTG_Game, Parameters_MTG_Game
|
|
from business_objects.tcg.mtg_game_player import MTG_Game_Player, Parameters_MTG_Game_Player, MTG_Game_Player_Temp
|
|
from business_objects.tcg.mtg_game_round import MTG_Game_Round, Parameters_MTG_Game_Round, MTG_Game_Round_Temp
|
|
from business_objects.tcg.mtg_game_round_player_damage import MTG_Game_Round_Player_Damage, Parameters_MTG_Game_Round_Player_Damage, MTG_Game_Round_Player_Damage_Temp
|
|
from business_objects.tcg.mtg_game_session_member import MTG_Game_Session_Member, Parameters_MTG_Game_Session_Member
|
|
from business_objects.tcg.statistic import Statistic, Statistic_Temp, Parameters_Statistic
|
|
from business_objects.sql_error import SQL_Error, Parameters_SQL_Error
|
|
from datastores.datastore_base import DataStore_Base
|
|
from helpers.helper_app import Helper_App
|
|
from helpers.helper_db_sql import Helper_DB_SQL
|
|
from extensions import db
|
|
# external
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from sqlalchemy import Uuid
|
|
from sqlalchemy.types import Text, Boolean, Integer
|
|
from sqlalchemy.dialects.postgresql import TIMESTAMP
|
|
from datetime import datetime
|
|
|
|
db = SQLAlchemy()
|
|
|
|
|
|
class DataStore_MTG(DataStore_Base):
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
@classmethod
|
|
def get_many_mtg_game(cls, game_filters):
|
|
_m = f'{cls.__qualname__}.get_many_mtg_game'
|
|
# user = cls.get_user_session()
|
|
argument_dict = game_filters.to_json()
|
|
argument_types = Parameters_MTG_Game.get_type_hints()
|
|
# Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
|
|
games = []
|
|
errors = []
|
|
try:
|
|
game_result = cls.db_function_execute('tcg.public.FN_TCG_MTG_Game_Get_Many', argument_dict, argument_types)
|
|
game_result_set = game_result.fetchall()
|
|
# Helper_App.console_log(f'raw games: {game_result_set}')
|
|
games = []
|
|
game_indexes = {}
|
|
for row in game_result_set:
|
|
new_game = MTG_Game.from_db_mtg_game(row)
|
|
game_indexes[new_game.game_id] = len(games)
|
|
games.append(new_game)
|
|
# Helper_App.console_log(f'game {str(type(new_game))}: {new_game}')
|
|
except Exception as e:
|
|
# Helper_App.console_log(f'Error: {str(e)}')
|
|
error = SQL_Error()
|
|
error.msg = str(e)
|
|
errors.append(error)
|
|
|
|
return games, errors
|
|
|
|
@classmethod
|
|
def save_mtg_game(cls, game):
|
|
_m = f'{cls.__qualname__}.save_mtg_game'
|
|
user = cls.get_user_session()
|
|
guid = Helper_DB_SQL.create_guid()
|
|
game_id = None
|
|
success = None
|
|
argument_dict = {
|
|
'a_comment': 'Save game'
|
|
, 'a_end_on': game.end_on
|
|
, 'a_game_id': game.game_id
|
|
, 'a_guid': guid
|
|
, 'a_location_name': game.location_name
|
|
, 'a_notes': game.notes
|
|
, 'a_starting_life': game.starting_life
|
|
, 'a_start_on': game.start_on
|
|
, 'a_user_id': user.user_id
|
|
, 'a_do_delete': False
|
|
, 'a_is_commander': True
|
|
, 'a_is_draft': False
|
|
, 'a_is_sealed': False
|
|
, 'o_game_id': game_id
|
|
, 'o_success': success
|
|
}
|
|
argument_type_hints = {
|
|
'a_comment': Text
|
|
, 'a_end_on': TIMESTAMP
|
|
, 'a_game_id': Integer
|
|
, 'a_guid': Uuid
|
|
, 'a_location_name': Text
|
|
, 'a_notes': Text
|
|
, 'a_starting_life': Integer
|
|
, 'a_start_on': TIMESTAMP
|
|
, 'a_user_id': Integer
|
|
, 'a_do_delete': Boolean
|
|
, 'a_is_commander': Boolean
|
|
, 'a_is_draft': Boolean
|
|
, 'a_is_sealed': Boolean
|
|
, 'o_game_id': Integer
|
|
, 'o_success': Boolean
|
|
}
|
|
# Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
|
|
errors = []
|
|
try:
|
|
rows = cls.db_procedure_execute('tcg.public.USP_TCG_MTG_Game_Save', argument_dict, argument_type_hints)
|
|
row = rows[0] if rows else None
|
|
game_id = row['o_game_id'] if row else None
|
|
success = row['o_success'] if row else False
|
|
# Helper_App.console_log(f'Game ID: {game_id}')
|
|
# Helper_App.console_log(f'Success: {success}')
|
|
if not success:
|
|
errors = cls.get_many_error(guid = guid)
|
|
except Exception as e:
|
|
# Helper_App.console_log(f'Error: {str(e)}')
|
|
error = SQL_Error()
|
|
error.msg = str(e)
|
|
errors.append(error)
|
|
|
|
cls.clear_error(guid = guid)
|
|
|
|
return game_id, errors
|
|
|
|
@classmethod
|
|
def get_many_mtg_game_player(cls, player_filters):
|
|
_m = f'{cls.__qualname__}.get_many_mtg_game_player'
|
|
argument_dict = player_filters.to_json()
|
|
argument_types = Parameters_MTG_Game_Player.get_type_hints()
|
|
# Helper_App.console_log(f'{_m}\nargument_dict: {argument_dict}')
|
|
|
|
players = []
|
|
errors = []
|
|
try:
|
|
result = cls.db_function_execute('tcg.public.FN_TCG_MTG_Game_Player_Get_Many', argument_dict, argument_types)
|
|
result_set = result.fetchall()
|
|
# Helper_App.console_log(f'raw players: {result_set}')
|
|
for row in result_set:
|
|
new_player = MTG_Game_Player.from_db_mtg_game_player(row)
|
|
players.append(new_player)
|
|
# Helper_App.console_log(f'player {str(type(new_player))}: {new_player}')
|
|
except Exception as e:
|
|
# Helper_App.console_log(f'Error: {str(e)}')
|
|
error = SQL_Error()
|
|
error.msg = str(e)
|
|
errors.append(error)
|
|
|
|
return players, errors
|
|
|
|
@classmethod
|
|
def get_many_mtg_game_round(cls, round_filters):
|
|
_m = f'{cls.__qualname__}.get_many_mtg_game_round'
|
|
argument_dict = round_filters.to_json()
|
|
argument_types = Parameters_MTG_Game_Round.get_type_hints()
|
|
# Helper_App.console_log(f'{_m}\nargument_dict: {argument_dict}')
|
|
|
|
rounds = []
|
|
errors = []
|
|
try:
|
|
result = cls.db_function_execute('tcg.public.FN_TCG_MTG_Game_Round_Get_Many', argument_dict, argument_types)
|
|
result_set = result.fetchall()
|
|
# Helper_App.console_log(f'raw rounds: {result_set}')
|
|
for row in result_set:
|
|
new_round = MTG_Game_Round.from_db_mtg_game_round(row)
|
|
rounds.append(new_round)
|
|
# Helper_App.console_log(f'round {str(type(new_round))}: {new_round}')
|
|
except Exception as e:
|
|
# Helper_App.console_log(f'Error: {str(e)}')
|
|
error = SQL_Error()
|
|
error.msg = str(e)
|
|
errors.append(error)
|
|
|
|
return rounds, errors
|
|
|
|
@classmethod
|
|
def get_many_mtg_game_round_player_damage(cls, damage_filters):
|
|
_m = f'{cls.__qualname__}.get_many_mtg_game_round_player_damage'
|
|
argument_dict = damage_filters.to_json()
|
|
argument_types = Parameters_MTG_Game_Round_Player_Damage.get_type_hints()
|
|
# Helper_App.console_log(f'{_m}\nargument_dict: {argument_dict}')
|
|
|
|
damages = []
|
|
errors = []
|
|
try:
|
|
result = cls.db_function_execute('tcg.public.FN_TCG_MTG_Game_Round_Player_Damage_Get_Many', argument_dict, argument_types)
|
|
result_set = result.fetchall()
|
|
# Helper_App.console_log(f'raw damages: {result_set}')
|
|
for row in result_set:
|
|
new_damage = MTG_Game_Round_Player_Damage.from_db_mtg_game_round_player_damage(row)
|
|
damages.append(new_damage)
|
|
# Helper_App.console_log(f'damage {str(type(new_damage))}: {new_damage.damage_id}')
|
|
except Exception as e:
|
|
# Helper_App.console_log(f'Error: {str(e)}')
|
|
error = SQL_Error()
|
|
error.msg = str(e)
|
|
errors.append(error)
|
|
|
|
return damages, errors
|
|
|
|
@classmethod
|
|
def get_many_mtg_deck(cls, deck_filters):
|
|
_m = f'{cls.__qualname__}.get_many_mtg_deck'
|
|
argument_dict = deck_filters.to_json()
|
|
argument_types = Parameters_MTG_Deck.get_type_hints()
|
|
# Helper_App.console_log(f'{_m}\nargument_dict: {argument_dict}')
|
|
|
|
decks = []
|
|
errors = []
|
|
try:
|
|
result = cls.db_function_execute('tcg.public.FN_TCG_MTG_Deck_Get_Many', argument_dict, argument_types)
|
|
result_set = result.fetchall()
|
|
# Helper_App.console_log(f'raw decks: {result_set}')
|
|
for row in result_set:
|
|
new_deck = MTG_Deck.from_db_mtg_deck(row)
|
|
decks.append(new_deck)
|
|
# Helper_App.console_log(f'deck {str(type(new_deck))}: {new_deck}')
|
|
except Exception as e:
|
|
# Helper_App.console_log(f'Error: {str(e)}')
|
|
error = SQL_Error()
|
|
error.msg = str(e)
|
|
errors.append(error)
|
|
|
|
return decks, errors
|
|
|
|
|
|
@classmethod
|
|
def save_mtg_game_player(cls, players):
|
|
_m = f'{cls.__qualname__}.save_mtg_game_player'
|
|
user = cls.get_user_session()
|
|
guid = Helper_DB_SQL.create_guid()
|
|
success = None
|
|
argument_dict = {
|
|
'a_comment': 'Save game player'
|
|
, 'a_guid': guid
|
|
, 'a_user_id': user.user_id
|
|
, 'o_success': success
|
|
}
|
|
argument_type_hints = {
|
|
'a_comment': Text
|
|
, 'a_guid': Uuid
|
|
, 'a_user_id': Integer
|
|
, 'o_success': Boolean
|
|
}
|
|
# Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
|
|
objs_player_temp = []
|
|
for player in players:
|
|
obj_player_temp = MTG_Game_Player_Temp.from_player(player = player, guid = guid)
|
|
objs_player_temp.append(obj_player_temp)
|
|
|
|
success = False
|
|
errors = []
|
|
try:
|
|
cls.upload_bulk(
|
|
permanent_table_name = MTG_Game_Player_Temp.__tablename__
|
|
, records = objs_player_temp
|
|
, batch_size = 1000
|
|
)
|
|
rows = cls.db_procedure_execute('tcg.public.USP_TCG_MTG_Game_Player_Save', argument_dict, argument_type_hints)
|
|
row = rows[0] if rows else None
|
|
success = row['o_success'] if row else False
|
|
# Helper_App.console_log(f'Success: {success}')
|
|
if not success:
|
|
errors = cls.get_many_error(guid = guid)
|
|
# Helper_App.console_log(f'Errors: {str(errors)}')
|
|
except Exception as e:
|
|
# Helper_App.console_log(f'Error: {str(e)}')
|
|
error = SQL_Error()
|
|
error.msg = str(e)
|
|
errors.append(error)
|
|
|
|
cls.clear_error(guid = guid)
|
|
|
|
return success, errors
|
|
|
|
@classmethod
|
|
def save_mtg_game_round_player_damage(cls, rounds, damages):
|
|
_m = f'{cls.__qualname__}.save_mtg_game_round_player_damage'
|
|
user = cls.get_user_session()
|
|
guid = Helper_DB_SQL.create_guid()
|
|
success = None
|
|
argument_dict = {
|
|
'a_comment': 'Save game player'
|
|
, 'a_guid': guid
|
|
, 'a_user_id': user.user_id
|
|
, 'o_success': success
|
|
}
|
|
argument_type_hints = {
|
|
'a_comment': Text
|
|
, 'a_guid': Uuid
|
|
, 'a_user_id': Integer
|
|
, 'o_success': Boolean
|
|
}
|
|
# Helper_App.console_log(f'argument_dict: {argument_dict}')
|
|
|
|
objs_round_temp = []
|
|
for round in rounds:
|
|
obj_round_temp = MTG_Game_Round_Temp.from_round(round = round, guid = guid)
|
|
objs_round_temp.append(obj_round_temp)
|
|
|
|
objs_damage_temp = []
|
|
for damage in damages:
|
|
obj_damage_temp = MTG_Game_Round_Player_Damage_Temp.from_damage(damage = damage, guid = guid)
|
|
objs_damage_temp.append(obj_damage_temp)
|
|
|
|
success = False
|
|
errors = []
|
|
try:
|
|
cls.upload_bulk(
|
|
permanent_table_name = MTG_Game_Round_Temp.__tablename__
|
|
, records = objs_round_temp
|
|
, batch_size = 1000
|
|
)
|
|
cls.upload_bulk(
|
|
permanent_table_name = MTG_Game_Round_Player_Damage_Temp.__tablename__
|
|
, records = objs_damage_temp
|
|
, batch_size = 1000
|
|
)
|
|
rows = cls.db_procedure_execute('tcg.public.USP_TCG_MTG_Game_Round_Damage_Save', argument_dict, argument_type_hints)
|
|
row = rows[0] if rows else None
|
|
success = row['o_success'] if row else False
|
|
# Helper_App.console_log(f'Success: {success}')
|
|
if not success:
|
|
errors = cls.get_many_error(guid = guid)
|
|
# Helper_App.console_log(f'Errors: {str(errors)}')
|
|
except Exception as e:
|
|
# Helper_App.console_log(f'Error: {str(e)}')
|
|
error = SQL_Error()
|
|
error.msg = str(e)
|
|
errors.append(error)
|
|
|
|
cls.clear_error(guid = guid)
|
|
|
|
return success, errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_mtg_deck_commander_bracket(cls, parameters_commander_bracket):
|
|
_m = f'{cls.__qualname__}.get_many_mtg_deck_commander_bracket'
|
|
argument_dict = parameters_commander_bracket.to_json()
|
|
argument_types = Parameters_MTG_Deck_Commander_Bracket.get_type_hints()
|
|
# Helper_App.console_log(f'{_m}\nargument_dict: {argument_dict}')
|
|
|
|
commander_brackets = []
|
|
errors = []
|
|
try:
|
|
result = cls.db_function_execute('tcg.public.FN_TCG_MTG_Deck_Commander_Bracket_Get_Many', argument_dict, argument_types)
|
|
result_set = result.fetchall()
|
|
# Helper_App.console_log(f'raw commander brackets: {result_set}')
|
|
for row in result_set:
|
|
new_commander_bracket = MTG_Deck_Commander_Bracket.from_db_mtg_deck_commander_bracket(row)
|
|
commander_brackets.append(new_commander_bracket)
|
|
# Helper_App.console_log(f'commander bracket {str(type(new_commander_bracket))}: {new_commander_bracket}')
|
|
except Exception as e:
|
|
# Helper_App.console_log(f'Error: {str(e)}')
|
|
error = SQL_Error()
|
|
error.msg = str(e)
|
|
errors.append(error)
|
|
|
|
return commander_brackets, errors
|
|
|
|
|
|
@classmethod
|
|
def get_many_statistic(cls, parameters_statistic):
|
|
_m = f'{cls.__qualname__}.get_many_statistic'
|
|
argument_dict = parameters_statistic.to_json()
|
|
argument_types = Parameters_Statistic.get_type_hints()
|
|
# Helper_App.console_log(f'{_m}\nargument_dict: {argument_dict}')
|
|
|
|
statistics = []
|
|
errors = []
|
|
try:
|
|
result = cls.db_function_execute('tcg.public.FN_TCG_Statistic_Get_Many', argument_dict, argument_types)
|
|
result_set = result.fetchall()
|
|
# Helper_App.console_log(f'raw commander brackets: {result_set}')
|
|
for row in result_set:
|
|
new_statistic = Statistic.from_db_statistic(row)
|
|
statistics.append(new_statistic)
|
|
# Helper_App.console_log(f'statistic {str(type(new_statistic))}: {new_statistic}')
|
|
except Exception as e:
|
|
# Helper_App.console_log(f'Error: {str(e)}')
|
|
error = SQL_Error()
|
|
error.msg = str(e)
|
|
errors.append(error)
|
|
|
|
return statistics, errors
|
|
|
|
@classmethod
|
|
def get_many_mtg_game_session_member(cls, member_filters):
|
|
argument_dict = member_filters.to_json()
|
|
argument_types = Parameters_MTG_Game_Session_Member.get_type_hints()
|
|
|
|
members = []
|
|
errors = []
|
|
try:
|
|
result = cls.db_function_execute('tcg.public.FN_TCG_MTG_Game_Session_Member_Get_Many', argument_dict, argument_types)
|
|
result_set = result.fetchall()
|
|
for row in result_set:
|
|
new_member = MTG_Game_Session_Member.from_db_mtg_game_player(row)
|
|
members.append(new_member)
|
|
except Exception as e:
|
|
error = SQL_Error()
|
|
error.msg = str(e)
|
|
errors.append(error)
|
|
|
|
return members, errors
|
|
|
|
|