156 lines
4.7 KiB
Python
156 lines
4.7 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Response
|
|
from typing import Optional
|
|
import logging
|
|
import pydantic
|
|
from pandas import DataFrame
|
|
|
|
from ..db_engine import db, Award, model_to_dict
|
|
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/awards',
|
|
tags=['awards']
|
|
)
|
|
|
|
|
|
class AwardModel(pydantic.BaseModel):
|
|
name: str
|
|
season: int
|
|
timing: str = 'In-Season'
|
|
card_id: Optional[int] = None
|
|
team_id: Optional[int] = None
|
|
image: Optional[str] = None
|
|
|
|
|
|
@app.get('/api/v1/awards')
|
|
async def get_awards(
|
|
name: Optional[str] = None, season: Optional[int] = None, timing: Optional[str] = None,
|
|
card_id: Optional[int] = None, team_id: Optional[int] = None, image: Optional[str] = None,
|
|
csv: Optional[bool] = None):
|
|
all_awards = Award.select()
|
|
|
|
if all_awards.count() == 0:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail=f'There are no awards to filter')
|
|
|
|
if name is not None:
|
|
all_awards = all_awards.where(Award.name == name)
|
|
if season is not None:
|
|
all_awards = all_awards.where(Award.season == season)
|
|
if timing is not None:
|
|
all_awards = all_awards.where(Award.timing == timing)
|
|
if card_id is not None:
|
|
all_awards = all_awards.where(Award.card_id == card_id)
|
|
if team_id is not None:
|
|
all_awards = all_awards.where(Award.team_id == team_id)
|
|
if image is not None:
|
|
all_awards = all_awards.where(Award.image == image)
|
|
|
|
if csv:
|
|
data_list = [['id', 'name', 'season', 'timing', 'card', 'team', 'image']]
|
|
for line in all_awards:
|
|
data_list.append([
|
|
line.id, line.name, line.season, line.timing, line.card, line.team, line.image
|
|
])
|
|
return_val = DataFrame(data_list).to_csv(header=False, index=False)
|
|
|
|
db.close()
|
|
return Response(content=return_val, media_type='text/csv')
|
|
|
|
else:
|
|
return_val = {'count': all_awards.count(), 'awards': []}
|
|
for x in all_awards:
|
|
return_val['awards'].append(model_to_dict(x))
|
|
|
|
db.close()
|
|
return return_val
|
|
|
|
|
|
@app.get('/api/v1/awards/{award_id}')
|
|
async def get_one_award(award_id, csv: Optional[bool] = None):
|
|
try:
|
|
this_award = Award.get_by_id(award_id)
|
|
except Exception:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail=f'No award found with id {award_id}')
|
|
|
|
if csv:
|
|
data_list = [
|
|
['id', 'name', 'season', 'timing', 'card', 'team', 'image'],
|
|
[this_award.id, this_award.name, this_award.season, this_award.timing, this_award.card,
|
|
this_award.team, this_award.image]
|
|
]
|
|
return_val = DataFrame(data_list).to_csv(header=False, index=False)
|
|
|
|
db.close()
|
|
return Response(content=return_val, media_type='text/csv')
|
|
|
|
else:
|
|
return_val = model_to_dict(this_award)
|
|
db.close()
|
|
return return_val
|
|
|
|
|
|
@app.post('/api/v1/awards')
|
|
async def post_awards(award: AwardModel, 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 awards. This event has been logged.'
|
|
)
|
|
|
|
this_award = Award(
|
|
name=award.name,
|
|
season=award.season,
|
|
timing=award.season,
|
|
card_id=award.card_id,
|
|
team_id=award.team_id,
|
|
image=award.image
|
|
)
|
|
|
|
saved = this_award.save()
|
|
if saved == 1:
|
|
return_val = model_to_dict(this_award)
|
|
db.close()
|
|
return return_val
|
|
else:
|
|
db.close()
|
|
raise HTTPException(
|
|
status_code=418,
|
|
detail='Well slap my ass and call me a teapot; I could not save that roster'
|
|
)
|
|
|
|
|
|
@app.delete('/api/v1/awards/{award_id}')
|
|
async def delete_award(award_id, 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 delete awards. This event has been logged.'
|
|
)
|
|
try:
|
|
this_award = Award.get_by_id(award_id)
|
|
except Exception:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail=f'No award found with id {award_id}')
|
|
|
|
count = this_award.delete_instance()
|
|
db.close()
|
|
|
|
if count == 1:
|
|
raise HTTPException(status_code=200, detail=f'Award {award_id} has been deleted')
|
|
else:
|
|
raise HTTPException(status_code=500, detail=f'Award {award_id} was not deleted')
|
|
|
|
|