tank-tactics/server/app.py

import os
import datetime
import uuid
from enum import Enum
from typing import Optional

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import aiosqlite

DB_FILE = os.environ.get("DB_LOCATION", "tank.db")
LAST_UPDATE = datetime.datetime.now()


app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=False,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.on_event("startup")
async def startup():
    async with aiosqlite.connect(DB_FILE) as db:
        await db.execute('CREATE TABLE IF NOT EXISTS games (\
            id TEXT NOT NULL PRIMARY KEY, \
            name TEXT NOT NULL, \
            state TEXT CHECK( state IN ("LOBBY", "PLAYING", "FINISHED")) NOT NULL DEFAULT "LOBBY", \
            width INTEGER NOT NULL, \
            height INTEGER NOT NULL, \
            distribute TEXT DEFAULT NULL, \
            players TEXT NOT NULL DEFAULT "" \
        );')
        await db.commit()


class CreateGame(BaseModel):
    name: str
    width: int
    height: int


class GameInfo(BaseModel):
    id: str
    name: str


class GamesList(BaseModel):
    games: list[GameInfo]


class GameState(str, Enum):
    lobby = "LOBBY"
    playing = "PLAYING"
    finished = "FINISHED"


class Game(BaseModel):
    name: str
    state: GameState
    width: int
    height: int
    distribute: Optional[str]
    players: str


@app.get("/games")
async def games() -> GamesList:
    async with aiosqlite.connect(DB_FILE) as db:
        async with db.execute("SELECT id, name FROM games;") as cursor:
            results = await cursor.fetchall()
    return GamesList(games=[GameInfo(id=r[0], name=r[1]) for r in results])


@app.put("/game")
async def create_game(create: CreateGame) -> GameInfo:
    game_id = uuid.uuid4().hex
    async with aiosqlite.connect(DB_FILE) as db:
        await db.execute("INSERT INTO games \
            (id, name, width, height) VALUES \
            (:id, :name, :width, :height)\
        ;", {
            "id": game_id,
            "name": create.name,
            "width": create.width,
            "height": create.height,
        })
        await db.commit()
    return GameInfo(id=game_id, name=create.name)


@app.get("/game/{game_id}")
async def get_game(game_id: str) -> Game:
    async with aiosqlite.connect(DB_FILE) as db:
        async with db.execute("SELECT name, state, width, height, distribute, players FROM games WHERE id = ?", (game_id,)) as cursor:
            result = await cursor.fetchone()
    if result is None:
        raise HTTPException(status_code=404)
    return Game(
        name=result[0],
        state=result[1],
        width=result[2],
        height=result[3],
        distribute=result[4],
        players=result[5],
    )