Provide Unpack Args & Improved typehints

aiodns/__init__.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import asyncio
 import functools
 import logging
@@ -10,9 +12,8 @@
     Any,
     Callable,
     Literal,
-    Optional,
+    TypedDict,
     TypeVar,
-    Union,
     overload,
 )
 
@@ -56,14 +57,74 @@
     'ANY': pycares.QUERY_CLASS_ANY,
 }
 
+if sys.version_info >= (3, 11):
+    from typing import Unpack
+
+    class DNSResolverKwargs(TypedDict):
+        flags: int | None
+        timeout: float | None
+        tries: int | None
+        ndots: int | None
+        tcp_port: int | None
+        udp_port: int | None
+        domains: Sequence[str] | None
+        lookups: str | None
+        socket_send_buffer_size: int | None
+        socket_receive_buffer_size: int | None
+        rotate: bool
+        local_ip: str | None
+        local_dev: str | None
+        resolvconf_path: str | None
+
 
 class DNSResolver:
+    @overload
     def __init__(
         self,
-        nameservers: Optional[Sequence[str]] = None,
-        loop: Optional[asyncio.AbstractEventLoop] = None,
+        nameservers: Sequence[str] | None = None,
+        loop: asyncio.AbstractEventLoop | None = None,
+    ) -> None: ...
+
+    if sys.version_info >= (3, 11):
+
+        @overload
+        def __init__(
+            self,
+            nameservers: Sequence[str] | None = None,
+            loop: asyncio.AbstractEventLoop | None = None,
+            **kwargs: Unpack[DNSResolverKwargs],
+        ) -> None: ...
+    else:
+        # Reserve backwards compatability for older versions
+        # of Python
+        @overload
+        def __init__(
+            self,
+            nameservers: Sequence[str] | None = None,
+            loop: asyncio.AbstractEventLoop | None = None,
+            *,
+            flags: int | None = None,
+            timeout: float | None = None,
+            tries: int | None = None,
+            ndots: int | None = None,
+            tcp_port: int | None = None,
+            udp_port: int | None = None,
+            domains: Sequence[str] | None = None,
+            lookups: str | None = None,
+            socket_send_buffer_size: int | None = None,
+            socket_receive_buffer_size: int | None = None,
+            rotate: bool = False,
+            local_ip: str | None = None,
+            local_dev: str | None = None,
+            resolvconf_path: str | None = None,
+        ) -> None: ...
+
+    def __init__(
+        self,
+        nameservers: Sequence[str] | None = None,
+        loop: asyncio.AbstractEventLoop | None = None,
         **kwargs: Any,
-    ) -> None:  # TODO(PY311): Use Unpack for kwargs.
+    ) -> None:
         self._closed = True
         self.loop = loop or asyncio.get_event_loop()
         if TYPE_CHECKING:
@@ -76,7 +137,7 @@ def __init__(
             self.nameservers = nameservers
         self._read_fds: set[int] = set()
         self._write_fds: set[int] = set()
-        self._timer: Optional[asyncio.TimerHandle] = None
+        self._timer: asyncio.TimerHandle | None = None
         self._closed = False
 
     def _make_channel(self, **kwargs: Any) -> tuple[bool, pycares.Channel]:
@@ -125,12 +186,12 @@ def nameservers(self) -> Sequence[str]:
         return self._channel.servers
 
     @nameservers.setter
-    def nameservers(self, value: Iterable[Union[str, bytes]]) -> None:
+    def nameservers(self, value: Iterable[str | bytes]) -> None:
         self._channel.servers = value
 
     @staticmethod
     def _callback(
-        fut: asyncio.Future[_T], result: _T, errorno: Optional[int]
+        fut: asyncio.Future[_T], result: _T, errorno: int | None
     ) -> None:
         if fut.cancelled():
             return
@@ -143,7 +204,7 @@ def _callback(
 
     def _get_future_callback(
         self,
-    ) -> tuple['asyncio.Future[_T]', Callable[[_T, int], None]]:
+    ) -> tuple[asyncio.Future[_T], Callable[[_T, int], None]]:
         """Return a future and a callback to set the result of the future."""
         cb: Callable[[_T, int], None]
         future: asyncio.Future[_T] = self.loop.create_future()
@@ -159,52 +220,52 @@ def _get_future_callback(
 
     @overload
     def query(
-        self, host: str, qtype: Literal['A'], qclass: Optional[str] = ...
+        self, host: str, qtype: Literal['A'], qclass: str | None = ...
     ) -> asyncio.Future[list[pycares.ares_query_a_result]]: ...
     @overload
     def query(
-        self, host: str, qtype: Literal['AAAA'], qclass: Optional[str] = ...
+        self, host: str, qtype: Literal['AAAA'], qclass: str | None = ...
     ) -> asyncio.Future[list[pycares.ares_query_aaaa_result]]: ...
     @overload
     def query(
-        self, host: str, qtype: Literal['CAA'], qclass: Optional[str] = ...
+        self, host: str, qtype: Literal['CAA'], qclass: str | None = ...
     ) -> asyncio.Future[list[pycares.ares_query_caa_result]]: ...
     @overload
     def query(
-        self, host: str, qtype: Literal['CNAME'], qclass: Optional[str] = ...
+        self, host: str, qtype: Literal['CNAME'], qclass: str | None = ...
     ) -> asyncio.Future[pycares.ares_query_cname_result]: ...
     @overload
     def query(
-        self, host: str, qtype: Literal['MX'], qclass: Optional[str] = ...
+        self, host: str, qtype: Literal['MX'], qclass: str | None = ...
     ) -> asyncio.Future[list[pycares.ares_query_mx_result]]: ...
     @overload
     def query(
-        self, host: str, qtype: Literal['NAPTR'], qclass: Optional[str] = ...
+        self, host: str, qtype: Literal['NAPTR'], qclass: str | None = ...
     ) -> asyncio.Future[list[pycares.ares_query_naptr_result]]: ...
     @overload
     def query(
-        self, host: str, qtype: Literal['NS'], qclass: Optional[str] = ...
+        self, host: str, qtype: Literal['NS'], qclass: str | None = ...
     ) -> asyncio.Future[list[pycares.ares_query_ns_result]]: ...
     @overload
     def query(
-        self, host: str, qtype: Literal['PTR'], qclass: Optional[str] = ...
+        self, host: str, qtype: Literal['PTR'], qclass: str | None = ...
     ) -> asyncio.Future[list[pycares.ares_query_ptr_result]]: ...
     @overload
     def query(
-        self, host: str, qtype: Literal['SOA'], qclass: Optional[str] = ...
+        self, host: str, qtype: Literal['SOA'], qclass: str | None = ...
     ) -> asyncio.Future[pycares.ares_query_soa_result]: ...
     @overload
     def query(
-        self, host: str, qtype: Literal['SRV'], qclass: Optional[str] = ...
+        self, host: str, qtype: Literal['SRV'], qclass: str | None = ...
     ) -> asyncio.Future[list[pycares.ares_query_srv_result]]: ...
     @overload
     def query(
-        self, host: str, qtype: Literal['TXT'], qclass: Optional[str] = ...
+        self, host: str, qtype: Literal['TXT'], qclass: str | None = ...
     ) -> asyncio.Future[list[pycares.ares_query_txt_result]]: ...
 
     def query(
-        self, host: str, qtype: str, qclass: Optional[str] = None
-    ) -> Union[asyncio.Future[list[Any]], asyncio.Future[Any]]:
+        self, host: str, qtype: str, qclass: str | None = None
+    ) -> asyncio.Future[list[Any]] | asyncio.Future[Any]:
         try:
             qtype = query_type_map[qtype]
         except KeyError as e:
@@ -215,7 +276,7 @@ def query(
             except KeyError as e:
                 raise ValueError(f'invalid query class: {qclass}') from e
 
-        fut: Union[asyncio.Future[list[Any]], asyncio.Future[Any]]
+        fut: asyncio.Future[list[Any]] | asyncio.Future[Any]
         fut, cb = self._get_future_callback()
         self._channel.query(host, qtype, cb, query_class=qclass)
         return fut
@@ -232,7 +293,7 @@ def getaddrinfo(
         self,
         host: str,
         family: socket.AddressFamily = socket.AF_UNSPEC,
-        port: Optional[int] = None,
+        port: int | None = None,
         proto: int = 0,
         type: int = 0,
         flags: int = 0,
@@ -246,7 +307,7 @@ def getaddrinfo(
 
     def getnameinfo(
         self,
-        sockaddr: Union[tuple[str, int], tuple[str, int, int, int]],
+        sockaddr: tuple[str, int] | tuple[str, int, int, int],
         flags: int = 0,
     ) -> asyncio.Future[pycares.ares_nameinfo_result]:
         fut: asyncio.Future[pycares.ares_nameinfo_result]
@@ -345,15 +406,15 @@ async def close(self) -> None:
         """
         self._cleanup()
 
-    async def __aenter__(self) -> 'DNSResolver':
+    async def __aenter__(self) -> DNSResolver:
         """Enter the async context manager."""
         return self
 
     async def __aexit__(
         self,
-        exc_type: Optional[type[BaseException]],
-        exc_val: Optional[BaseException],
-        exc_tb: Optional[TracebackType],
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
     ) -> None:
         """Exit the async context manager."""
         await self.close()