cal/paper-dynasty-discord
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add cascading delete from Game and Cardset to GameCardsetLink

Browse Source
This commit is contained in:
Cal Corum 2024-10-13 20:06:12 -05:00
parent 59fa207212
commit b2fd6fc9c4
1 changed files with 54 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

108
in_game/gameplay_models.py
View File

@ -40,7 +40,7 @@ class Game(SQLModel, table=True):
ai_team: str | None = Field(default=None)
game_type: str | None = Field(default=None)
cardset_links: list[GameCardsetLink] = Relationship(back_populates='game')
cardset_links: list[GameCardsetLink] = Relationship(back_populates='game', cascade_delete=True)
lineups: list['Lineup'] = Relationship(back_populates='game', cascade_delete=True)
@property
@ -67,7 +67,7 @@ class CardsetBase(SQLModel):
class Cardset(CardsetBase, table=True):
game_links: list[GameCardsetLink] = Relationship(back_populates='cardset')
game_links: list[GameCardsetLink] = Relationship(back_populates='cardset', cascade_delete=True)
players: list['Player'] = Relationship(back_populates='cardset')
@ -197,6 +197,58 @@ class Player(PlayerBase, table=True):
lineups: list['Lineup'] = Relationship(back_populates='player', cascade_delete=True)
def player_description(player: Player = None, player_dict: dict = None) -> str:
if player is None and player_dict is None:
err = 'One of "player" or "player_dict" must be included to get full description'
logging.error(f'gameplay_models - player_description - {err}')
raise TypeError(err)
if player is not None:
return f'{player.description} {player.name}'
r_val = f'{player_dict['description']}'
if 'name' in player_dict:
r_val += f' {player_dict["name"]}'
elif 'p_name' in player_dict:
r_val += f' {player_dict["p_name"]}'
return r_val
async def get_player(session: Session, player_id: int, skip_cache: bool = False) -> Player:
if not skip_cache:
this_player = session.get(Player, player_id)
if this_player is not None:
logging.info(f'we found a cached player: {this_player} / created: {this_player.created}')
tdelta = datetime.datetime.now() - this_player.created
logging.debug(f'tdelta: {tdelta}')
if tdelta.total_seconds() < CACHE_LIMIT:
return this_player
else:
session.delete(this_player)
session.commit()
def cache_player(json_data: dict) -> Player:
valid_player = PlayerBase.model_validate(json_data, from_attributes=True)
db_player = Player.model_validate(valid_player)
session.add(db_player)
session.commit()
session.refresh(db_player)
return db_player
p_query = await db_get('players', object_id=player_id, params=[('inc_dex', False)])
if p_query is not None:
if 'id' not in p_query:
p_query['id'] = p_query['player_id']
if 'name' not in p_query:
p_query['name'] = p_query['p_name']
return cache_player(p_query)
err = 'Player not found'
logging.error(f'gameplay_models - get_player - {err}')
class CardBase(SQLModel):
id: int | None = Field(default=None, primary_key=True)
player_id: int = Field(foreign_key='player.id', index=True, ondelete='CASCADE')
@ -273,58 +325,6 @@ class Lineup(SQLModel, table=True):
card: Card = Relationship(back_populates='lineups')
def player_description(player: Player = None, player_dict: dict = None) -> str:
if player is None and player_dict is None:
err = 'One of "player" or "player_dict" must be included to get full description'
logging.error(f'gameplay_models - player_description - {err}')
raise TypeError(err)
if player is not None:
return f'{player.description} {player.name}'
r_val = f'{player_dict['description']}'
if 'name' in player_dict:
r_val += f' {player_dict["name"]}'
elif 'p_name' in player_dict:
r_val += f' {player_dict["p_name"]}'
return r_val
async def get_player(session: Session, player_id: int, skip_cache: bool = False) -> Player:
if not skip_cache:
this_player = session.get(Player, player_id)
if this_player is not None:
logging.info(f'we found a cached player: {this_player} / created: {this_player.created}')
tdelta = datetime.datetime.now() - this_player.created
logging.debug(f'tdelta: {tdelta}')
if tdelta.total_seconds() < CACHE_LIMIT:
return this_player
else:
session.delete(this_player)
session.commit()
def cache_player(json_data: dict) -> Player:
valid_player = PlayerBase.model_validate(json_data, from_attributes=True)
db_player = Player.model_validate(valid_player)
session.add(db_player)
session.commit()
session.refresh(db_player)
return db_player
p_query = await db_get('players', object_id=player_id, params=[('inc_dex', False)])
if p_query is not None:
if 'id' not in p_query:
p_query['id'] = p_query['player_id']
if 'name' not in p_query:
p_query['name'] = p_query['p_name']
return cache_player(p_query)
err = 'Player not found'
logging.error(f'gameplay_models - get_player - {err}')
"""
BEGIN DEVELOPMENT HELPERS
"""