move snake builder into game_runtime.py to not pass it around very where

server/Server.py

@@ -31,7 +31,6 @@ from server.services import (
   DashboardQueryService,
 )
 
-
 class Server:
   default_snake_config = {
     'apiversion': '1',
@@ -97,7 +96,6 @@ class Server:
     self._startup_worker_metrics_cleared = False
 
     self.logger = build_logger('Battlesnake', debug_env_var='DEBUG_SERVER')
-    self.snake_builder = SnakeBuilder
     self.snake_version = self._get_snake_version()
     self.gameplay_database = None
     if gameplay_db_enabled:
@@ -214,20 +212,9 @@ class Server:
 
   def _get_snake_version(self) -> str:
     configured_version = SnakeBuilder.get_version(self.snake_type)
-    if configured_version:
+    if configured_version is None:
-      return configured_version
-
-    try:
-      snake = SnakeBuilder.build(self.snake_type)
-    except Exception:
       return self.default_snake_config['version']
-
+    return str(configured_version)
-    version = getattr(snake, 'version', None)
-    if version is None:
-      version = getattr(snake, 'VERSION', None)
-    if not version:
-      return self.default_snake_config['version']
-    return str(version)
 
   def _get_stale_game_timeout_sec(self) -> int:
     return max(30, env_int('SNAKE_STUCK_GAME_TIMEOUT_SEC', 180))
@@ -236,7 +223,7 @@ class Server:
     self.store_game_state = True
 
   def _cleanup_database(self):
-    storage = StorageLoader.build(self.storage_type)()
+    storage = StorageLoader.build(self.storage_type)
     return storage.cleanup()
 
   async def _on_dashboard_games_update_notice(self, trigger:str) -> None:

server/blueprints/battlesnake.py

@@ -25,7 +25,7 @@ def create_battlesnake_blueprint(server:'Server') -> Blueprint:
     server.metrics_collector.record_http_request('start')
     await server.game_runtime.prune_stale_games()
     game_state = await request.get_json()
-    await server.game_runtime.create_game_board(game_state, snake_builder=server.snake_builder)
+    await server.game_runtime.create_game_board(game_state)
     await server.gameplay_tracking.record_gameplay_start(game_state)
     await await_log(server.logger.info(f'GAME START: {game_state['game']}'))
     return 'ok'
@@ -35,7 +35,7 @@ def create_battlesnake_blueprint(server:'Server') -> Blueprint:
     server.metrics_collector.record_http_request('move')
     game_state = await request.get_json()
     move_started = time.perf_counter()
-    game_board = cast(GameBoard, await server.game_runtime.get_game_board(game_state, snake_builder=server.snake_builder))
+    game_board = cast(GameBoard, await server.game_runtime.get_game_board(game_state))
     next_move = game_board.snake_neat_make_a_move()
     await server.game_runtime.persist_game_board(game_state['game']['id'], game_board)
     await server.gameplay_tracking.record_gameplay_turn(game_state, next_move, game_board)
@@ -53,7 +53,7 @@ def create_battlesnake_blueprint(server:'Server') -> Blueprint:
     await server.game_runtime.prune_stale_games()
     game_state = await request.get_json()
     if server.store_game_state:
-      game_board = cast(GameBoard, await server.game_runtime.get_game_board(game_state, snake_builder=server.snake_builder, end=True))
+      game_board = cast(GameBoard, await server.game_runtime.get_game_board(game_state, end=True))
       if server.check_tls_security:
         await game_board.save(
           StorageLoader.build(server.storage_type),

server/services/game_runtime.py

@@ -1,14 +1,20 @@
-from typing import cast
+from typing import Protocol, cast
 import time
 
 from server.metrics import MetricsCollector
 from server.GameBoard import GameBoard
 
-from server.storage import StorageLoader
 from snakes import SnakeBuilder
 
+class GameStateStoreLike(Protocol):
+  async def save(self, game_id: str, game_board: GameBoard) -> None: ...
+
+  async def load(self, game_id: str) -> object | None: ...
+
+  async def delete(self, game_id: str) -> None: ...
+
 class GameRuntimeService:
-  def __init__(self, game_state_store:StorageLoader, snake_type:str, game_state_local_cache:bool, stale_game_timeout_sec:int):
+  def __init__(self, game_state_store:GameStateStoreLike, snake_type:str, game_state_local_cache:bool, stale_game_timeout_sec:int):
     self.game_state_store = game_state_store
     self.snake_type = snake_type
     self.game_state_local_cache = game_state_local_cache
@@ -22,7 +28,7 @@ class GameRuntimeService:
   def attach_metrics_collector(self, metrics_collector:MetricsCollector) -> None:
     self.metrics_collector = metrics_collector
 
-  async def create_game_board(self, game_state:dict, snake_builder:SnakeBuilder) -> GameBoard:
+  async def create_game_board(self, game_state:dict) -> GameBoard:
     game_id = game_state['game']['id']
     new_game_board = GameBoard(
       game_id=game_id,
@@ -31,7 +37,7 @@ class GameRuntimeService:
       ruleset=game_state['game']['ruleset'],
       source=game_state['game']['source'],
       map=game_state['game']['map'],
-      snake_class=snake_builder.build(self.snake_type),
+      snake_class=SnakeBuilder.build(self.snake_type),
     )
     await new_game_board.start_game(game_state)
 
@@ -57,7 +63,7 @@ class GameRuntimeService:
     self.game_last_seen_unix.pop(game_id, None)
     await self.game_state_store.delete(game_id)
 
-  async def get_game_board(self, game_state:dict, snake_builder:SnakeBuilder, end:bool=False) -> GameBoard:
+  async def get_game_board(self, game_state:dict, end:bool=False) -> GameBoard:
     game_id = game_state['game']['id']
     game_board:GameBoard
     if self.game_state_local_cache and game_id in self.running_games:
@@ -69,13 +75,12 @@ class GameRuntimeService:
         if self.game_state_local_cache:
           self.running_games[game_id] = game_board
       else:
-        game_board = await self.create_game_board(game_state, snake_builder)
+        game_board = await self.create_game_board(game_state)
         if self.metrics_collector is not None:
           await self.metrics_collector.record_game_autocreated()
 
     if not end:
       self.game_move_counts[game_id] = self.game_move_counts.get(game_id, 0) + 1
-
     self.game_last_seen_unix[game_id] = int(time.time())
 
     game_board.read_game_data(game_state)