Added StratGame to db

app/main.py

@@ -7,7 +7,7 @@ from fastapi import FastAPI
 from .routers_v2 import (
     current, teams, rarity, cardsets, players, packtypes, packs, cards, events, results, rewards,
     batstats, pitstats, notifications, paperdex, gamerewards, gauntletrewards, gauntletruns, battingcards,
-    battingcardratings, pitchingcards, pitchingcardratings, cardpositions, scouting, mlbplayers)
+    battingcardratings, pitchingcards, pitchingcardratings, cardpositions, scouting, mlbplayers, stratgame)
 
 app = FastAPI(
     responses={404: {'description': 'Not found'}}
@@ -41,3 +41,4 @@ app.include_router(pitchingcardratings.router)
 app.include_router(cardpositions.router)
 app.include_router(scouting.router)
 app.include_router(mlbplayers.router)
+app.include_router(stratgame.router)

app/routers_v2/stratgame.py

@@ -0,0 +1,173 @@
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from typing import Literal, Optional, List
+import logging
+import pandas as pd
+import pydantic
+from pydantic import validator, root_validator
+
+from ..db_engine import db, StratGame, model_to_dict, chunked, PitchingCard, Player, query_to_csv, Team, fn
+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/stratgames',
+    tags=['stratgames']
+)
+
+
+class GameModel(pydantic.BaseModel):
+    season: int
+    game_type: str
+    away_team_id: int
+    home_team_id: int
+    week: int = 1
+    away_score: int = 0
+    home_score: int = 0
+    away_team_value: int = None
+    home_team_value: int = None
+    away_team_ranking: int = None
+    home_team_ranking: int = None
+    ranked: bool = False
+    short_game: bool = False
+
+
+class GameList(pydantic.BaseModel):
+    games: List[GameModel]
+
+
+@router.get('')
+async def get_games(
+        season: list = Query(default=None), away_team_id: list = Query(default=None),
+        home_team_id: list = Query(default=None), team1_id: list = Query(default=None),
+        team2_id: list = Query(default=None), game_type: list = Query(default=None), ranked: Optional[bool] = None,
+        short_game: Optional[bool] = None, csv: Optional[bool] = False, short_output: bool = False):
+    all_games = StratGame.select()
+
+    if season is not None:
+        all_games = all_games.where(StratGame.season << season)
+    if away_team_id is not None:
+        all_games = all_games.where(StratGame.away_team_id << away_team_id)
+    if home_team_id is not None:
+        all_games = all_games.where(StratGame.home_team_id << home_team_id)
+    if team1_id is not None:
+        all_games = all_games.where(
+            (StratGame.away_team_id << team1_id) | (StratGame.home_team_id << team1_id)
+        )
+    if team2_id is not None:
+        all_games = all_games.where(
+            (StratGame.away_team_id << team2_id) | (StratGame.home_team_id << team2_id)
+        )
+    if game_type is not None:
+        g_list = [x.lower() for x in game_type]
+        all_games = all_games.where(fn.Lower(StratGame.game_type) << g_list)
+    if ranked is not None:
+        all_games = all_games.where(StratGame.ranked == ranked)
+    if short_game is not None:
+        all_games = all_games.where(StratGame.short_game == short_game)
+
+    if csv:
+        return_vals = [model_to_dict(x) for x in all_games]
+        for x in return_vals:
+            x['away_team_name'] = x['away_team']['lname']
+            x['home_team_name'] = x['home_team']['lname']
+            del x['away_team'], x['home_team']
+
+        db.close()
+        return Response(content=pd.DataFrame(return_vals).to_csv(index=False), media_type='text/csv')
+
+    return_val = {'count': all_games.count(), 'games': [
+        model_to_dict(x, recurse=not short_output) for x in all_games
+    ]}
+    db.close()
+    return return_val
+
+
+@router.get('/{game_id}')
+async def get_one_game(game_id: int):
+    this_game = StratGame.get_or_none(StratGame.id == game_id)
+    if not this_game:
+        db.close()
+        raise HTTPException(status_code=404, detail=f'StratGame ID {game_id} not found')
+
+    g_result = model_to_dict(this_game)
+    db.close()
+    return g_result
+
+
+@router.patch('/{game_id}')
+async def patch_game(
+        game_id: int, game_type: Optional[str] = None, away_score: Optional[int] = None,
+        home_score: Optional[int] = None, token: str = Depends(oauth2_scheme)):
+    if not valid_token(token):
+        logging.warning(f'patch_game - Bad Token: {token}')
+        raise HTTPException(status_code=401, detail='Unauthorized')
+
+    this_game = StratGame.get_or_none(StratGame.id == game_id)
+    if not this_game:
+        db.close()
+        raise HTTPException(status_code=404, detail=f'StratGame ID {game_id} not found')
+
+    if away_score is not None:
+        this_game.away_score = away_score
+    if home_score is not None:
+        this_game.home_score = home_score
+    if game_type is not None:
+        this_game.game_type = game_type
+
+    if this_game.save() == 1:
+        g_result = model_to_dict(this_game)
+        db.close()
+        return g_result
+    else:
+        db.close()
+        raise HTTPException(status_code=500, detail=f'Unable to patch game {game_id}')
+
+
+@router.post('')
+async def post_games(game_list: GameList, token: str = Depends(oauth2_scheme)):
+    if not valid_token(token):
+        logging.warning(f'post_games - Bad Token: {token}')
+        raise HTTPException(status_code=401, detail='Unauthorized')
+
+    new_games = []
+    for x in game_list.games:
+        if Team.get_or_none(Team.id == x.away_team_id) is None:
+            raise HTTPException(status_code=404, detail=f'Team ID {x.away_team_id} not found')
+        if Team.get_or_none(Team.id == x.home_team_id) is None:
+            raise HTTPException(status_code=404, detail=f'Team ID {x.home_team_id} not found')
+
+        new_games.append(x.dict())
+
+    with db.atomic():
+        for batch in chunked(new_games, 16):
+            StratGame.insert_many(batch).on_conflict_replace().execute()
+    db.close()
+
+    return f'Inserted {len(new_games)} games'
+
+
+@router.delete('/{game_id}')
+async def delete_game(game_id: int, token: str = Depends(oauth2_scheme)):
+    if not valid_token(token):
+        logging.warning(f'delete_game - Bad Token: {token}')
+        raise HTTPException(status_code=401, detail='Unauthorized')
+
+    this_game = StratGame.get_or_none(StratGame.id == game_id)
+    if not this_game:
+        db.close()
+        raise HTTPException(status_code=404, detail=f'StratGame ID {game_id} not found')
+
+    count = this_game.delete_instance()
+    db.close()
+
+    if count == 1:
+        return f'StratGame {game_id} has been deleted'
+    else:
+        raise HTTPException(status_code=500, detail=f'StratGame {game_id} could not be deleted')
+
+