change from sync to async and from flask to quart

main.py

@@ -1,4 +1,4 @@
-#!/usr/bin/env python3
+#!/usr/bin/env -S uv run --script
 
 # Welcome to
 # __________         __    __  .__                               __

server/Files.py

@@ -1,16 +1,24 @@
+import aiofiles.os
+import aiofiles
 import os
 
-def read_file(path, callback=None):
+async def read_file(path: str, callback=None):
-  if os.path.exists(path):
+  if not await aiofiles.os.path.exists(path):
-    with open(path, 'r') as f:
-      data = callback(f)
-    return data
-  else:
     return None
 
-def save_file(path, data, callback=None, *args, **kwargs):
+  async with aiofiles.open(path, "r") as f:
-  if not os.path.exists(path):
+    if callback:
-    os.makedirs(os.path.dirname(path), exist_ok=True)
+      return await callback(f)
+    return await f.read()
 
-  with open(path, 'w') as f:
+
-    callback(data, f, *args, **kwargs)
+async def save_file(path: str, data, callback=None, *args, **kwargs):
+  dir_path = os.path.dirname(path)
+  if dir_path:
+    await aiofiles.os.makedirs(dir_path, exist_ok=True)
+
+  async with aiofiles.open(path, "w") as f:
+    if callback:
+      await callback(data, f, *args, **kwargs)
+    else:
+      await f.write(data)

server/GameBoard.py

@@ -92,7 +92,7 @@ class GameBoard:
 
     self._set_turn(game_data["turn"])
 
-  def start_game(self, game_data:dict):
+  async def start_game(self, game_data:dict):
     self.init_snakes = len(game_data['board']['snakes'])
 
   def end_game(self, game_data:dict):
@@ -134,7 +134,7 @@ class GameBoard:
 
     return {"name": self.type, "is_ladder": self.is_ladder}
 
-  def save(self, store_class, **kwargs):
+  async def save(self, store_class, **kwargs):
     store = store_class(**kwargs)
-    store.save(self)
+    await store.save(self)
     del store

server/Server.py

@@ -4,8 +4,7 @@ from server.SnakeBuilder import SnakeBuilder
 
 from server.storage.StorageLoader import StorageLoader
 
-from flask import Flask, jsonify
+from quart import Quart, request, jsonify
-from flask import request
 import logging, json, os, re
 
 class Server:
@@ -25,33 +24,67 @@ class Server:
 
     self.running_games:dict[str, GameBoard] = {}
 
-    self.app = Flask("Battlesnake")
+    self.app = Quart("Battlesnake")
 
+    # info is called when you create your Battlesnake on play.battlesnake.com
+    # and controls your Battlesnake's appearance
+    # TIP: If you open your Battlesnake URL in a browser you should see this data
     @self.app.get("/")
     async def on_info():
-      return self._info()
+      snake_config = await self._read_json_config_or_create()
 
+      print("INFO Snake:", snake_config)
+      return snake_config
+
+    # start is called when your Battlesnake begins a game
     @self.app.post("/start")
     async def on_start():
-      game_state = request.get_json()
+      game_state = await request.get_json()
-      self._start(game_state)
+      await self._create_game_board(game_state)
+      print("GAME START:", game_state["game"])
       return "ok"
 
+    # move is called when your Battlesnake game is running game
     @self.app.post("/move")
     async def on_move():
-      game_state = request.get_json()
+      game_state = await request.get_json()
-      return self._move(game_state)
+      game_board = await self._get_game_board(game_state)
+      next_move = game_board.snake_neat_make_a_move()
 
+      if self.debug:
+        print("TURN:", f'{game_state["turn"]:3},', "MOVE:", f"{next_move:5}")
+
+      return {"move": next_move}
+
+    # end is called when your Battlesnake finishes a game
     @self.app.post("/end")
     async def on_end():
-      game_state = request.get_json()
+      game_state = await request.get_json()
-      self._end(game_state)
+      if self.store_game_state:
+        game_board = await self._get_game_board(game_state, end=True)
+        #if not game_board.get_winner() == "me" and not game_board.get_turn() <= self.store_game_when_win_and_moves_are_bigger_as:
+        if self.check_tls_security:
+          await game_board.save(
+            StorageLoader.build(self.storage_type),
+            file_path=os.path.join(self.data_path, 'data'),
+            database=os.getenv("EDGEDB_DATABASE", None),
+            tls_security=None
+          )
+        else:
+          await game_board.save(
+            StorageLoader.build(self.storage_type),
+            file_path=os.path.join(self.data_path, 'data'),
+            database=os.getenv("EDGEDB_DATABASE", None),
+          )
+
+      print("GAME ENDED: Winner is", [ x["name"] for x in game_state["board"]['snakes']])
+      self._delete_game_board(game_state)
       return "ok"
 
     @self.app.after_request
     async def identify_server(response):
       response.headers.set(
-        "server", "battlesnake/github/starter-snake-python"
+        "server", "battlesnake/gitea/snake-python"
       )
       return response
 
@@ -66,20 +99,20 @@ class Server:
     print(f"\nRunning Battlesnake at http://{host}:{port} with the {' '.join(re.findall('[A-Z][^A-Z]*', self.snake_type))}")
     self.app.run(host=host, port=port, debug=debug)
 
-  def _read_json_config_or_create(self):
+  async def _read_json_config_or_create(self):
-    snake_config = read_file(self.config_file, json.load)
+    snake_config = await read_file(self.config_file, json.load)
     if not snake_config:
-      snake_config = self._override_snake_config_with_environment_variables(self.default_snake_config)
+      return await self._override_snake_config_with_environment_variables(self.default_snake_config)
+    return await self._override_snake_config_with_environment_variables(snake_config)
 
-    return self._override_snake_config_with_environment_variables(snake_config)
+  async def _override_snake_config_with_environment_variables(self, config: dict[str, str]) -> dict[str, str]:
-
+    for key in ("author", "color", "head", "tail"):
-  def _override_snake_config_with_environment_variables(self, config:dict[str]):
+      value = os.environ.get(f"SNAKE_{key.upper()}")
-    for key in ["author", "color", "head", "tail"]:
+      if value is not None:
-      if os.environ.get(f"SNAKE_{key.upper()}", None):
+        config[key] = value
-        config[key.lower()] = os.environ.get(f"SNAKE_{key.upper()}")
     return config
 
-  def _create_game_board(self, game_state:dict):
+  async def _create_game_board(self, game_state:dict):
     new_game_board = GameBoard(
       game_id=game_state["game"]["id"],
       width=game_state['board']['width'],
@@ -89,7 +122,7 @@ class Server:
       map=game_state['game']['map'],
       snake_class=SnakeBuilder.build(self.snake_type)
     )
-    new_game_board.start_game(game_state)
+    await new_game_board.start_game(game_state)
 
     self.running_games[game_state["game"]["id"]] = new_game_board
     return new_game_board
@@ -97,11 +130,11 @@ class Server:
   def _delete_game_board(self, game_state):
     del self.running_games[game_state["game"]["id"]]
 
-  def _get_game_board(self, game_state:str, end:bool=False):
+  async def _get_game_board(self, game_state:str, end:bool=False):
     try:
       game_board = self.running_games[game_state["game"]["id"]]
     except KeyError:
-      game_board = self._create_game_board(game_state)
+      game_board = await self._create_game_board(game_state)
 
     game_board.read_game_data(game_state)
     if end:
@@ -112,52 +145,6 @@ class Server:
   def enable_store_game_state(self):
     self.store_game_state = True
 
-  # info is called when you create your Battlesnake on play.battlesnake.com
-  # and controls your Battlesnake's appearance
-  # TIP: If you open your Battlesnake URL in a browser you should see this data
-  def _info(self) -> dict:
-    snake_config = self._read_json_config_or_create()
-
-    print("INFO Snake:", snake_config)
-    return snake_config
-
-  # start is called when your Battlesnake begins a game
-  def _start(self, game_state:dict):
-    self._create_game_board(game_state)
-    print("GAME START:", game_state["game"])
-
-  # move is called when your Battlesnake game is running game
-  def _move(self, game_state:dict) -> dict:
-    game_board = self._get_game_board(game_state)
-    next_move = game_board.snake_neat_make_a_move()
-
-    if self.debug:
-      print("TURN:", f'{game_state["turn"]:3},', "MOVE:", f"{next_move:5}")
-
-    return {"move": next_move}
-
-  # end is called when your Battlesnake finishes a game
-  def _end(self, game_state:dict):
-    if self.store_game_state:
-      game_board = self._get_game_board(game_state, end=True)
-      #if not game_board.get_winner() == "me" and not game_board.get_turn() <= self.store_game_when_win_and_moves_are_bigger_as:
-      if self.check_tls_security:
-        game_board.save(
-          StorageLoader.build(self.storage_type),
-          file_path=os.path.join(self.data_path, 'data'),
-          database=os.getenv("EDGEDB_DATABASE", None),
-          tls_security=None
-        )
-      else:
-        game_board.save(
-          StorageLoader.build(self.storage_type),
-          file_path=os.path.join(self.data_path, 'data'),
-          database=os.getenv("EDGEDB_DATABASE", None),
-        )
-
-    print("GAME ENDED: Winner is", [ x["name"] for x in game_state["board"]['snakes']])
-    self._delete_game_board(game_state)
-
   def _cleanup_database(self):
     storage = StorageLoader.build(self.storage_type)()
     return storage.cleanup()

server/storage/EdgeDB.py

@@ -45,7 +45,7 @@ class EdgeDB:
 
     return data
 
-  def insert(self, game_board:GameBoard):
+  async def insert(self, game_board:GameBoard):
     game_type = game_board.get_type_of_game()
 
     self.run_query_with_reconnection(
@@ -104,8 +104,8 @@ class EdgeDB:
       snake_type=game_board.snake_class.__class__.__name__,
     )
 
-  def save(self, game_board:GameBoard):
+  async def save(self, game_board:GameBoard):
-    self.insert(game_board)
+    await self.insert(game_board)
 
   def __del__(self):
     self.client.close()

server/storage/LocalStorage.py

@@ -28,7 +28,7 @@ class LocalStorage:
 
     return os.path.join(storage_folder, file_name)
 
-  def save(self, game_board:GameBoard):
+  async def save(self, game_board:GameBoard):
     game_type = game_board.get_type_of_game()
     save_file_path = self._get_correct_folder_for_save_file(
       game_board,
@@ -38,7 +38,7 @@ class LocalStorage:
       True if game_board.winner_snake_names and "me" in game_board.winner_snake_names else False
     )
 
-    save_file(save_file_path, {
+    await save_file(save_file_path, {
       "winner": game_board.winner_snake_names,
       "game": {
         "url": game_board.url,