"""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, Response, jsonify, request from functools import wraps PayloadType = dict[str, Any] GameHandler = Callable[[PayloadType], PayloadType] class PyServer: """A tiny stateless Flask app that serves discovery and game routes.""" def __init__(self, name: str, import_name: str = __name__, profile: Optional[PayloadType] = None, subpath : str = "", ) -> None: """ Create a PyServer. :param name: Preferred display name for discovery. :type name: str :param import_name: Flask import name. :type import_name: str :param profile: Additional root-level discovery metadata. :type profile: Optional[Dict[str, Any]] :type subpath: str :raises ValueError: The input contains invalid information. """ 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, PayloadType] = {} self.__registered_routes: set[str] = set() # Register the root self.__add_api_endpoint("", "", lambda _ : self.__discovery()) # self.app.add_url_rule("/", # endpoint="botman_discovery", # view_func=self.__discovery, # methods=["GET"] # ) def __add_api_endpoint(self, name : str, route : str, func : GameHandler) -> None: """ Create a new API endpoint. :param name: The name of the action to undertake. :type name: str :param func: The player's function that determines what action to take. :type func: Callable[[dict[str, Any]], dict[str, Any]] :raises ValueError: The URL has already been registered. """ url = self.__make_url(name, route) if url in self.__registered_routes: raise ValueError( f"Route already registered: {url}" ) self.__registered_routes.add(url) return self.app.add_url_rule(url, endpoint="botman_" + name.replace("/", "_"), view_func=self.__func_wrapper(func), methods=["GET"], ) def __discovery(self) -> PayloadType: """ Return the server's discovery information. :return: The personal discovery information. :rtype: dict[str, Any] """ return { "name": self.name, "games": dict(self.__games), **self.profile, } def __func_wrapper(self, func : GameHandler) -> Callable[[], Response]: """ Wrapper that catches an incoming request, parses it, and responds with a player's action response. """ @wraps(func) def exec(): payload = request.get_json(silent=True) or {} result = func(payload) or {} return jsonify(result) return exec def __make_url(self, name : str, route : str) -> str: return "/" + "/".join([ name.strip("/"), route.strip("/") ]).strip("/") def add_game( self, name: str, profile: PayloadType, actions: dict[str, GameHandler], required_actions: Optional[list[str]] = None, ) -> None: """ Register a stateless game with one or more action routes. :param name: Base route name for the game. :type name: str :param profile: Game-specific discovery metadata. :type profile: dict[str, Any] :param actions: Mapping of action subpaths to handlers. :type actions: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] :param required_actions: Optional completeness check. If provided, all required action names must exist in ``actions`` and map to callables. :type required_actions: Optional[list[str]] :raises AssertionError: Some of the required actions aren't present. :raises ValueError: Some of the requested URL paths are already occupied. """ # Verify that all required actions are present if required_actions is not None: missing_actions = [ action for action in required_actions if action not in actions ] if len(missing_actions) > 0: raise AssertionError( f"Missing required action handlers: {', '.join(sorted(missing_actions))}" ) # Verify that there are no duplicate URLs being registered # Even though this will automatically be checked later, checking this # now ensures that the operation is atomic and doesn't add # games partially. new_routes : set[str] = set() for route in actions: url = self.__make_url(name, route) if url in self.__registered_routes: raise ValueError( "Route {url} was already registered" ) if url in new_routes: raise ValueError( "Route {url} was registered twice by the same function call" ) new_routes.add(url) # Register all actions for route, func in actions.items(): self.__add_api_endpoint(name=name, route=route, func=func) # Add profile data self.__games[name] = profile def add_tic_tac_toe(self, on_move: GameHandler, profile: PayloadType = {}) -> None: """ Convenience registration for tic-tac-toe. The game is exposed at `/tic-tac-toe`. :param profile: The player's custom profile. :type profile: dict[str, Any] :param on_move: """ self.add_game( name="tic-tac-toe", profile=profile, actions={"": on_move}, required_actions=[""], ) 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)