Bram
/
Bot-Man-Toe
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Refactor PyServer class 

Improve readability and add documentation
main
Bram van den Heuvel 2026-06-06 17:48:28 +02:00
parent dcc8bb9636
commit f9911507f4
2 changed files with 132 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
main.py
View File

@ -83,7 +83,7 @@ if __name__ == "__main__":
c = PyClient([ c = PyClient([
"http://127.0.0.1:5001", "http://127.0.0.1:5001",
"http://127.0.0.1:5002" "http://127.0.0.1:5002",
], debug=True) ], debug=True)
out = c.play_game("/tic-tac-toe", TicTacToe) out = c.play_game("tic-tac-toe", TicTacToe)

201
pyserver/server.py
View File

@ -5,124 +5,183 @@ from __future__ import annotations
from collections.abc import Callable, Mapping from collections.abc import Callable, Mapping
from typing import Any, Optional from typing import Any, Optional
from flask import Flask, jsonify, request from flask import Flask, Response, jsonify, request
from functools import wraps
PayloadType = dict[str, Any]
GameHandler = Callable[[Any], Any] GameHandler = Callable[[PayloadType], PayloadType]
class PyServer: class PyServer:
"""A tiny stateless Flask app that serves discovery and game routes.""" """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. Create a PyServer.
:param name: Preferred display name for discovery. :param name: Preferred display name for discovery.
:param profile: Additional root-level discovery metadata. :type name: str
:param import_name: Flask import name. :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.name = name
self.profile = dict(profile or {}) self.profile = dict(profile or {})
if "name" in self.profile: 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: 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.app = Flask(import_name)
self._games: dict[str, dict[str, Any]] = {} self.__games: dict[str, PayloadType] = {}
self._registered_routes: set[str] = set() 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: :param name: The name of the action to undertake.
def discovery() -> Any: :type name: str
payload = {"name": self.name, **self.profile, "games": dict(self._games)} :param func: The player's function that determines what action to take.
return jsonify(payload) :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 return self.app.add_url_rule(url,
def _normalize_segment(segment: str) -> str: endpoint="botman_" + name.replace("/", "_"),
normalized = segment.strip("/") view_func=self.__func_wrapper(func),
if not normalized: methods=["GET"],
raise ValueError("Route segments must not be empty.") )
return normalized
def __discovery(self) -> PayloadType:
"""
Return the server's discovery information.
def _compose_route(self, game_name: str, action_name: str) -> str: :return: The personal discovery information.
normalized_game = self._normalize_segment(game_name) :rtype: dict[str, Any]
normalized_action = action_name.strip("/") """
return f"/{normalized_game}" if not normalized_action else f"/{normalized_game}/{normalized_action}" return {
"name": self.name,
"games": dict(self.__games),
**self.profile,
}
def _register_action(self, route: str, handler: GameHandler) -> None: def __func_wrapper(self, func : GameHandler) -> Callable[[], Response]:
if route in self._registered_routes: """
raise ValueError(f"Route already registered: {route}") 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 __make_url(self, name : str, route : str) -> str:
return "/" + "/".join([ name.strip("/"), route.strip("/") ]).strip("/")
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( def add_game(
self, self,
game_name: str, name: str,
custom_profile: Mapping[str, Any], profile: PayloadType,
actions: Mapping[str, GameHandler], actions: dict[str, GameHandler],
required_actions: Optional[tuple[str, ...]] = None, required_actions: Optional[list[str]] = None,
) -> None: ) -> None:
""" """
Register a stateless game with one or more action routes. Register a stateless game with one or more action routes.
:param game_name: Base route name for the game. :param name: Base route name for the game.
:param custom_profile: Game-specific discovery metadata. :type name: str
:param profile: Game-specific discovery metadata.
:type profile: dict[str, Any]
:param actions: Mapping of action subpaths to handlers. :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 :param required_actions: Optional completeness check. If provided, all
required action names must exist in ``actions`` and map to callables. 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: if required_actions is not None:
missing_actions = [action_name for action_name in required_actions if action_name not in actions] missing_actions = [ action for action in required_actions if action not in actions ]
if missing_actions:
raise ValueError(f"Missing required action handlers: {', '.join(sorted(missing_actions))}")
prepared_routes: list[tuple[str, GameHandler]] = [] if len(missing_actions) > 0:
for action_name, handler in actions.items(): raise AssertionError(
if not isinstance(action_name, str): f"Missing required action handlers: {', '.join(sorted(missing_actions))}"
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.") # Verify that there are no duplicate URLs being registered
prepared_routes.append((self._compose_route(game_name, action_name), handler)) # 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] for route in actions:
if len(set(normalized_routes)) != len(normalized_routes): url = self.__make_url(name, route)
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 url in self.__registered_routes:
if duplicate_routes: raise ValueError(
raise ValueError(f"Route already registered: {duplicate_routes[0]}") "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: def add_tic_tac_toe(self, profile: PayloadType, on_move: GameHandler) -> None:
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. 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: def start(self, host: str = "127.0.0.1", port: int = 5000, debug: bool = False, **kwargs: Any) -> None:
"""Start the Flask development server.""" """Start the Flask development server."""