141 lines
4.4 KiB
Python
141 lines
4.4 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, DatabaseError, DoesNotExist
|
|
from ..db_helpers import upsert_gauntlet_rewards
|
|
from ..dependencies import oauth2_scheme, valid_token
|
|
|
|
|
|
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,
|
|
limit: int = 100,
|
|
):
|
|
all_rewards = GauntletReward.select().order_by(GauntletReward.id)
|
|
|
|
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)
|
|
|
|
limit = max(0, min(limit, 500))
|
|
all_rewards = all_rewards.limit(limit)
|
|
|
|
return_val = {"count": all_rewards.count(), "rewards": []}
|
|
for x in all_rewards:
|
|
return_val["rewards"].append(model_to_dict(x))
|
|
|
|
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 DoesNotExist:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"No gauntlet reward found with id {gauntletreward_id}",
|
|
)
|
|
|
|
return_val = model_to_dict(this_reward)
|
|
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("Bad Token: [REDACTED]")
|
|
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:
|
|
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)
|
|
return r_curr
|
|
else:
|
|
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("Bad Token: [REDACTED]")
|
|
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():
|
|
# Use PostgreSQL-compatible upsert helper
|
|
upsert_gauntlet_rewards(all_rewards, batch_size=15)
|
|
|
|
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}")
|