130 lines
5.2 KiB
Python
130 lines
5.2 KiB
Python
"""Stateless Flask server helpers for Bot-Man-Toe games."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable, Mapping
|
|
from typing import Any, Optional
|
|
|
|
from flask import Flask, jsonify, request
|
|
|
|
|
|
GameHandler = Callable[[Any], Any]
|
|
|
|
|
|
class PyServer:
|
|
"""A tiny stateless Flask app that serves discovery and game routes."""
|
|
|
|
def __init__(self, name: str, profile: Optional[Mapping[str, Any]] = None, import_name: str = __name__) -> None:
|
|
"""
|
|
Create a PyServer.
|
|
|
|
:param name: Preferred display name for discovery.
|
|
:param profile: Additional root-level discovery metadata.
|
|
:param import_name: Flask import name.
|
|
"""
|
|
self.name = name
|
|
self.profile = dict(profile or {})
|
|
|
|
if "name" in self.profile:
|
|
raise ValueError("Root profile metadata must not define 'name'.")
|
|
if "games" in self.profile:
|
|
raise ValueError("Root profile metadata must not define 'games'.")
|
|
|
|
self.app = Flask(import_name)
|
|
self._games: dict[str, dict[str, Any]] = {}
|
|
self._registered_routes: set[str] = set()
|
|
|
|
self._register_root()
|
|
|
|
def _register_root(self) -> None:
|
|
def discovery() -> Any:
|
|
payload = {"name": self.name, **self.profile, "games": dict(self._games)}
|
|
return jsonify(payload)
|
|
|
|
self.app.add_url_rule("/", endpoint="botman_discovery", view_func=discovery, methods=["GET"])
|
|
|
|
@staticmethod
|
|
def _normalize_segment(segment: str) -> str:
|
|
normalized = segment.strip("/")
|
|
if not normalized:
|
|
raise ValueError("Route segments must not be empty.")
|
|
return normalized
|
|
|
|
def _compose_route(self, game_name: str, action_name: str) -> str:
|
|
normalized_game = self._normalize_segment(game_name)
|
|
normalized_action = action_name.strip("/")
|
|
return f"/{normalized_game}" if not normalized_action else f"/{normalized_game}/{normalized_action}"
|
|
|
|
def _register_action(self, route: str, handler: GameHandler) -> None:
|
|
if route in self._registered_routes:
|
|
raise ValueError(f"Route already registered: {route}")
|
|
|
|
endpoint = f"botman_{len(self._registered_routes)}"
|
|
|
|
def view() -> Any:
|
|
payload = request.get_json(silent=True)
|
|
return jsonify(handler(payload))
|
|
|
|
self.app.add_url_rule(route, endpoint=endpoint, view_func=view, methods=["GET"])
|
|
self._registered_routes.add(route)
|
|
|
|
def add_game(
|
|
self,
|
|
game_name: str,
|
|
custom_profile: Mapping[str, Any],
|
|
actions: Mapping[str, GameHandler],
|
|
required_actions: Optional[tuple[str, ...]] = None,
|
|
) -> None:
|
|
"""
|
|
Register a stateless game with one or more action routes.
|
|
|
|
:param game_name: Base route name for the game.
|
|
:param custom_profile: Game-specific discovery metadata.
|
|
:param actions: Mapping of action subpaths to handlers.
|
|
:param required_actions: Optional completeness check. If provided, all
|
|
required action names must exist in ``actions`` and map to callables.
|
|
"""
|
|
if not isinstance(custom_profile, Mapping):
|
|
raise TypeError("custom_profile must be a mapping.")
|
|
if not isinstance(actions, Mapping) or not actions:
|
|
raise ValueError("actions must be a non-empty mapping.")
|
|
|
|
normalized_profile = dict(custom_profile)
|
|
|
|
if required_actions is not None:
|
|
missing_actions = [action_name for action_name in required_actions if action_name not in actions]
|
|
if missing_actions:
|
|
raise ValueError(f"Missing required action handlers: {', '.join(sorted(missing_actions))}")
|
|
|
|
prepared_routes: list[tuple[str, GameHandler]] = []
|
|
for action_name, handler in actions.items():
|
|
if not isinstance(action_name, str):
|
|
raise TypeError("Action names must be strings.")
|
|
if handler is None or not callable(handler):
|
|
raise TypeError(f"Action handler for '{action_name}' must be callable.")
|
|
prepared_routes.append((self._compose_route(game_name, action_name), handler))
|
|
|
|
normalized_routes = [route for route, _ in prepared_routes]
|
|
if len(set(normalized_routes)) != len(normalized_routes):
|
|
raise ValueError("Action routes must be unique after normalization.")
|
|
|
|
duplicate_routes = [route for route, _ in prepared_routes if route in self._registered_routes or route in self._games]
|
|
if duplicate_routes:
|
|
raise ValueError(f"Route already registered: {duplicate_routes[0]}")
|
|
|
|
for route, handler in prepared_routes:
|
|
self._register_action(route, handler)
|
|
self._games[route] = dict(normalized_profile)
|
|
|
|
def add_tic_tac_toe(self, custom_profile: Mapping[str, Any], on_move: GameHandler) -> None:
|
|
"""
|
|
Convenience registration for tic-tac-toe.
|
|
|
|
The game is exposed at ``/tic-tac-toe``.
|
|
"""
|
|
self.add_game("tic-tac-toe", custom_profile, {"": on_move})
|
|
|
|
def start(self, host: str = "127.0.0.1", port: int = 5000, debug: bool = False, **kwargs: Any) -> None:
|
|
"""Start the Flask development server."""
|
|
self.app.run(host=host, port=port, debug=debug, **kwargs)
|