Merge pull request #107 from amacfie/fix-implicit-optional

Fix "implicit optional", e.g. `arg: int = None`

Author: pogzyb

asyncwhois/__init__.py

@@ -50,7 +50,7 @@ def whois(
     authoritative_only: bool = False,  # todo: deprecate and remove this argument
     find_authoritative_server: bool = True,
     ignore_not_found: bool = False,
-    proxy_url: str = None,
+    proxy_url: Optional[str] = None,
     timeout: int = 10,
     tldextract_obj: TLDExtract = None,
 ) -> tuple[str, dict]:
@@ -148,7 +148,7 @@ async def aio_whois(
     authoritative_only: bool = False,  # todo: deprecate and remove this argument
     find_authoritative_server: bool = True,
     ignore_not_found: bool = False,
-    proxy_url: str = None,
+    proxy_url: Optional[str] = None,
     timeout: int = 10,
     tldextract_obj: TLDExtract = None,
 ) -> tuple[str, dict]:
@@ -267,7 +267,7 @@ def whois_domain(
     domain: str,
     authoritative_only: bool = False,
     ignore_not_found: bool = False,
-    proxy_url: str = None,
+    proxy_url: Optional[str] = None,
     timeout: int = 10,
     tldextract_obj: TLDExtract = None,
 ) -> DomainLookup:
@@ -304,7 +304,7 @@ async def aio_whois_domain(
     domain: str,
     authoritative_only: bool = False,
     ignore_not_found: bool = False,
-    proxy_url: str = None,
+    proxy_url: Optional[str] = None,
     timeout: int = 10,
     tldextract_obj: TLDExtract = None,
 ) -> DomainLookup:
@@ -398,7 +398,7 @@ async def aio_rdap_domain(
 def whois_ipv4(
     ipv4: Union[IPv4Address, str],
     authoritative_only: bool = False,
-    proxy_url: str = None,
+    proxy_url: Optional[str] = None,
     timeout: int = 10,
 ) -> NumberLookup:
     """
@@ -428,7 +428,7 @@ def whois_ipv4(
 async def aio_whois_ipv4(
     ipv4: Union[IPv4Address, str],
     authoritative_only: bool = False,
-    proxy_url: str = None,
+    proxy_url: Optional[str] = None,
     timeout: int = 10,
 ) -> NumberLookup:
     """
@@ -506,7 +506,7 @@ async def aio_rdap_ipv4(
 def whois_ipv6(
     ipv6: Union[IPv6Address, str],
     authoritative_only: bool = False,
-    proxy_url: str = None,
+    proxy_url: Optional[str] = None,
     timeout: int = 10,
 ) -> NumberLookup:
     """
@@ -536,7 +536,7 @@ def whois_ipv6(
 async def aio_whois_ipv6(
     ipv6: Union[IPv6Address, str],
     authoritative_only: bool = False,
-    proxy_url: str = None,
+    proxy_url: Optional[str] = None,
     timeout: int = 10,
 ) -> NumberLookup:
     """

asyncwhois/client.py

@@ -1,5 +1,5 @@
 import ipaddress
-from typing import Union, Any
+from typing import Union, Any, Optional
 
 from tldextract.tldextract import extract, TLDExtract
 import whodap
@@ -51,7 +51,7 @@ def __init__(
         authoritative_only: bool = False,
         find_authoritative_server: bool = True,
         ignore_not_found: bool = False,
-        proxy_url: str = None,
+        proxy_url: Optional[str] = None,
         whodap_client: whodap.DNSClient = None,
         timeout: int = 10,
         tldextract_obj: TLDExtract = None,
@@ -125,7 +125,7 @@ class NumberClient(Client):
     def __init__(
         self,
         authoritative_only: bool = False,
-        proxy_url: str = None,
+        proxy_url: Optional[str] = None,
         whodap_client: Union[whodap.IPv4Client, whodap.IPv6Client] = None,
         timeout: int = 10,
     ):

asyncwhois/query.py

@@ -2,7 +2,7 @@
 import ipaddress
 import re
 import socket
-from typing import Tuple, Generator, Union
+from typing import Tuple, Generator, Union, Optional
 from contextlib import contextmanager, asynccontextmanager
 
 from python_socks.sync import Proxy
@@ -21,7 +21,7 @@ class Query:
 
     def __init__(
         self,
-        proxy_url: str = None,
+        proxy_url: Optional[str] = None,
         timeout: int = 10,
         find_authoritative_server: bool = True,
     ):
@@ -39,7 +39,7 @@ def _find_match(regex: str, blob: str) -> str:
 
     @contextmanager
     def _create_connection(
-        self, address: Tuple[str, int], proxy_url: str = None
+        self, address: Tuple[str, int], proxy_url: Optional[str] = None
     ) -> Generator[socket.socket, None, None]:
         s = None
         try:
@@ -58,7 +58,7 @@ def _create_connection(
 
     @asynccontextmanager
     async def _aio_create_connection(
-        self, address: Tuple[str, int], proxy_url: str = None
+        self, address: Tuple[str, int], proxy_url: Optional[str] = None
     ) -> Generator[Tuple[asyncio.StreamReader, asyncio.StreamWriter], None, None]:
         # init
         reader, writer = None, None
@@ -107,7 +107,7 @@ async def _aio_send_and_recv(
                 result += received.decode("utf-8", errors="ignore")
         return result
 
-    def run(self, search_term: str, server: str = None) -> list[str]:
+    def run(self, search_term: str, server: Optional[str] = None) -> list[str]:
         """
         Submits the `search_term` to the WHOIS server and returns a list of query responses.
         """
@@ -133,7 +133,9 @@ def _continue_querying(current_server: str, next_server: str) -> bool:
             and not next_server.startswith("www.")
         )
 
-    async def aio_run(self, search_term: str, server: str = None) -> list[str]:
+    async def aio_run(
+        self, search_term: str, server: Optional[str] = None
+    ) -> list[str]:
         data = search_term + "\r\n"
         if not server:
             if ":" in data:  # ipv6
@@ -201,8 +203,8 @@ async def _aio_do_query(
 class DomainQuery(Query):
     def __init__(
         self,
-        server: str = None,
-        proxy_url: str = None,
+        server: Optional[str] = None,
+        proxy_url: Optional[str] = None,
         timeout: int = 10,
         find_authoritative_server: bool = True,
     ):
@@ -219,12 +221,14 @@ def _get_server_name(domain_name: str) -> Union[str, None]:
                 return server
         return None
 
-    def run(self, search_term: str, server: str = None) -> list[str]:
+    def run(self, search_term: str, server: Optional[str] = None) -> list[str]:
         if not server:
             server = self._get_server_name(search_term)
         return super().run(str(search_term), server)
 
-    async def aio_run(self, search_term: str, server: str = None) -> list[str]:
+    async def aio_run(
+        self, search_term: str, server: Optional[str] = None
+    ) -> list[str]:
         if not server:
             server = self._get_server_name(search_term)
         return await super().aio_run(str(search_term), server)
@@ -233,8 +237,8 @@ async def aio_run(self, search_term: str, server: str = None) -> list[str]:
 class NumberQuery(Query):
     def __init__(
         self,
-        server: str = None,
-        proxy_url: str = None,
+        server: Optional[str] = None,
+        proxy_url: Optional[str] = None,
         timeout: int = 10,
     ):
         super().__init__(proxy_url, timeout)
@@ -252,7 +256,7 @@ def _get_server_name(ip: Union[ipaddress.IPv4Address, ipaddress.IPv6Address]):
     def run(
         self,
         search_term: Union[ipaddress.IPv4Address, ipaddress.IPv6Address],
-        server: str = None,
+        server: Optional[str] = None,
     ) -> list[str]:
         if not server:
             server = self._get_server_name(search_term)
@@ -261,7 +265,7 @@ def run(
     async def aio_run(
         self,
         search_term: Union[ipaddress.IPv4Address, ipaddress.IPv6Address],
-        server: str = None,
+        server: Optional[str] = None,
     ) -> list[str]:
         if not server:
             server = self._get_server_name(search_term)

asyncwhois/tldparsers.py

@@ -1,5 +1,4 @@
-"""TLD-specific domain parser classes
-"""
+"""TLD-specific domain parser classes"""
 
 from datetime import datetime
 import re