Switch to FastAPI

Dockerfile

@@ -1,16 +1,14 @@
 FROM python:3.11.4-slim
-# FROM continuumio/conda-ci-linux-64-python3.8
-# FROM continuumio/miniconda3
 
-# Install app
-COPY . /app
 WORKDIR /app
+COPY . .
 
-# Install dependencies
 RUN pip install --upgrade pip && pip install -r requirements.txt
-# RUN conda env create -f environment.yaml
+
-# ENV CONDA_DEFAULT_ENV battlesnake
+ENV HOST 0.0.0.0
+ENV PORT 8000
+ENV ROOT_PATH /
+ENV SNAKE starter_snake
 
 # Run Battlesnake
-ENTRYPOINT [ "python", "server.py" ]
+ENTRYPOINT uvicorn src.server:app --host $HOST --port $PORT --root-path $ROOT_PATH
-CMD [ "starter_snake" ]

requirements.txt

@@ -1,10 +1,33 @@
+annotated-types==0.5.0
+anyio==3.7.1
 blinker==1.6.2
 click==8.1.7
+fastapi==0.103.1
 Flask==2.3.3
+h11==0.14.0
+h2==4.1.0
+hpack==4.0.0
+httptools==0.6.0
+hyperframe==6.0.1
+idna==3.4
 itsdangerous==2.1.2
 Jinja2==3.1.2
 MarkupSafe==2.1.3
-mypy==1.5.1
+marshmallow==3.20.1
-mypy-extensions==1.0.0
+packaging==23.1
+priority==2.0.0
+pydantic==2.3.0
+pydantic-extra-types==2.1.0
+pydantic_core==2.6.3
+python-dotenv==1.0.0
+PyYAML==6.0.1
+sniffio==1.3.0
+starlette==0.27.0
+typing-inspect==0.9.0
 typing_extensions==4.7.1
+uvicorn==0.23.2
+uvloop==0.17.0
+watchfiles==0.20.0
+websockets==11.0.3
 Werkzeug==2.3.7
+wsproto==1.2.0

server.py

@@ -1,63 +0,0 @@
-import argparse
-import logging
-import os
-import typing
-
-from flask import Flask
-from flask import request
-
-from snakes.snake import Snake
-
-
-def run_server(snake: Snake, host: str, port: int):
-    path = snake.name
-    app = Flask("Battlesnake")
-
-    @app.get(f"/{path}/")
-    def on_info():
-        return snake.info()
-
-    @app.post(f"/{path}/start")
-    def on_start():
-        game_state = request.get_json()
-        snake.start(game_state)
-        return "ok"
-
-    @app.post(f"/{path}/move")
-    def on_move():
-        game_state = request.get_json()
-        return snake.move(game_state)
-
-    @app.post(f"/{path}/end")
-    def on_end():
-        game_state = request.get_json()
-        snake.end(game_state)
-        return "ok"
-
-    @app.after_request
-    def identify_server(response):
-        response.headers.set(
-            "server", "shillerben/gitea/starter-snake-python"
-        )
-        return response
-
-    logging.getLogger("werkzeug").setLevel(logging.ERROR)
-
-    print(f"\nRunning Battlesnake at http://{host}:{port}/{path}")
-    app.run(host=host, port=port)
-
-
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser()
-    parser.add_argument("-b", "--bind", default="0.0.0.0", help="host ip to listen on")
-    parser.add_argument("-p", "--port", type=int, default=8000, help="host port to listen on")
-    parser.add_argument("snake", help="snake to run")
-    args = parser.parse_args()
-
-    host = args.bind
-    port = args.port
-    snake_name = args.snake
-
-    snake = Snake.SNAKES[snake_name]()
-
-    run_server(snake, host, port)

src/api.py

@@ -0,0 +1,166 @@
+from typing import Union, List
+from enum import StrEnum, auto
+from pydantic import BaseModel, PositiveInt, NonNegativeInt, conint, constr
+
+
+'''
+Enums
+'''
+
+class GameMode(StrEnum):
+    EMPTY_STR = ""
+    STANDARD = auto()
+    SOLO = auto()
+    ROYALE = auto()
+    SQUAD = auto()
+    CONSTRICTOR = auto()
+    WRAPPED = auto()
+
+class Source(StrEnum):
+    TOURNAMENT = auto()
+    LEAGUE = auto()
+    ARENA = auto()
+    CHALLENGE = auto()
+    CUSTOM = auto()
+
+class GameMap(StrEnum):
+    EMPTY_STR = ""
+    STANDARD = auto()
+    EMPTY = auto()
+    ARCADE_MAZE = auto()
+    ROYALE = auto()
+    SOLO_MAZE = auto()
+    HZ_INNER_WALL = auto()
+    HZ_RINGS = auto()
+    HZ_COLUMNS = auto()
+    HZ_RIVERS_BRIDGES = auto()
+    HZ_SPIRAL = auto()
+    HZ_SCATTER = auto()
+    HZ_GROW_BOX = auto()
+    HZ_EXPAND_BOX = auto()
+    HZ_EXPAND_SCATTER = auto()
+
+class GameMove(StrEnum):
+    UP = auto()
+    DOWN = auto()
+    LEFT = auto()
+    RIGHT = auto()
+
+
+'''
+Models
+'''
+
+class RoyaleRulesetSettings(BaseModel):
+    shrinkEveryNTurns: NonNegativeInt
+
+class SquadRulesetSettings(BaseModel):
+    allowBodyCollisions: bool
+    sharedElimination: bool
+    sharedHealth: bool
+    sharedLength: bool
+
+class RulesetSettings(BaseModel):
+    foodSpawnChance: conint(ge=0, le=100, strict=True)
+    minimumFood: NonNegativeInt
+    hazardDamagePerTurn: NonNegativeInt
+    royale: RoyaleRulesetSettings
+    squad: SquadRulesetSettings
+
+class Ruleset(BaseModel):
+    name: GameMode
+    version: str
+    settings: RulesetSettings
+
+
+class Coordinate(BaseModel):
+    x: NonNegativeInt
+    y: NonNegativeInt
+
+
+class BattlesnakeCustomizations(BaseModel):
+    color: str
+    head: str
+    tail: str
+
+
+class Battlesnake(BaseModel):
+    id: str
+    name: str
+    health: conint(ge=0, le=100, strict=True)
+    body: List[Coordinate]
+    latency: str  # previous latency in milliseconds, str(int)
+    head: Coordinate
+    length: PositiveInt
+    shout: constr(max_length=256)
+    squad: str
+    customizations: BattlesnakeCustomizations
+
+
+
+class Board(BaseModel):
+    height: PositiveInt
+    width: PositiveInt
+    food: List[Coordinate]
+    hazards: List[Coordinate]
+    snakes: List[Battlesnake]
+
+
+class Game(BaseModel):
+    id: str
+    ruleset: Ruleset
+    map: GameMap
+    timeout: NonNegativeInt  # milliseconds
+    source: Source
+
+
+'''
+GET /
+'''
+class GetResponse(BaseModel):
+    apiversion: str
+    author: Union[str, None] = None
+    color: Union[str, None] = None
+    head: Union[str, None] = None
+    tail: Union[str, None] = None
+    version: Union[str, None] = None
+
+
+'''
+POST /start
+'''
+class StartRequest(BaseModel):
+    game: Game
+    turn: NonNegativeInt
+    board: Board
+    you: Battlesnake
+
+class StartResponse(BaseModel):
+    pass
+
+
+'''
+POST /move
+'''
+class MoveRequest(BaseModel):
+    game: Game
+    turn: NonNegativeInt
+    board: Board
+    you: Battlesnake
+
+class MoveResponse(BaseModel):
+    move: GameMove
+    shout: constr(max_length=256) = ""
+
+
+'''
+POST /end
+'''
+class EndRequest(BaseModel):
+    game: Game
+    turn: NonNegativeInt
+    board: Board
+    you: Battlesnake
+
+class EndResponse(BaseModel):
+    pass

src/server.py

@@ -0,0 +1,45 @@
+import logging
+import os
+
+from fastapi import FastAPI, Request
+
+from .snakes.snake import Snake
+from . import api
+
+
+snake_name = os.getenv("SNAKE")
+
+if snake_name is None:
+    print("SNAKE env variable is not set!")
+    exit(1)
+
+snake = Snake.SNAKES[snake_name]()
+
+app = FastAPI(title=snake.name,
+              summary=f"API for {snake.name}")
+
+@app.get("/")
+def on_info() -> api.GetResponse:
+    print("INFO")
+    return snake.info()
+
+@app.post("/start")
+def on_start(request: api.StartRequest) -> api.StartResponse:
+    return snake.start(request)
+
+@app.post("/move")
+def on_move(request: api.MoveRequest) -> api.MoveResponse:
+    return snake.move(request)
+
+@app.post("/end")
+def on_end(request: api.EndRequest) -> api.EndResponse:
+    return snake.end(request)
+
+
+@app.middleware("http")
+async def identify_server(request: Request, call_next):
+    response = await call_next(request)
+    response.headers["server"] = "shillerben/gitea/starter-snake-python"
+    return response
+
+logging.getLogger("werkzeug").setLevel(logging.ERROR)

src/snakes/snake.py

@@ -1,5 +1,7 @@
 import typing
 
+from .. import api
+
 
 class Snake:
 
@@ -9,7 +11,7 @@ class Snake:
     def __init_subclass__(cls) -> None:
         cls.SNAKES[cls.name] = cls
 
-    def info(self) -> typing.Dict:
+    def info(self) -> api.GetResponse:
         '''info is called when you create your Battlesnake on play.battlesnake.com
         and controls your Battlesnake's appearance
 
@@ -17,15 +19,15 @@ class Snake:
         '''
         raise NotImplementedError()
 
-    def start(self, game_state: typing.Dict) -> None:
+    def start(self, game_state: api.StartRequest) -> api.StartResponse:
         '''start is called when your Battlesnake begins a game'''
         raise NotImplementedError()
 
-    def end(self, game_state: typing.Dict) -> None:
+    def end(self, game_state: api.EndRequest) -> api.EndResponse:
         '''end is called when your Battlesnake finishes a game'''
         raise NotImplementedError()
 
-    def move(self, game_state: typing.Dict) -> typing.Dict:
+    def move(self, game_state: api.MoveRequest) -> api.MoveResponse:
         '''move is called on every turn and returns your next move
         Valid moves are "up", "down", "left", or "right"
         See https://docs.battlesnake.com/api/example-move for available data

src/snakes/starter_snake.py

@@ -1,6 +1,7 @@
 
 import random
-import typing
+
+from .. import api
 
 from .snake import Snake
 
@@ -10,40 +11,41 @@ class StarterSnake(Snake):
 
     name = "starter_snake"
 
-    def info(self) -> typing.Dict:
+    def info(self) -> api.GetResponse:
         print("INFO")
+        return api.GetResponse(
+            apiversion="1",
+            author="shillerben",
+            color="#BBDEFB",
+            head="default",
+            tail="default",
+        )
 
-        return {
+    def start(self, game_state: api.StartRequest) -> api.StartResponse:
-            "apiversion": "1",
-            "author": "",  # TODO: Your Battlesnake Username
-            "color": "#888888",  # TODO: Choose color
-            "head": "default",  # TODO: Choose head
-            "tail": "default",  # TODO: Choose tail
-        }
-
-    def start(self, game_state: typing.Dict):
         print("GAME START")
+        return api.StartResponse()
 
-    def end(self, game_state: typing.Dict):
+    def end(self, game_state: api.EndRequest) -> api.EndResponse:
-        print("GAME OVER\n")
+        print("GAME OVER")
+        return api.EndResponse()
 
-    def move(self, game_state: typing.Dict) -> typing.Dict:
+    def move(self, game_state: api.MoveRequest) -> api.MoveResponse:
         is_move_safe = {"up": True, "down": True, "left": True, "right": True}
 
         # We've included code to prevent your Battlesnake from moving backwards
-        my_head = game_state["you"]["body"][0]  # Coordinates of your head
+        my_head = game_state.you.body[0]  # Coordinates of your head
-        my_neck = game_state["you"]["body"][1]  # Coordinates of your "neck"
+        my_neck = game_state.you.body[1]  # Coordinates of your "neck"
 
-        if my_neck["x"] < my_head["x"]:  # Neck is left of head, don't move left
+        if my_neck.x < my_head.x:  # Neck is left of head, don't move left
             is_move_safe["left"] = False
 
-        elif my_neck["x"] > my_head["x"]:  # Neck is right of head, don't move right
+        elif my_neck.x > my_head.x:  # Neck is right of head, don't move right
             is_move_safe["right"] = False
 
-        elif my_neck["y"] < my_head["y"]:  # Neck is below head, don't move down
+        elif my_neck.y < my_head.y:  # Neck is below head, don't move down
             is_move_safe["down"] = False
 
-        elif my_neck["y"] > my_head["y"]:  # Neck is above head, don't move up
+        elif my_neck.y > my_head.y:  # Neck is above head, don't move up
             is_move_safe["up"] = False
 
         # TODO: Step 1 - Prevent your Battlesnake from moving out of bounds
@@ -63,8 +65,8 @@ class StarterSnake(Snake):
                 safe_moves.append(move)
 
         if len(safe_moves) == 0:
-            print(f"MOVE {game_state['turn']}: No safe moves detected! Moving down")
+            print(f"MOVE {game_state.turn}: No safe moves detected! Moving down")
-            return {"move": "down"}
+            return api.MoveResponse(move=api.GameMove.UP)
 
         # Choose a random move from the safe ones
         next_move = random.choice(safe_moves)
@@ -72,5 +74,5 @@ class StarterSnake(Snake):
         # TODO: Step 4 - Move towards food instead of random, to regain health and survive longer
         # food = game_state['board']['food']
 
-        print(f"MOVE {game_state['turn']}: {next_move}")
+        print(f"MOVE {game_state.turn}: {next_move}")
-        return {"move": next_move}
+        return api.MoveResponse(move=api.GameMove(next_move))