paper-dynasty-database/app/routers_v2/gauntletrewards.py

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 ..db_helpers import upsert_gauntlet_rewards
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().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)

    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():
        # Use PostgreSQL-compatible upsert helper
        upsert_gauntlet_rewards(all_rewards, batch_size=15)
    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}")