From e755db61eac211424ec62c3ef01c6bd817d1dd66 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alex=20N=C3=B8rgaard?= Date: Tue, 17 Jan 2023 02:10:36 +0000 Subject: [PATCH] Switch to `Any` return types as requested --- a2s/a2s_async.py | 90 +++++------------------------------------------- a2s/a2s_sync.py | 86 +++------------------------------------------ 2 files changed, 13 insertions(+), 163 deletions(-) diff --git a/a2s/a2s_async.py b/a2s/a2s_async.py index 2e56668..1a8385a 100644 --- a/a2s/a2s_async.py +++ b/a2s/a2s_async.py @@ -6,7 +6,7 @@ import logging import time from typing import ( TYPE_CHECKING, - Dict, + Any, List, NoReturn, Optional, @@ -14,7 +14,6 @@ from typing import ( Type, TypeVar, Union, - overload, ) from a2s.a2s_fragment import A2SFragment, decode_fragment @@ -22,8 +21,8 @@ from a2s.byteio import ByteReader from a2s.defaults import DEFAULT_RETRIES from a2s.exceptions import BrokenMessageError -from .info import GoldSrcInfo, InfoProtocol, SourceInfo -from .players import Player, PlayersProtocol +from .info import InfoProtocol +from .players import PlayersProtocol from .rules import RulesProtocol if TYPE_CHECKING: @@ -36,102 +35,29 @@ PROTOCOLS = Union[InfoProtocol, PlayersProtocol, RulesProtocol] logger: logging.Logger = logging.getLogger("a2s") -T = TypeVar("T", InfoProtocol, PlayersProtocol, RulesProtocol) +ProtocolT = TypeVar("ProtocolT", InfoProtocol, PlayersProtocol, RulesProtocol) -@overload async def request_async( address: Tuple[str, int], timeout: float, encoding: str, - a2s_proto: Type[InfoProtocol], -) -> Union[SourceInfo, GoldSrcInfo]: - ... - - -@overload -async def request_async( - address: Tuple[str, int], - timeout: float, - encoding: str, - a2s_proto: Type[PlayersProtocol], -) -> List[Player]: - ... - - -@overload -async def request_async( - address: Tuple[str, int], - timeout: float, - encoding: str, - a2s_proto: Type[RulesProtocol], -) -> Dict[Union[str, bytes], Union[str, bytes]]: - ... - - -async def request_async( - address: Tuple[str, int], timeout: float, encoding: str, a2s_proto: Type[T] -) -> Union[ - SourceInfo, - GoldSrcInfo, - List[Player], - Dict[Union[str, bytes], Union[str, bytes]], -]: + a2s_proto: Type[ProtocolT], +) -> Any: conn = await A2SStreamAsync.create(address, timeout) response = await request_async_impl(conn, encoding, a2s_proto) conn.close() return response -@overload -async def request_async_impl( - conn: A2SStreamAsync, - encoding: str, - a2s_proto: Type[InfoProtocol], - challenge: int = ..., - retries: int = ..., - ping: Optional[float] = ..., -) -> Union[SourceInfo, GoldSrcInfo]: - ... - - -@overload -async def request_async_impl( - conn: A2SStreamAsync, - encoding: str, - a2s_proto: Type[PlayersProtocol], - challenge: int = ..., - retries: int = ..., - ping: Optional[float] = ..., -) -> List[Player]: - ... - - -@overload -async def request_async_impl( - conn: A2SStreamAsync, - encoding: str, - a2s_proto: Type[RulesProtocol], - challenge: int = ..., - retries: int = ..., - ping: Optional[float] = ..., -) -> Dict[Union[str, bytes], Union[str, bytes]]: - ... - - async def request_async_impl( conn: A2SStreamAsync, encoding: str, - a2s_proto: Type[T], + a2s_proto: Type[ProtocolT], challenge: int = 0, retries: int = 0, ping: Optional[float] = None, -) -> Union[ - SourceInfo, - GoldSrcInfo, - Dict[Union[str, bytes], Union[str, bytes]], - List[Player], -]: +) -> Any: send_time = time.monotonic() resp_data = await conn.request(a2s_proto.serialize_request(challenge)) recv_time = time.monotonic() diff --git a/a2s/a2s_sync.py b/a2s/a2s_sync.py index 5b1fe6a..5184fc0 100644 --- a/a2s/a2s_sync.py +++ b/a2s/a2s_sync.py @@ -4,15 +4,15 @@ import io import logging import socket import time -from typing import Dict, List, Optional, Tuple, Type, TypeVar, Union, overload +from typing import Any, Optional, Tuple, Type, TypeVar, Union from a2s.a2s_fragment import decode_fragment from a2s.byteio import ByteReader from a2s.defaults import DEFAULT_RETRIES from a2s.exceptions import BrokenMessageError -from .info import GoldSrcInfo, InfoProtocol, SourceInfo -from .players import Player, PlayersProtocol +from .info import InfoProtocol +from .players import PlayersProtocol from .rules import RulesProtocol HEADER_SIMPLE = b"\xFF\xFF\xFF\xFF" @@ -25,86 +25,15 @@ logger: logging.Logger = logging.getLogger("a2s") T = TypeVar("T", InfoProtocol, RulesProtocol, PlayersProtocol) -@overload -def request_sync( - address: Tuple[str, int], - timeout: float, - encoding: str, - a2s_proto: Type[InfoProtocol], -) -> Union[SourceInfo, GoldSrcInfo]: - ... - - -@overload -def request_sync( - address: Tuple[str, int], - timeout: float, - encoding: str, - a2s_proto: Type[PlayersProtocol], -) -> List[Player]: - ... - - -@overload -def request_sync( - address: Tuple[str, int], - timeout: float, - encoding: str, - a2s_proto: Type[RulesProtocol], -) -> Dict[Union[str, bytes], Union[str, bytes]]: - ... - - def request_sync( address: Tuple[str, int], timeout: float, encoding: str, a2s_proto: Type[T] -) -> Union[ - List[Player], - GoldSrcInfo, - SourceInfo, - Dict[Union[str, bytes], Union[str, bytes]], -]: +) -> Any: conn = A2SStream(address, timeout) response = request_sync_impl(conn, encoding, a2s_proto) conn.close() return response -@overload -def request_sync_impl( - conn: A2SStream, - encoding: str, - a2s_proto: Type[InfoProtocol], - challenge: int = ..., - retries: int = ..., - ping: Optional[float] = ..., -) -> Union[SourceInfo, GoldSrcInfo]: - ... - - -@overload -def request_sync_impl( - conn: A2SStream, - encoding: str, - a2s_proto: Type[PlayersProtocol], - challenge: int = ..., - retries: int = ..., - ping: Optional[float] = ..., -) -> List[Player]: - ... - - -@overload -def request_sync_impl( - conn: A2SStream, - encoding: str, - a2s_proto: Type[RulesProtocol], - challenge: int = ..., - retries: int = ..., - ping: Optional[float] = ..., -) -> Dict[Union[str, bytes], Union[str, bytes]]: - ... - - def request_sync_impl( conn: A2SStream, encoding: str, @@ -112,12 +41,7 @@ def request_sync_impl( challenge: int = 0, retries: int = 0, ping: Optional[float] = None, -) -> Union[ - SourceInfo, - GoldSrcInfo, - List[Player], - Dict[Union[str, bytes], Union[str, bytes]], -]: +) -> Any: send_time = time.monotonic() resp_data = conn.request(a2s_proto.serialize_request(challenge)) recv_time = time.monotonic()