diff --git a/main.py b/main.py index 9005a6e..f51ebe6 100644 --- a/main.py +++ b/main.py @@ -83,7 +83,7 @@ if __name__ == "__main__": c = PyClient([ "http://127.0.0.1:5001", - "http://127.0.0.1:5002" + "http://127.0.0.1:5002", ], debug=True) - out = c.play_game("/tic-tac-toe", TicTacToe) + out = c.play_game("tic-tac-toe", TicTacToe) diff --git a/pyserver/server.py b/pyserver/server.py index 288323d..57d8a85 100644 --- a/pyserver/server.py +++ b/pyserver/server.py @@ -5,124 +5,183 @@ from __future__ import annotations from collections.abc import Callable, Mapping from typing import Any, Optional -from flask import Flask, jsonify, request +from flask import Flask, Response, jsonify, request +from functools import wraps - -GameHandler = Callable[[Any], Any] +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, profile: Optional[Mapping[str, Any]] = None, import_name: str = __name__) -> None: + 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. - :param profile: Additional root-level discovery metadata. + :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'.") + raise ValueError( + "Root profile metadata must not define 'name'." + ) if "games" in self.profile: - raise ValueError("Root profile metadata must not define 'games'.") + 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.__games: dict[str, PayloadType] = {} + self.__registered_routes: set[str] = set() - self._register_root() + # 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. - def _register_root(self) -> None: - def discovery() -> Any: - payload = {"name": self.name, **self.profile, "games": dict(self._games)} - return jsonify(payload) + :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) - self.app.add_url_rule("/", endpoint="botman_discovery", view_func=discovery, methods=["GET"]) + if url in self.__registered_routes: + raise ValueError( + f"Route already registered: {url}" + ) + + self.__registered_routes.add(url) - @staticmethod - def _normalize_segment(segment: str) -> str: - normalized = segment.strip("/") - if not normalized: - raise ValueError("Route segments must not be empty.") - return normalized + 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. - 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}" + :return: The personal discovery information. + :rtype: dict[str, Any] + """ + return { + "name": self.name, + "games": dict(self.__games), + **self.profile, + } - def _register_action(self, route: str, handler: GameHandler) -> None: - if route in self._registered_routes: - raise ValueError(f"Route already registered: {route}") + 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 - 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 __make_url(self, name : str, route : str) -> str: + return "/" + "/".join([ name.strip("/"), route.strip("/") ]).strip("/") def add_game( self, - game_name: str, - custom_profile: Mapping[str, Any], - actions: Mapping[str, GameHandler], - required_actions: Optional[tuple[str, ...]] = None, + 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 game_name: Base route name for the game. - :param custom_profile: Game-specific discovery metadata. + :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. """ - 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) + # Verify that all required actions are present 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))}") + missing_actions = [ action for action in required_actions if action not in 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)) + 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() - 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.") + for route in actions: + url = self.__make_url(name, route) - 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]}") + 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 - 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: + def add_tic_tac_toe(self, profile: PayloadType, on_move: GameHandler) -> None: """ Convenience registration for tic-tac-toe. - The game is exposed at ``/tic-tac-toe``. + The game is exposed at `/tic-tac-toe`. """ - self.add_game("tic-tac-toe", custom_profile, {"": 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."""