reuse the code from quart_common.web.env to import env_bool and env_int into all classes that are useing this functions

server/Server.py

@@ -1,4 +1,5 @@
 from quart_common.web.logger import build_logger, await_log
+from quart_common.web.env import env_bool, env_int
 from server.Files import read_file
 
 from server.game_state_store import GameStateStoreBuilder
@@ -64,7 +65,7 @@ class Server:
     self.dashboard_ws_hub = DashboardWebSocketHub()
     dashboard_event_origin = f'worker-{os.getpid()}-{int(time.time() * 1000)}'
     dashboard_events_channel = os.getenv('DASHBOARD_EVENTS_CHANNEL', 'snake:dashboard:events')
-    dashboard_events_enabled = (self.metrics_backend_normalized == 'redis' and self._env_bool('DASHBOARD_EVENTS_ENABLED', True))
+    dashboard_events_enabled = (self.metrics_backend_normalized == 'redis' and env_bool('DASHBOARD_EVENTS_ENABLED', True))
 
     self.metrics_collector = MetricsCollector(
       metrics_manager=MetricsStoreBuilder.build(
@@ -80,8 +81,8 @@ class Server:
       game_last_seen_unix=self.game_last_seen_unix,
       game_move_counts=self.game_move_counts,
     )
-    self.clear_worker_metrics_on_startup = self._env_bool('METRICS_CLEAR_WORKERS_ON_STARTUP', True)
-    self.worker_metrics_startup_lock_ttl_sec = self._env_int('METRICS_STARTUP_CLEANUP_LOCK_TTL_SEC', 300)
+    self.clear_worker_metrics_on_startup = env_bool('METRICS_CLEAR_WORKERS_ON_STARTUP', True)
+    self.worker_metrics_startup_lock_ttl_sec = env_int('METRICS_STARTUP_CLEANUP_LOCK_TTL_SEC', 300)
     self.dashboard_running_game_stale_sec = 600
     self._startup_worker_metrics_cleared = False
 
@@ -204,26 +205,7 @@ class Server:
     return str(version)
 
   def _get_stale_game_timeout_sec(self) -> int:
-    value = os.getenv('SNAKE_STUCK_GAME_TIMEOUT_SEC', '180')
-    try:
-      return max(30, int(value))
-    except ValueError:
-      return 180
-
-  def _env_bool(self, name:str, default:bool=False) -> bool:
-    value = os.getenv(name)
-    if value is None:
-      return default
-    return value.strip().lower() in {'1', 'true', 'yes', 'on'}
-
-  def _env_int(self, name:str, default:int) -> int:
-    value = os.getenv(name)
-    if value is None:
-      return default
-    try:
-      return int(value)
-    except ValueError:
-      return default
+    return max(30, env_int('SNAKE_STUCK_GAME_TIMEOUT_SEC', 180))
 
   async def _create_game_board(self, game_state:dict) -> GameBoard:
     game_id = game_state['game']['id']

server/bootstrap.py

@@ -2,6 +2,7 @@ from typing import TypedDict
 from pathlib import Path
 import os
 
+from quart_common.web.env import env_bool, env_int
 from server.Server import Server
 
 class RunConfig(TypedDict):
@@ -9,18 +10,13 @@ class RunConfig(TypedDict):
   port: int
   debug: bool
 
-def env_bool(name:str, default:bool=False) -> bool:
-  value = os.environ.get(name)
-  if value is None:
-    return default
-  return value.lower() in {'1', 'true', 'yes', 'on'}
 
 def build_server_from_env(default_snake_type:str) -> Server:
   data_path = str(Path(__file__).resolve().parent.parent)
   redis_url = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
   game_state_backend = os.environ.get('GAME_STATE_BACKEND', 'memory')
   game_state_redis_url = os.environ.get('GAME_STATE_REDIS_URL', redis_url)
-  game_state_ttl_sec = int(os.environ.get('GAME_STATE_TTL_SEC', '900'))
+  game_state_ttl_sec = env_int('GAME_STATE_TTL_SEC', 900)
 
   metrics_backend = os.environ.get('METRICS_BACKEND', None)
   if metrics_backend is None:
@@ -31,14 +27,14 @@ def build_server_from_env(default_snake_type:str) -> Server:
   if metrics_ttl_sec_raw is None:
     metrics_ttl_sec = (game_state_ttl_sec if metrics_backend.strip().lower() == 'redis' else None)
   else:
-    metrics_ttl_sec = int(metrics_ttl_sec_raw)
+    metrics_ttl_sec = env_int('METRICS_TTL_SEC', game_state_ttl_sec)
 
   gameplay_db_enabled = env_bool('GAMEPLAY_DB_ENABLED', True)
   gameplay_db_path = os.environ.get(
     'GAMEPLAY_DB_PATH',
     os.path.join(data_path, 'data', 'database', 'gameplay.sqlite3'),
   )
-  gameplay_db_busy_timeout_ms = int(os.environ.get('GAMEPLAY_DB_BUSY_TIMEOUT_MS', '5000'))
+  gameplay_db_busy_timeout_ms = env_int('GAMEPLAY_DB_BUSY_TIMEOUT_MS', 5000)
 
   server = Server(
     data_path=data_path,
@@ -66,6 +62,6 @@ def build_server_from_env(default_snake_type:str) -> Server:
 def build_run_config() -> RunConfig:
   return {
     'host': os.environ.get('HOST', '0.0.0.0'),
-    'port': int(os.environ.get('PORT', '8000')),
+    'port': env_int('PORT', 8000),
     'debug': env_bool('DEBUG'),
   }

server/dataset/RLBootstrapDataset.py

@@ -2,34 +2,18 @@ from pathlib import Path
 from typing import Any
 import os
 
+from quart_common.web.env import env_bool, env_int
 from server.dataset.DatasetIO import DatasetIO
 
 class RLBootstrapDataset:
   def __init__(self):
-    self.enabled = self._env_bool("RL_BOOTSTRAP_ENABLED", default=False)
-    self.min_base_rows = self._env_int("RL_MIN_BASE_ROWS", default=5000)
+    self.enabled = env_bool("RL_BOOTSTRAP_ENABLED", default=False)
+    self.min_base_rows = env_int("RL_MIN_BASE_ROWS", default=5000)
     self.base_dataset_path = Path(os.getenv("RL_BASE_DATASET", "data/dataset/best_moves.jsonl"))
     self.output_path = Path(os.getenv("RL_BOOTSTRAP_OUTPUT", "data/dataset/rl_bootstrap.jsonl"))
     self.max_bytes = int(float(os.getenv("RL_BOOTSTRAP_MAX_MB", "50")) * 1024 * 1024)
     self.needs_more_data = False
 
-  @staticmethod
-  def _env_bool(name:str, default:bool=False) -> bool:
-    value = os.getenv(name)
-    if value is None:
-      return default
-    return value.lower() in {"1", "true", "yes", "on"}
-
-  @staticmethod
-  def _env_int(name:str, default:int) -> int:
-    value = os.getenv(name)
-    if value is None:
-      return default
-    try:
-      return int(value)
-    except ValueError:
-      return default
-
   def refresh_state(self):
     if not self.enabled:
       self.needs_more_data = False