Refactor: Updated Return Types for Better Clarity

opengsq/protocols/battlefield.py

@@ -76,7 +76,7 @@ async def get_version(self) -> VersionInfo:
         data = await self.__get_data(self._version)
         return VersionInfo(data[0], data[1])
 
-    async def get_players(self) -> list:
+    async def get_players(self) -> list[dict[str, str]]:
         """
         Asynchronously retrieves the list of players on the game server.
 

opengsq/protocols/doom3.py

@@ -1,9 +1,11 @@
 import re
+from typing import Union
 
 from opengsq.binary_reader import BinaryReader
 from opengsq.exceptions import InvalidPacketException
 from opengsq.protocol_base import ProtocolBase
 from opengsq.protocol_socket import UdpClient
+from opengsq.responses.doom3 import Status
 
 
 class Doom3(ProtocolBase):
@@ -19,7 +21,7 @@ class Doom3(ProtocolBase):
         "etqw": ["id", "ping", "name", "clantag_pos", "clantag", "typeflag"],
     }
 
-    async def get_info(self, strip_color=True) -> dict:
+    async def get_status(self, strip_color=True) -> Status:
         """
         Asynchronously retrieves the information of the game server.
 
@@ -65,22 +67,26 @@ async def get_info(self, strip_color=True) -> dict:
 
             info[key] = Doom3.strip_colors(val) if strip_color else val
 
+        players = []
+
         stream_position = br.stream_position
 
         # Try parse the fields
         for mod in self._player_fields.keys():
             try:
-                info["players"] = self.__parse_player(
-                    br, self._player_fields[mod], strip_color
-                )
+                players = self.__parse_player(br, self._player_fields[mod], strip_color)
                 break
             except Exception:
-                info["players"] = []
+                players = []
                 br.stream_position = stream_position
 
-        return info
+        status = Status(info, players)
+
+        return status
 
-    def __parse_player(self, br: BinaryReader, fields: list, strip_color: bool):
+    def __parse_player(
+        self, br: BinaryReader, fields: list, strip_color: bool
+    ) -> list[dict[str, Union[int, str]]]:
         """
         Parses the player information from the BinaryReader object.
 
@@ -130,17 +136,17 @@ def strip_colors(text: str) -> str:
     async def main_async():
         # doom3
         doom3 = Doom3(host="66.85.14.240", port=27666, timeout=5.0)
-        info = await doom3.get_info()
+        info = await doom3.get_status()
         print(info)
 
         # etqw
         doom3 = Doom3(host="178.162.135.83", port=27735, timeout=5.0)
-        info = await doom3.get_info()
+        info = await doom3.get_status()
         print(info)
 
         # quake4
         doom3 = Doom3(host="88.99.0.7", port=28007, timeout=5.0)
-        info = await doom3.get_info()
+        info = await doom3.get_status()
         print(info)
 
     asyncio.run(main_async())

opengsq/protocols/eos.py

@@ -1,3 +1,4 @@
+from typing import Any
 import aiohttp
 import base64
 import json
@@ -152,7 +153,7 @@ async def get_matchmaking(
 
         return Matchmaking(sessions=data["sessions"], count=data["count"])
 
-    async def get_info(self) -> dict:
+    async def get_info(self) -> dict[str, Any]:
         """
         Retrieves the matchmaking information for the current server.
 

opengsq/protocols/fivem.py

@@ -1,3 +1,4 @@
+from typing import Any
 import aiohttp
 import time
 
@@ -11,7 +12,7 @@ class FiveM(ProtocolBase):
 
     full_name = "FiveM Protocol"
 
-    async def _get(self, filename: str) -> dict:
+    async def _get(self, filename: str) -> dict[str, Any]:
         """
         Asynchronously retrieves the JSON data from the given filename on the server.
 
@@ -24,23 +25,23 @@ async def _get(self, filename: str) -> dict:
             async with session.get(url) as response:
                 return await response.json(content_type=None)
 
-    async def get_info(self) -> dict:
+    async def get_info(self) -> dict[str, Any]:
         """
         Asynchronously retrieves the information of the game server.
 
         :return: A dictionary containing the information of the game server.
         """
         return await self._get("info")
 
-    async def get_players(self) -> list:
+    async def get_players(self) -> list[dict[str, Any]]:
         """
         Asynchronously retrieves the list of players on the game server.
 
         :return: A list of players on the game server.
         """
         return await self._get("players")
 
-    async def get_dynamic(self) -> dict:
+    async def get_dynamic(self) -> dict[str, Any]:
         """
         Asynchronously retrieves the dynamic data of the game server.
 

opengsq/protocols/gamespy1.py

@@ -21,7 +21,7 @@ class __Request:
         STATUS = b"\\status\\xserverquery"
         TEAMS = b"\\teams\\"
 
-    async def get_basic(self) -> dict:
+    async def get_basic(self) -> dict[str, str]:
         """
         Asynchronously retrieves the basic information of the game server.
 
@@ -32,7 +32,7 @@ async def get_basic(self) -> dict:
         )
 
     # Server may still response with Legacy version
-    async def get_info(self, xserverquery: bool = True) -> dict:
+    async def get_info(self, xserverquery: bool = True) -> dict[str, str]:
         """
         Asynchronously retrieves the information of the current game running on the server.
 
@@ -46,12 +46,12 @@ async def get_info(self, xserverquery: bool = True) -> dict:
         )
         return self.__parse_as_key_values(await self.__connect_and_send(data))
 
-    async def get_rules(self, xserverquery: bool = True) -> list:
+    async def get_rules(self, xserverquery: bool = True) -> dict[str, str]:
         """
         Asynchronously retrieves the rules of the current game running on the server.
 
         :param xserverquery: A boolean indicating whether to use XServerQuery.
-        :return: A list containing the rules of the current game.
+        :return: A dictionary containing the rules of the current game.
         """
         data = (
             xserverquery
@@ -60,7 +60,7 @@ async def get_rules(self, xserverquery: bool = True) -> list:
         )
         return self.__parse_as_key_values(await self.__connect_and_send(data))
 
-    async def get_players(self, xserverquery: bool = True) -> list:
+    async def get_players(self, xserverquery: bool = True) -> list[dict[str, str]]:
         """
         Asynchronously retrieves the information of each player on the server.
 
@@ -99,7 +99,7 @@ async def get_status(self, xserverquery: bool = True) -> Status:
 
         return Status(info, players, teams)
 
-    async def get_teams(self) -> list:
+    async def get_teams(self) -> list[dict[str, str]]:
         """
         Asynchronously retrieves the information of each team on the server.
 
@@ -149,7 +149,9 @@ async def __connect_and_send(self, data) -> BinaryReader:
 
         return br
 
-    def __parse_as_key_values(self, br: BinaryReader, is_status=False):
+    def __parse_as_key_values(
+        self, br: BinaryReader, is_status=False
+    ) -> dict[str, str]:
         kv = {}
 
         # Bind key value
@@ -170,7 +172,9 @@ def __parse_as_key_values(self, br: BinaryReader, is_status=False):
 
         return kv
 
-    def __parse_as_object(self, br: BinaryReader, is_player=False):
+    def __parse_as_object(
+        self, br: BinaryReader, is_player=False
+    ) -> list[dict[str, str]]:
         items, keyhashes, filters = {}, [], []
 
         while br.remaining_bytes() > 0:

opengsq/protocols/minecraft.py

@@ -1,6 +1,7 @@
 import json
 import re
 import struct
+from typing import Any
 
 from opengsq.responses.minecraft import StatusPre17
 from opengsq.binary_reader import BinaryReader
@@ -16,7 +17,7 @@ class Minecraft(ProtocolBase):
 
     full_name = "Minecraft Protocol"
 
-    async def get_status(self, version=47, strip_color=True) -> dict:
+    async def get_status(self, version=47, strip_color=True) -> dict[str, Any]:
         """
         Asynchronously retrieves the status of the game server.
 
@@ -122,7 +123,7 @@ async def get_status_pre17(self, strip_color=True) -> StatusPre17:
         return StatusPre17(**result)
 
     @staticmethod
-    def strip_colors(text: str):
+    def strip_colors(text: str) -> get_status_pre17:
         """
         Strips color codes from the given text.
 

opengsq/protocols/quake3.py

@@ -25,7 +25,7 @@ def __init__(self, host: str, port: int, timeout: float = 5.0):
         self._request_header = b"getstatus"
         self._response_header = "statusResponse\n"
 
-    async def get_info(self, strip_color=True) -> dict:
+    async def get_info(self, strip_color=True) -> dict[str, str]:
         """
         Asynchronously retrieves the information of the game server.
 

opengsq/protocols/teamspeak3.py

@@ -23,7 +23,7 @@ def __init__(self, host: str, port: int, voice_port: int, timeout: float = 5):
         super().__init__(host, port, timeout)
         self._voice_port = voice_port
 
-    async def get_info(self) -> dict:
+    async def get_info(self) -> dict[str, str]:
         """
         Asynchronously retrieves the information of the game server.
 
@@ -32,7 +32,7 @@ async def get_info(self) -> dict:
         response = await self.__send_and_receive(b"serverinfo")
         return self.__parse_kvs(response)
 
-    async def get_clients(self) -> list[dict]:
+    async def get_clients(self) -> list[dict[str, str]]:
         """
         Asynchronously retrieves the list of clients on the game server.
 
@@ -41,7 +41,7 @@ async def get_clients(self) -> list[dict]:
         response = await self.__send_and_receive(b"clientlist")
         return self.__parse_rows(response)
 
-    async def get_channels(self) -> list[dict]:
+    async def get_channels(self) -> list[dict[str, str]]:
         """
         Asynchronously retrieves the list of channels on the game server.
 
@@ -87,7 +87,7 @@ def __parse_rows(self, response: bytes):
         """
         return [self.__parse_kvs(row) for row in response.split(b"|")]
 
-    def __parse_kvs(self, response: bytes):
+    def __parse_kvs(self, response: bytes) -> dict[str, str]:
         """
         Parses the key-value pairs from the given response.
 
@@ -111,7 +111,6 @@ def __parse_kvs(self, response: bytes):
 
 if __name__ == "__main__":
     import asyncio
-    import json
 
     async def main_async():
         teamspeak3 = TeamSpeak3(

opengsq/protocols/unreal2.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import re
+from typing import Any
 
 from opengsq.responses.unreal2 import Player, Status
 from opengsq.binary_reader import BinaryReader
@@ -58,7 +59,7 @@ async def get_details(self) -> Status:
             skill=self._read_string(br),
         )
 
-    async def get_rules(self) -> dict:
+    async def get_rules(self) -> dict[str, Any]:
         """
         Asynchronously gets the rules of a server.
 
@@ -131,7 +132,7 @@ async def get_players(self) -> list[Player]:
         return players
 
     @staticmethod
-    def strip_colors(text: str):
+    def strip_colors(text: str) -> str:
         """
         Strips color codes from a string.
 

opengsq/responses/ase/player.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from typing import Optional
 
 
 @dataclass
@@ -9,20 +10,20 @@ class Player:
     Represents a player in the game.
     """
 
-    name: str
+    name: Optional[str] = None
     """The name of the player."""
 
-    team: str
+    team: Optional[str] = None
     """The team of the player."""
 
-    skin: str
+    skin: Optional[str] = None
     """The skin of the player."""
 
-    score: int
+    score: Optional[int] = None
     """The score of the player."""
 
-    ping: int
+    ping: Optional[int] = None
     """The ping of the player."""
 
-    time: int
+    time: Optional[int] = None
     """The time of the player."""

opengsq/responses/doom3/__init__.py

@@ -0,0 +1 @@
+from .status import Status

opengsq/responses/doom3/status.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Union
+
+
+@dataclass
+class Status:
+    """
+    Represents the server status.
+    """
+
+    info: dict[str, str]
+    """Server's info."""
+
+    players: list[dict[str, Union[int, str]]]
+    """Server's players."""

opengsq/responses/source/player.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+from typing import Optional
 
 
 @dataclass
@@ -16,8 +17,8 @@ class Player:
     duration: float
     """Player Duration"""
 
-    deaths: int = None
+    deaths: Optional[int] = None
     """Player Deaths"""
 
-    money: int = None
+    money: Optional[int] = None
     """Player Money"""

tests/protocols/test_doom3.py

@@ -11,6 +11,6 @@
 
 
 @pytest.mark.asyncio
-async def test_get_info():
-    result = await doom3.get_info()
-    await handler.save_result("test_get_info", result)
+async def test_get_status():
+    result = await doom3.get_status()
+    await handler.save_result("test_get_status", result)