140 lines
4.5 KiB
Python
140 lines
4.5 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from typing import Optional, List
|
|
import logging
|
|
import pydantic
|
|
|
|
from ..db_engine import db, GauntletReward, model_to_dict, chunked, DatabaseError
|
|
from ..dependencies import oauth2_scheme, valid_token, LOG_DATA
|
|
|
|
logging.basicConfig(
|
|
filename=LOG_DATA['filename'],
|
|
format=LOG_DATA['format'],
|
|
level=LOG_DATA['log_level']
|
|
)
|
|
|
|
router = APIRouter(
|
|
prefix='/api/v2/gauntletrewards',
|
|
tags=['gauntletrewards']
|
|
)
|
|
|
|
|
|
class GauntletRewardModel(pydantic.BaseModel):
|
|
name: str
|
|
gauntlet_id: Optional[int] = 0
|
|
reward_id: Optional[int] = 0
|
|
win_num: Optional[int] = 0
|
|
loss_max: Optional[int] = 1
|
|
|
|
|
|
class GauntletRewardList(pydantic.BaseModel):
|
|
rewards: List[GauntletRewardModel]
|
|
|
|
|
|
@router.get('')
|
|
async def v1_gauntletreward_get(
|
|
name: Optional[str] = None, gauntlet_id: Optional[int] = None, reward_id: list = Query(default=None),
|
|
win_num: Optional[int] = None, loss_max: Optional[int] = None):
|
|
all_rewards = GauntletReward.select()
|
|
|
|
if name is not None:
|
|
all_rewards = all_rewards.where(GauntletReward.name == name)
|
|
if gauntlet_id is not None:
|
|
all_rewards = all_rewards.where(GauntletReward.gauntlet_id == gauntlet_id)
|
|
if reward_id is not None:
|
|
all_rewards = all_rewards.where(GauntletReward.reward_id << reward_id)
|
|
if win_num is not None:
|
|
all_rewards = all_rewards.where(GauntletReward.win_num == win_num)
|
|
if loss_max is not None:
|
|
all_rewards = all_rewards.where(GauntletReward.loss_max >= loss_max)
|
|
|
|
all_rewards = all_rewards.order_by(-GauntletReward.loss_max, GauntletReward.win_num)
|
|
|
|
return_val = {'count': all_rewards.count(), 'rewards': []}
|
|
for x in all_rewards:
|
|
return_val['rewards'].append(model_to_dict(x))
|
|
|
|
db.close()
|
|
return return_val
|
|
|
|
|
|
@router.get('/{gauntletreward_id}')
|
|
async def v1_gauntletreward_get_one(gauntletreward_id):
|
|
try:
|
|
this_reward = GauntletReward.get_by_id(gauntletreward_id)
|
|
except Exception:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail=f'No gauntlet reward found with id {gauntletreward_id}')
|
|
|
|
return_val = model_to_dict(this_reward)
|
|
db.close()
|
|
return return_val
|
|
|
|
|
|
@router.patch('/{gauntletreward_id}')
|
|
async def v1_gauntletreward_patch(
|
|
gauntletreward_id, name: Optional[str] = None, gauntlet_id: Optional[int] = None,
|
|
reward_id: Optional[int] = None, win_num: Optional[int] = None, loss_max: Optional[int] = None,
|
|
token: str = Depends(oauth2_scheme)):
|
|
if not valid_token(token):
|
|
logging.warning(f'Bad Token: {token}')
|
|
db.close()
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail='You are not authorized to patch gauntlet rewards. This event has been logged.'
|
|
)
|
|
|
|
this_reward = GauntletReward.get_or_none(GauntletReward.id == gauntletreward_id)
|
|
if this_reward is None:
|
|
db.close()
|
|
raise KeyError(f'Gauntlet Reward ID {gauntletreward_id} not found')
|
|
|
|
if gauntlet_id is not None:
|
|
this_reward.gauntlet_id = gauntlet_id
|
|
if reward_id is not None:
|
|
this_reward.reward_id = reward_id
|
|
if win_num is not None:
|
|
this_reward.win_num = win_num
|
|
if loss_max is not None:
|
|
this_reward.loss_max = loss_max
|
|
if name is not None:
|
|
this_reward.name = name
|
|
|
|
if this_reward.save():
|
|
r_curr = model_to_dict(this_reward)
|
|
db.close()
|
|
return r_curr
|
|
else:
|
|
db.close()
|
|
raise DatabaseError(f'Unable to patch gauntlet reward {gauntletreward_id}')
|
|
|
|
|
|
@router.post('')
|
|
async def v1_gauntletreward_post(gauntletreward: GauntletRewardList, token: str = Depends(oauth2_scheme)):
|
|
if not valid_token(token):
|
|
logging.warning(f'Bad Token: {token}')
|
|
db.close()
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail='You are not authorized to post gauntlets. This event has been logged.'
|
|
)
|
|
|
|
all_rewards = []
|
|
for x in gauntletreward.rewards:
|
|
all_rewards.append(x.dict())
|
|
|
|
with db.atomic():
|
|
for batch in chunked(all_rewards, 15):
|
|
GauntletReward.insert_many(batch).on_conflict_replace().execute()
|
|
db.close()
|
|
|
|
return f'Inserted {len(all_rewards)} gauntlet rewards'
|
|
|
|
|
|
@router.delete('/{gauntletreward_id}')
|
|
async def v1_gauntletreward_delete(gauntletreward_id):
|
|
if GauntletReward.delete_by_id(gauntletreward_id) == 1:
|
|
return f'Deleted gauntlet reward ID {gauntletreward_id}'
|
|
|
|
raise DatabaseError(f'Unable to delete gauntlet run {gauntletreward_id}')
|
|
|