Use warnings.deprecated

pyproject.toml

@@ -20,6 +20,7 @@ dependencies = [
     "runstats>=0.0.1",
     "tenacity>=8.2.0",
     "tqdm>=4.16.0",
+    'typing-extensions>=4.5.0; python_version < "3.13"',
     "w3lib>=2.1.1",
 ]
 classifiers = [

tox.ini

@@ -34,13 +34,17 @@ deps =
 
 [testenv:min]
 basepython = python3.10
-deps = {[min]deps}
+deps =
+    {[min]deps}
+    typing-extensions==4.5.0
 
 [testenv:min-x402]
 basepython = python3.10
 deps =
     {[min]deps}
     eth_account==0.13.7
+    # x402 pulls in pydantic, which requires a newer typing-extensions.
+    typing-extensions==4.12.2
     x402==0.1.1
 
 [testenv:mypy]

zyte_api/_retry.py

@@ -2,11 +2,11 @@
 
 import asyncio
 import logging
+import sys
 from collections import Counter
 from datetime import timedelta
 from itertools import count
-from typing import TYPE_CHECKING, Any, cast
-from warnings import warn
+from typing import TYPE_CHECKING
 
 from aiohttp import client_exceptions
 from tenacity import (
@@ -27,9 +27,12 @@
 
 from ._errors import RequestError
 
-if TYPE_CHECKING:
-    from collections.abc import Callable
+if sys.version_info >= (3, 13):
+    from warnings import deprecated as _deprecated
+else:
+    from typing_extensions import deprecated as _deprecated
 
+if TYPE_CHECKING:
     from tenacity.wait import wait_base
 
 logger = logging.getLogger(__name__)
@@ -169,14 +172,6 @@ def _402_error(exc: BaseException) -> bool:
     return isinstance(exc, RequestError) and exc.status == 402
 
 
-def _deprecated(message: str, callable_: Callable) -> Callable:
-    def wrapper(factory: Any, retry_state: RetryCallState) -> Callable:
-        warn(message, DeprecationWarning, stacklevel=3)
-        return callable_(retry_state=retry_state)
-
-    return wrapper
-
-
 class RetryFactory:
     """Factory class that builds the :class:`tenacity.AsyncRetrying` object
     that defines the :ref:`default retry policy <default-retry-policy>`.
@@ -235,29 +230,6 @@ class CustomRetryFactory(RetryFactory):
     )
     download_error_wait: wait_base = network_error_wait
 
-    temporary_download_error_stop: stop_base = cast(
-        "stop_base",
-        _deprecated(
-            (
-                "The zyte_api.RetryFactory.temporary_download_error_stop() method "
-                "is deprecated and will be removed in a future version. Use "
-                "download_error_stop() instead."
-            ),
-            download_error_stop,
-        ),
-    )
-    temporary_download_error_wait: wait_base = cast(
-        "wait_base",
-        _deprecated(
-            (
-                "The zyte_api.RetryFactory.temporary_download_error_wait() method "
-                "is deprecated and will be removed in a future version. Use "
-                "download_error_wait() instead."
-            ),
-            download_error_wait,
-        ),
-    )
-
     throttling_stop: stop_base = stop_never
 
     undocumented_error_stop: stop_base = stop_on_count(2)
@@ -266,6 +238,22 @@ class CustomRetryFactory(RetryFactory):
     x402_error_stop: stop_base = stop_on_count(2)
     x402_error_wait: wait_base = wait_none()
 
+    @_deprecated(
+        "The zyte_api.RetryFactory.temporary_download_error_stop() method is"
+        " deprecated and will be removed in a future version. Use"
+        " download_error_stop() instead."
+    )
+    def temporary_download_error_stop(self, retry_state: RetryCallState) -> bool:
+        return self.download_error_stop(retry_state=retry_state)
+
+    @_deprecated(
+        "The zyte_api.RetryFactory.temporary_download_error_wait() method is"
+        " deprecated and will be removed in a future version. Use"
+        " download_error_wait() instead."
+    )
+    def temporary_download_error_wait(self, retry_state: RetryCallState) -> float:
+        return self.download_error_wait(retry_state=retry_state)
+
     def wait(self, retry_state: RetryCallState) -> float:
         assert retry_state.outcome, "Unexpected empty outcome"
         exc = retry_state.outcome.exception()

zyte_api/_utils.py

@@ -1,31 +1,22 @@
+import sys
 from typing import Any
-from warnings import warn
 
 import aiohttp
 from aiohttp import TCPConnector
 
 from .constants import API_TIMEOUT
 
+if sys.version_info >= (3, 13):
+    from warnings import deprecated as _deprecated
+else:
+    from typing_extensions import deprecated as _deprecated
+
 # 120 seconds is probably too long, but we are concerned about the case with
 # many concurrent requests and some processing logic running in the same reactor,
 # thus, saturating the CPU. This will make timeouts more likely.
 _AIO_API_TIMEOUT = aiohttp.ClientTimeout(total=API_TIMEOUT + 120)
 
 
-def deprecated_create_session(
-    connection_pool_size: int = 100, **kwargs: Any
-) -> aiohttp.ClientSession:
-    warn(
-        (
-            "zyte_api.aio.client.create_session is deprecated, use "
-            "ZyteAPI.session or AsyncZyteAPI.session instead."
-        ),
-        DeprecationWarning,
-        stacklevel=2,
-    )
-    return create_session(connection_pool_size=connection_pool_size, **kwargs)
-
-
 def create_session(
     connection_pool_size: int = 100, **kwargs: Any
 ) -> aiohttp.ClientSession:
@@ -34,3 +25,9 @@ def create_session(
     if "connector" not in kwargs:
         kwargs["connector"] = TCPConnector(limit=connection_pool_size, force_close=True)
     return aiohttp.ClientSession(**kwargs)
+
+
+deprecated_create_session = _deprecated(
+    "zyte_api.aio.client.create_session is deprecated, use ZyteAPI.session or"
+    " AsyncZyteAPI.session instead."
+)(create_session)