diff --git a/a2s/a2s_async.py b/a2s/a2s_async.py index 73ac321..29718c3 100644 --- a/a2s/a2s_async.py +++ b/a2s/a2s_async.py @@ -27,7 +27,7 @@ import asyncio import io import logging import time -from typing import TYPE_CHECKING, Dict, List, NoReturn, Optional, Tuple, Type, TypeVar, Union +from typing import TYPE_CHECKING, Dict, List, NoReturn, Optional, Tuple, Type, TypeVar, Union, overload from a2s.a2s_fragment import A2SFragment, decode_fragment from a2s.byteio import ByteReader @@ -48,7 +48,28 @@ PROTOCOLS = Union[InfoProtocol, PlayersProtocol, RulesProtocol] logger: logging.Logger = logging.getLogger("a2s") -T = TypeVar("T", bound=PROTOCOLS) +T = TypeVar("T", 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[str, str]: + ... async def request_async( @@ -60,6 +81,42 @@ async def request_async( 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[str, str]: + ... + + async def request_async_impl( conn: A2SStreamAsync, encoding: str, diff --git a/a2s/a2s_async.pyi b/a2s/a2s_async.pyi deleted file mode 100644 index 2f95e11..0000000 --- a/a2s/a2s_async.pyi +++ /dev/null @@ -1,71 +0,0 @@ -""" -MIT License - -Copyright (c) 2020 Gabriel Huber - -Permission is hereby granted, free of charge, to any person obtaining a copy -of this software and associated documentation files (the "Software"), to deal -in the Software without restriction, including without limitation the rights -to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -copies of the Software, and to permit persons to whom the Software is -furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in all -copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -SOFTWARE. -""" - -from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Type, Union, overload - -if TYPE_CHECKING: - from .a2s_async import A2SStreamAsync - from .info import GoldSrcInfo, InfoProtocol, SourceInfo - from .players import Player, PlayersProtocol - from .rules import 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[str, str]: ... -@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[str, str]: ... diff --git a/a2s/a2s_sync.py b/a2s/a2s_sync.py index cf5f5e2..b26ace8 100644 --- a/a2s/a2s_sync.py +++ b/a2s/a2s_sync.py @@ -28,7 +28,7 @@ import io import logging import socket import time -from typing import Dict, List, Optional, Tuple, Type, TypeVar, Union +from typing import Dict, List, Optional, Tuple, Type, TypeVar, Union, overload from a2s.a2s_fragment import decode_fragment from a2s.byteio import ByteReader @@ -48,16 +48,71 @@ logger: logging.Logger = logging.getLogger("a2s") T = TypeVar("T", InfoProtocol, RulesProtocol, PlayersProtocol) +__all__ = ("A2SStream",) + + +@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[str, str]: + ... + def request_sync( address: Tuple[str, int], timeout: float, encoding: str, a2s_proto: Type[T] ) -> Union[List[Player], GoldSrcInfo, SourceInfo, Dict[str, str]]: conn = A2SStream(address, timeout) - response = request_sync_impl(conn, encoding, a2s_proto) # type: ignore + 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[str, str]: + ... + + def request_sync_impl( conn: A2SStream, encoding: str, a2s_proto: Type[T], challenge: int = 0, retries: int = 0, ping: Optional[float] = None ) -> Union[SourceInfo, GoldSrcInfo, Dict[str, str], List[Player]]: diff --git a/a2s/a2s_sync.pyi b/a2s/a2s_sync.pyi deleted file mode 100644 index 6946647..0000000 --- a/a2s/a2s_sync.pyi +++ /dev/null @@ -1,71 +0,0 @@ -""" -MIT License - -Copyright (c) 2020 Gabriel Huber - -Permission is hereby granted, free of charge, to any person obtaining a copy -of this software and associated documentation files (the "Software"), to deal -in the Software without restriction, including without limitation the rights -to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -copies of the Software, and to permit persons to whom the Software is -furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in all -copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -SOFTWARE. -""" - -from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Type, Union, overload - -if TYPE_CHECKING: - from .a2s_sync import A2SStream - from .info import GoldSrcInfo, InfoProtocol, SourceInfo - from .players import Player, PlayersProtocol - from .rules import RulesProtocol - -@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[str, str]: ... -@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[str, str]: ...