Bot-Man-Toe/pyserver/server.py

"""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)