99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
"""Stateless Flask server helpers for Bot-Man-Toe games."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from agents import Agent
|
|
from collections.abc import Callable, Mapping
|
|
from flask import Flask, Response, jsonify, request
|
|
from functools import wraps
|
|
from typing import Any, Optional
|
|
|
|
PayloadType = dict[str, Any]
|
|
GameHandler = Callable[[PayloadType], PayloadType]
|
|
|
|
|
|
class PyServer:
|
|
"""
|
|
A tiny stateless Flask app that serves discovery and routes
|
|
game behavior to an agent.
|
|
"""
|
|
|
|
def __init__(self, agent : Agent, import_name : str = __name__) -> None:
|
|
"""
|
|
Create a PyServer that serves the behavior of an Agent.
|
|
|
|
:param agent: The agent that one can communicate with.
|
|
:type agent: Agent
|
|
:param import_name: Flask application import name
|
|
:type import_name: str
|
|
"""
|
|
self.agent = agent
|
|
self.app = Flask(import_name)
|
|
|
|
self.__add_api_endpoint("", "", lambda _ : self.__discovery())
|
|
|
|
# Register endpoints
|
|
for name, (_, endpoints) in self.agent.registered_games.items():
|
|
if name == "":
|
|
continue
|
|
|
|
for route, func in endpoints.items():
|
|
self.__add_api_endpoint(name=name, route=route, func=func)
|
|
|
|
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)
|
|
|
|
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]
|
|
"""
|
|
d = dict(name=self.agent.name, author=self.agent.author)
|
|
if self.agent.version is not None:
|
|
d["version"] = self.agent.version
|
|
|
|
d = { **self.agent.profile, **d }
|
|
|
|
d["games"] = {}
|
|
for game, (profile, _) in self.agent.registered_games.items():
|
|
d["games"][game] = profile
|
|
|
|
return d
|
|
|
|
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 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)
|