Issue/#482 Add TypedDict definition for protocol messages

Pipfile

@@ -28,6 +28,7 @@ sortedcontainers = "*"
 sqlalchemy = ">=2.0.0"
 trueskill = "*"
 uvloop = {version = "*", markers = "sys_platform != 'win32'"}
+typing-extensions = "*"
 
 [dev-packages]
 flaky = "*"

README.md

@@ -145,11 +145,10 @@ the server API documentation:
 Work is ongoing to document these messages in a comprehensive way. For now, all
 commands that can be sent from the client -> server can be found via the server
 API documentation:
-[LobbyConnection](https://faforever.github.io/server/lobbyconnection.html)
-under the `command_*` methods. Check the source code for what fields the message
-is expected to have and any possible responses.
+[LobbyConnection](https://faforever.github.io/server/types/messages/).
 
-It may also be useful to look at the definitions in the
+For server -> client messages itt may also be useful to look at the definitions
+in the
 [faf-java-commons](https://github.com/FAForever/faf-java-commons/tree/develop/lobby/src/main/kotlin/com/faforever/commons/lobby)
 to see how the official client is deserializing messages from the server.
 

server/__init__.py

@@ -89,7 +89,10 @@
 See `server.protocol`.
 
 ### Application messages
-See `server.lobbyconnection`.
+See `server.types.messages` for type definitions explaining what messages, and
+fields are available.
+
+See `server.lobbyconnection` for the implementation of those message handlers.
 
 Terms:
 

server/game_service.py

@@ -197,9 +197,9 @@ def create_uid(self) -> int:
     def create_game(
         self,
         game_mode: str,
+        host: Player,
         game_class: type[Game] = CustomGame,
         visibility=VisibilityState.PUBLIC,
-        host: Optional[Player] = None,
         name: Optional[str] = None,
         map: Map = MAP_DEFAULT,
         password: Optional[str] = None,

server/games/game.py

@@ -6,7 +6,15 @@
 import time
 from collections import defaultdict
 from datetime import datetime
-from typing import TYPE_CHECKING, Any, Awaitable, Callable, Iterable, Optional
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Awaitable,
+    Callable,
+    Iterable,
+    Literal,
+    Optional
+)
 
 from sqlalchemy import and_, bindparam
 from sqlalchemy.exc import DBAPIError
@@ -32,6 +40,7 @@
 from server.rating import InclusiveRange, RatingType
 from server.timing import datetime_now
 from server.types import MAP_DEFAULT, Map
+from server.types.messages.server import GameInfo
 
 from ..players import Player, PlayerState
 from .typedefs import (
@@ -58,6 +67,14 @@ class GameError(Exception):
     pass
 
 
+_CLIENT_STATES: dict[GameState, Literal["closed", "open", "playing"]] = {
+    GameState.LOBBY: "open",
+    GameState.LIVE: "playing",
+    GameState.ENDED: "closed",
+    GameState.INITIALIZING: "closed",
+}
+
+
 class Game:
     """
     Object that lasts for the lifetime of a game on FAF.
@@ -71,7 +88,7 @@ def __init__(
         database: FAFDatabase,
         game_service: "GameService",
         game_stats_service: "GameStatsService",
-        host: Optional[Player] = None,
+        host: Player,
         name: str = "New Game",
         map: Map = MAP_DEFAULT,
         game_mode: str = FeaturedModType.FAF,
@@ -909,13 +926,8 @@ def is_visible_to_player(self, player: Player) -> bool:
         else:
             return player.id not in self.host.foes
 
-    def to_dict(self):
-        client_state = {
-            GameState.LOBBY: "open",
-            GameState.LIVE: "playing",
-            GameState.ENDED: "closed",
-            GameState.INITIALIZING: "closed",
-        }.get(self.state, "closed")
+    def to_dict(self) -> GameInfo:
+        client_state = _CLIENT_STATES.get(self.state, "closed")
         connected_players = self.get_connected_players()
         return {
             "command": "game_info",
@@ -949,6 +961,7 @@ def to_dict(self):
                 }
                 for team in self.teams if team is not None
             ],
+            # DEPRECATED: Use team_ids instead
             "teams": {
                 team: [
                     player.login for player in connected_players

server/ladder_service/ladder_service.py

@@ -47,6 +47,11 @@
 from server.metrics import MatchLaunch
 from server.players import Player, PlayerState
 from server.types import GameLaunchOptions, Map, NeroxisGeneratedMap
+from server.types.messages.server import (
+    MatchCancelled,
+    MatchFound,
+    SearchTimeoutTimeout
+)
 
 if TYPE_CHECKING:
     from server.lobbyconnection import LobbyConnection
@@ -298,7 +303,7 @@ def start_search(
         timeouts = self.violation_service.get_violations(players)
         if timeouts:
             self._logger.debug("timeouts: %s", timeouts)
-            times = [
+            times: list[SearchTimeoutTimeout] = [
                 {
                     "player": p.id,
                     "expires_at": violation.get_ban_expiration().isoformat()
@@ -463,7 +468,7 @@ def on_match_found(
         so it should only perform fast operations.
         """
         try:
-            msg = {"command": "match_found", "queue_name": queue.name}
+            msg: MatchFound = {"command": "match_found", "queue_name": queue.name}
 
             for player in s1.players + s2.players:
                 player.state = PlayerState.STARTING_AUTOMATCH
@@ -621,7 +626,7 @@ def make_game_options(player: Player) -> GameLaunchOptions:
                 await game.on_game_finish()
 
             game_id = game.id if game else None
-            msg = {"command": "match_cancelled", "game_id": game_id}
+            msg: MatchCancelled = {"command": "match_cancelled", "game_id": game_id}
             for player in all_players:
                 player.write_message(msg)
 

server/ladder_service/violation_service.py

@@ -10,6 +10,7 @@
 from server.decorators import with_logger
 from server.players import Player
 from server.timing import at_interval, datetime_now
+from server.types.messages.server import SearchViolationBody
 
 
 @dataclass
@@ -51,7 +52,7 @@ def is_expired(self, now: Optional[datetime] = None) -> bool:
         exp = self.time + timedelta(seconds=config.LADDER_VIOLATIONS_RESET_TIME)
         return exp <= now
 
-    def to_dict(self) -> dict:
+    def to_dict(self) -> SearchViolationBody:
         return {
             "count": self.count,
             "time": self.time.isoformat()