134 lines
3.3 KiB
Python
134 lines
3.3 KiB
Python
import datetime
|
|
import typing
|
|
|
|
import tortoise.timezone
|
|
from fastapi import APIRouter, FastAPI
|
|
from pydantic import BaseModel, Field, validator
|
|
from tortoise.contrib.fastapi import HTTPNotFoundError, register_tortoise
|
|
from tortoise.expressions import Q
|
|
|
|
import database
|
|
import models
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
admin_router = APIRouter(prefix="/admin")
|
|
|
|
|
|
class Key(BaseModel):
|
|
key: str
|
|
expires_at: datetime.datetime | None
|
|
|
|
class Config:
|
|
orm_mode = True
|
|
|
|
|
|
class Game(BaseModel):
|
|
title: str
|
|
auto_approve: bool
|
|
keys: list[Key]
|
|
|
|
@validator("keys", pre=True)
|
|
def to_list(cls, v):
|
|
return list(v)
|
|
|
|
class Config:
|
|
orm_mode = True
|
|
|
|
|
|
class GameIn(BaseModel):
|
|
title: str
|
|
auto_approve: bool
|
|
|
|
|
|
class GameUpdate(BaseModel):
|
|
auto_approve: bool | None = None
|
|
|
|
|
|
@admin_router.post("/games", response_model=Game)
|
|
async def create_game(game: GameIn):
|
|
return await models.Game.create(**game.dict())
|
|
|
|
|
|
@admin_router.put("/games/{game_title}", response_model=Game)
|
|
async def update_game(game_title: str, update: GameUpdate):
|
|
game = await models.Game.get(title=game_title).prefetch_related("keys")
|
|
return await game.update_from_dict(update.dict(exclude_unset=True)).save()
|
|
|
|
|
|
@admin_router.get("/games", response_model=list[Game])
|
|
async def get_games():
|
|
games = await models.Game.all().prefetch_related("keys")
|
|
return games
|
|
|
|
|
|
@admin_router.post("/games/{game_title}/keys")
|
|
async def create_key(game_title: str, expires_at: datetime.datetime | None = None):
|
|
game = await models.Game.get(title=game_title)
|
|
new_key = await models.GameKey.create(game=game, key="asdf", expires_at=expires_at)
|
|
return new_key.key
|
|
|
|
|
|
class AdminScore(BaseModel):
|
|
id: int
|
|
username: str
|
|
points: int
|
|
approved: bool
|
|
created_at: datetime.datetime
|
|
|
|
|
|
class AdminScoreIn(BaseModel):
|
|
approved: bool
|
|
|
|
|
|
@admin_router.post("/scores/{game_title}", response_model=list[AdminScore])
|
|
async def get_scores_for_game(game_title: str):
|
|
game = await models.Game.get(title=game_title)
|
|
return await models.Score.filter(game=game).order_by("-created_at")
|
|
|
|
|
|
@admin_router.put("/scores/{id}", response_model=AdminScore)
|
|
async def update_score(id: int, update: AdminScoreIn):
|
|
score = await models.Score.get(id=id)
|
|
return await score.update_from_dict(update.dict()).save()
|
|
|
|
|
|
app.include_router(admin_router)
|
|
|
|
|
|
class ScoreIn(BaseModel):
|
|
username: str = Field(max_length=20)
|
|
points: int = Field(ge=0)
|
|
key: str
|
|
|
|
|
|
class Score(BaseModel):
|
|
username: str
|
|
points: str
|
|
|
|
class Config:
|
|
orm_mode = True
|
|
|
|
|
|
@app.post("/scores", response_model=Score)
|
|
async def post_score(score: ScoreIn):
|
|
key_valid = Q(keys__expires_at=None) | Q(
|
|
keys__expires_at__gt=tortoise.timezone.now())
|
|
game = await models.Game.filter(
|
|
key_valid & Q(keys__key=score.key)
|
|
).get()
|
|
return await models.Score.create(
|
|
username=score.username,
|
|
points=score.points,
|
|
game=game,
|
|
approved=game.auto_approve)
|
|
|
|
register_tortoise(app, database.CONFIG, add_exception_handlers=True)
|
|
|
|
|
|
@app.get("/scores/{game_title}", response_model=list[Score])
|
|
async def get_scores(game_title: str):
|
|
game = await models.Game.filter(title=game_title).get()
|
|
return await models.Score.filter(game=game,
|
|
approved=True).order_by("-points")
|