Use modern type hints (#227)

proxy_scraper_checker/__main__.py

@@ -3,12 +3,11 @@ from __future__ import annotations
 import asyncio
 import logging
 import sys
-from typing import TYPE_CHECKING, Callable, Coroutine, Dict, Mapping
+from typing import TYPE_CHECKING, Callable, Coroutine, Mapping
 
 import aiofiles
 import rich.traceback
 from aiohttp import ClientSession, TCPConnector
-from aiohttp_socks import ProxyType
 from rich.console import Console
 from rich.logging import RichHandler
 from rich.progress import BarColumn, MofNCompleteColumn, Progress, TextColumn
@@ -29,6 +28,9 @@ if sys.version_info >= (3, 11):
 else:
     import tomli as tomllib
 
+if TYPE_CHECKING:
+    from aiohttp_socks import ProxyType
+
 logger = logging.getLogger(__name__)
 
 T = TypeVar("T")
@@ -62,7 +64,7 @@ def get_async_run() -> Callable[[Coroutine[Any, Any, T]], T]:
     return asyncio.run
 
 
-async def read_config(file: str, /) -> Dict[str, Any]:
+async def read_config(file: str, /) -> dict[str, Any]:
     async with aiofiles.open(file, "rb") as f:
         content = await f.read()
     return tomllib.loads(utils.bytes_decode(content))

proxy_scraper_checker/checker.py

@@ -2,15 +2,17 @@ from __future__ import annotations
 
 import asyncio
 import logging
-from typing import Mapping
-
-from aiohttp_socks import ProxyType
-from rich.progress import Progress, TaskID
+from typing import TYPE_CHECKING, Mapping
 
 from . import sort
-from .proxy import Proxy
-from .settings import Settings
-from .storage import ProxyStorage
+
+if TYPE_CHECKING:
+    from aiohttp_socks import ProxyType
+    from rich.progress import Progress, TaskID
+
+    from .proxy import Proxy
+    from .settings import Settings
+    from .storage import ProxyStorage
 
 logger = logging.getLogger(__name__)
 

proxy_scraper_checker/fs.py

@@ -1,12 +1,15 @@
 from __future__ import annotations
 
 import logging
-from pathlib import Path
+from typing import TYPE_CHECKING
 
 import platformdirs
 
 from .utils import asyncify
 
+if TYPE_CHECKING:
+    from pathlib import Path
+
 logger = logging.getLogger(__name__)
 CACHE_PATH = platformdirs.user_cache_path("proxy_scraper_checker")
 

proxy_scraper_checker/geodb.py

@@ -1,17 +1,20 @@
 from __future__ import annotations
 
-import asyncio
 import logging
 import stat
-from typing import Optional
+from typing import TYPE_CHECKING
 
 import aiofiles
 from aiohttp import ClientResponse, ClientSession, hdrs
-from rich.progress import Progress, TaskID
 
 from . import fs
 from .utils import IS_DOCKER, asyncify, bytes_decode
 
+if TYPE_CHECKING:
+    import asyncio
+
+    from rich.progress import Progress, TaskID
+
 logger = logging.getLogger(__name__)
 
 GEODB_URL = "https://raw.githubusercontent.com/P3TERX/GeoLite.mmdb/download/GeoLite2-City.mmdb"
@@ -19,7 +22,7 @@ GEODB_PATH = fs.CACHE_PATH / "geolocation_database.mmdb"
 GEODB_ETAG_PATH = GEODB_PATH.with_suffix(".mmdb.etag")
 
 
-async def _read_etag() -> Optional[str]:
+async def _read_etag() -> str | None:
     try:
         await fs.async_add_permission(GEODB_ETAG_PATH, stat.S_IRUSR)
         async with aiofiles.open(GEODB_ETAG_PATH, "rb") as etag_file:

proxy_scraper_checker/output.py

@@ -4,18 +4,20 @@ import json
 import logging
 import stat
 from shutil import rmtree
-from typing import Sequence, Union
+from typing import TYPE_CHECKING, Sequence
 
 import maxminddb
 
 from . import fs, sort
 from .geodb import GEODB_PATH
 from .null_context import NullContext
-from .proxy import Proxy
-from .settings import Settings
-from .storage import ProxyStorage
 from .utils import IS_DOCKER, asyncify
 
+if TYPE_CHECKING:
+    from .proxy import Proxy
+    from .settings import Settings
+    from .storage import ProxyStorage
+
 logger = logging.getLogger(__name__)
 
 
@@ -35,8 +37,8 @@ def save_proxies(*, settings: Settings, storage: ProxyStorage) -> None:
     if settings.output_json:
         if settings.enable_geolocation:
             fs.add_permission(GEODB_PATH, stat.S_IRUSR)
-            mmdb: Union[maxminddb.Reader, NullContext] = (
-                maxminddb.open_database(GEODB_PATH)
+            mmdb: maxminddb.Reader | NullContext = maxminddb.open_database(
+                GEODB_PATH
             )
         else:
             mmdb = NullContext()

proxy_scraper_checker/proxy.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import json
 from io import StringIO
 from time import perf_counter
-from typing import Optional
 
 import attrs
 from aiohttp import ClientSession
@@ -32,10 +31,10 @@ class Proxy:
     protocol: ProxyType
     host: str
     port: int
-    username: Optional[str]
-    password: Optional[str]
+    username: str | None
+    password: str | None
     timeout: float = attrs.field(init=False, eq=False)
-    exit_ip: Optional[str] = attrs.field(init=False, eq=False)
+    exit_ip: str | None = attrs.field(init=False, eq=False)
 
     async def check(self, *, settings: Settings) -> None:
         async with settings.semaphore:

proxy_scraper_checker/scraper.py

@@ -3,19 +3,23 @@ from __future__ import annotations
 import asyncio
 import itertools
 import logging
+from typing import TYPE_CHECKING
 
 import aiofiles
 from aiohttp import ClientResponseError, ClientSession, ClientTimeout
 from aiohttp_socks import ProxyType
-from rich.progress import Progress, TaskID
 
 from .http import get_response_text
 from .parsers import PROXY_REGEX
 from .proxy import Proxy
-from .settings import Settings
-from .storage import ProxyStorage
 from .utils import bytes_decode, is_http_url
 
+if TYPE_CHECKING:
+    from rich.progress import Progress, TaskID
+
+    from .settings import Settings
+    from .storage import ProxyStorage
+
 logger = logging.getLogger(__name__)
 
 

proxy_scraper_checker/settings.py

@@ -9,17 +9,7 @@ import stat
 import sys
 from pathlib import Path
 from types import MappingProxyType
-from typing import (
-    TYPE_CHECKING,
-    Callable,
-    Dict,
-    FrozenSet,
-    Iterable,
-    Mapping,
-    Optional,
-    Tuple,
-    Union,
-)
+from typing import TYPE_CHECKING, Callable, Iterable, Mapping
 from urllib.parse import urlparse
 
 import attrs
@@ -40,7 +30,7 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-def _get_supported_max_connections() -> Optional[int]:
+def _get_supported_max_connections() -> int | None:
     if sys.platform == "win32":
         if isinstance(
             asyncio.get_event_loop_policy(),
@@ -69,7 +59,7 @@ def _get_supported_max_connections() -> Optional[int]:
     return soft_limit
 
 
-def _get_max_connections(value: int, /) -> Optional[int]:
+def _get_max_connections(value: int, /) -> int | None:
     if value < 0:
         msg = "max_connections must be non-negative"
         raise ValueError(msg)
@@ -92,9 +82,7 @@ def _get_max_connections(value: int, /) -> Optional[int]:
     return max_supported
 
 
-def _semaphore_converter(
-    value: int, /
-) -> Union[asyncio.Semaphore, NullContext]:
+def _semaphore_converter(value: int, /) -> asyncio.Semaphore | NullContext:
     v = _get_max_connections(value)
     return asyncio.Semaphore(v) if v else NullContext()
 
@@ -104,8 +92,8 @@ def _timeout_converter(value: float, /) -> ClientTimeout:
 
 
 def _sources_converter(
-    value: Mapping[ProxyType, Optional[Iterable[str]]], /
-) -> Dict[ProxyType, FrozenSet[str]]:
+    value: Mapping[ProxyType, Iterable[str] | None], /
+) -> dict[ProxyType, frozenset[str]]:
     return {
         proxy_type: frozenset(sources)
         for proxy_type, sources in value.items()
@@ -130,7 +118,7 @@ class CheckWebsiteType(enum.Enum):
         _: object,
         supports_geolocation: bool,  # noqa: FBT001
         supports_anonymity: bool,  # noqa: FBT001
-        headers: Optional[Mapping[str, str]],
+        headers: Mapping[str, str] | None,
         /,
     ) -> None:
         self.supports_geolocation = supports_geolocation
@@ -145,10 +133,12 @@ class CheckWebsiteType(enum.Enum):
 
 async def _get_check_website_type_and_real_ip(
     *, check_website: str, session: ClientSession
-) -> Union[
-    Tuple[Literal[CheckWebsiteType.UNKNOWN], None],
-    Tuple[Literal[CheckWebsiteType.PLAIN_IP, CheckWebsiteType.HTTPBIN_IP], str],
-]:
+) -> (
+    tuple[Literal[CheckWebsiteType.UNKNOWN], None]
+    | tuple[
+        Literal[CheckWebsiteType.PLAIN_IP, CheckWebsiteType.HTTPBIN_IP], str
+    ]
+):
     try:
         async with session.get(check_website) as response:
             content = await response.read()
@@ -201,17 +191,17 @@ class Settings:
     )
     output_path: Path = attrs.field(converter=Path)
     output_txt: bool = attrs.field(validator=attrs.validators.instance_of(bool))
-    real_ip: Optional[str] = attrs.field(
+    real_ip: str | None = attrs.field(
         validator=attrs.validators.optional(attrs.validators.instance_of(str))
     )
-    semaphore: Union[asyncio.Semaphore, NullContext] = attrs.field(
+    semaphore: asyncio.Semaphore | NullContext = attrs.field(
         converter=_semaphore_converter
     )
     sort_by_speed: bool = attrs.field(
         validator=attrs.validators.instance_of(bool)
     )
     source_timeout: float = attrs.field(validator=attrs.validators.gt(0))
-    sources: Dict[ProxyType, FrozenSet[str]] = attrs.field(
+    sources: dict[ProxyType, frozenset[str]] = attrs.field(
         validator=attrs.validators.and_(
             attrs.validators.instance_of(dict),
             attrs.validators.min_len(1),
@@ -235,7 +225,7 @@ class Settings:
     @property
     def sorting_key(
         self,
-    ) -> Union[Callable[[Proxy], float], Callable[[Proxy], Tuple[int, ...]]]:
+    ) -> Callable[[Proxy], float] | Callable[[Proxy], tuple[int, ...]]:
         return (
             sort.timeout_sort_key
             if self.sort_by_speed

proxy_scraper_checker/sort.py

@@ -1,19 +1,20 @@
 from __future__ import annotations
 
-from typing import Tuple
+from typing import TYPE_CHECKING
 
 from aiohttp_socks import ProxyType
 
-from .proxy import Proxy
+if TYPE_CHECKING:
+    from .proxy import Proxy
 
 PROTOCOL_ORDER = (ProxyType.HTTP, ProxyType.SOCKS4, ProxyType.SOCKS5)
 
 
-def protocol_sort_key(proxy: Proxy, /) -> Tuple[int, ProxyType]:
+def protocol_sort_key(proxy: Proxy, /) -> tuple[int, ProxyType]:
     return (PROTOCOL_ORDER.index(proxy.protocol), proxy.protocol)
 
 
-def natural_sort_key(proxy: Proxy, /) -> Tuple[int, ...]:
+def natural_sort_key(proxy: Proxy, /) -> tuple[int, ...]:
     return (proxy.protocol.value, *map(int, proxy.host.split(".")), proxy.port)
 
 

proxy_scraper_checker/storage.py

@@ -2,12 +2,14 @@ from __future__ import annotations
 
 import itertools
 from collections import Counter
-from typing import Dict, Iterable, Iterator, Set, Tuple
-
-from aiohttp_socks import ProxyType
+from typing import TYPE_CHECKING, Iterable, Iterator
 
 from . import sort
-from .proxy import Proxy
+
+if TYPE_CHECKING:
+    from aiohttp_socks import ProxyType
+
+    from .proxy import Proxy
 
 
 class ProxyStorage:
@@ -15,7 +17,7 @@ class ProxyStorage:
 
     def __init__(self, *, protocols: Iterable[ProxyType]) -> None:
         self.enabled_protocols = set(protocols)
-        self._proxies: Set[Proxy] = set()
+        self._proxies: set[Proxy] = set()
 
     def add(self, proxy: Proxy, /) -> None:
         self.enabled_protocols.add(proxy.protocol)
@@ -24,7 +26,7 @@ class ProxyStorage:
     def remove(self, proxy: Proxy, /) -> None:
         self._proxies.remove(proxy)
 
-    def get_grouped(self) -> Dict[ProxyType, Tuple[Proxy, ...]]:
+    def get_grouped(self) -> dict[ProxyType, tuple[Proxy, ...]]:
         key = sort.protocol_sort_key
         return {
             **{
@@ -40,7 +42,7 @@ class ProxyStorage:
             },
         }
 
-    def get_count(self) -> Dict[ProxyType, int]:
+    def get_count(self) -> dict[ProxyType, int]:
         return {
             **{
                 proto: 0

ruff.toml

@@ -43,13 +43,9 @@ ignore = [
   "S308",
   "S311",
   "SIM105",
-  "TCH001",
-  "TCH002",
-  "TCH003",
   "TID252",
   "TRY400",
 ]
-ignore-init-module-imports = true
 select = ["ALL"]
 
 [lint.flake8-self]
@@ -62,6 +58,3 @@ split-on-trailing-comma = false
 
 [lint.pydocstyle]
 convention = "google"
-
-[lint.pyupgrade]
-keep-runtime-typing = true