asyncio alternative to deferToThread() (#7349)

scrapy/extensions/feedexport.py

@@ -21,14 +21,13 @@ from typing import IO, TYPE_CHECKING, Any, Protocol, TypeAlias, cast
 from urllib.parse import unquote, urlparse
 
 from twisted.internet.defer import Deferred, DeferredList
-from twisted.internet.threads import deferToThread
 from w3lib.url import file_uri_to_path
 from zope.interface import Interface, implementer
 
 from scrapy import Spider, signals
 from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
 from scrapy.extensions.postprocessing import PostProcessingManager
-from scrapy.utils.asyncio import is_asyncio_available
+from scrapy.utils.asyncio import is_asyncio_available, run_in_thread
 from scrapy.utils.conf import feed_complete_default_values_from_settings
 from scrapy.utils.defer import deferred_from_coro, ensure_awaitable
 from scrapy.utils.ftp import ftp_store_file
@@ -131,7 +130,7 @@ class BlockingFeedStorage(ABC):
         return NamedTemporaryFile(prefix="feed-", dir=path)
 
     def store(self, file: IO[bytes]) -> Deferred[None] | None:
-        return deferToThread(self._store_in_thread, file)
+        return deferred_from_coro(run_in_thread(self._store_in_thread, file))
 
     @abstractmethod
     def _store_in_thread(self, file: IO[bytes]) -> None:

scrapy/pipelines/files.py

@@ -23,15 +23,15 @@ from urllib.parse import urlparse
 
 from itemadapter import ItemAdapter
 from twisted.internet.defer import Deferred, maybeDeferred
-from twisted.internet.threads import deferToThread
 
 from scrapy.exceptions import IgnoreRequest, NotConfigured, ScrapyDeprecationWarning
 from scrapy.http import Request, Response
 from scrapy.http.request import NO_CALLBACK
 from scrapy.pipelines.media import FileInfo, FileInfoOrError, MediaPipeline
+from scrapy.utils.asyncio import run_in_thread
 from scrapy.utils.boto import is_botocore_available
 from scrapy.utils.datatypes import CaseInsensitiveDict
-from scrapy.utils.defer import ensure_awaitable
+from scrapy.utils.defer import deferred_from_coro, ensure_awaitable
 from scrapy.utils.ftp import ftp_store_file
 from scrapy.utils.log import failure_to_exc_info
 from scrapy.utils.python import to_bytes
@@ -198,13 +198,12 @@ class S3FilesStore:
 
     def _get_boto_key(self, path: str) -> Deferred[dict[str, Any]]:
         key_name = f"{self.prefix}{path}"
-        return cast(
-            "Deferred[dict[str, Any]]",
-            deferToThread(
+        return deferred_from_coro(
+            run_in_thread(
                 self.s3_client.head_object,  # type: ignore[attr-defined]
                 Bucket=self.bucket,
                 Key=key_name,
-            ),
+            )
         )
 
     def persist_file(
@@ -221,14 +220,16 @@ class S3FilesStore:
         extra = self._headers_to_botocore_kwargs(self.HEADERS)
         if headers:
             extra.update(self._headers_to_botocore_kwargs(headers))
-        return deferToThread(
-            self.s3_client.put_object,  # type: ignore[attr-defined]
-            Bucket=self.bucket,
-            Key=key_name,
-            Body=buf,
-            Metadata={k: str(v) for k, v in (meta or {}).items()},
-            ACL=self.POLICY,
-            **extra,
+        return deferred_from_coro(
+            run_in_thread(
+                self.s3_client.put_object,  # type: ignore[attr-defined]
+                Bucket=self.bucket,
+                Key=key_name,
+                Body=buf,
+                Metadata={k: str(v) for k, v in (meta or {}).items()},
+                ACL=self.POLICY,
+                **extra,
+            )
         )
 
     def _headers_to_botocore_kwargs(self, headers: dict[str, Any]) -> dict[str, Any]:
@@ -315,10 +316,9 @@ class GCSFilesStore:
             return {}
 
         blob_path = self._get_blob_path(path)
-        return cast(
-            "Deferred[StatInfo]",
-            deferToThread(self.bucket.get_blob, blob_path).addCallback(_onsuccess),
-        )
+        return deferred_from_coro(
+            run_in_thread(self.bucket.get_blob, blob_path)
+        ).addCallback(_onsuccess)
 
     def _get_content_type(self, headers: dict[str, str] | None) -> str:
         if headers and "Content-Type" in headers:
@@ -340,11 +340,13 @@ class GCSFilesStore:
         blob = self.bucket.blob(blob_path)
         blob.cache_control = self.CACHE_CONTROL
         blob.metadata = {k: str(v) for k, v in (meta or {}).items()}
-        return deferToThread(
-            blob.upload_from_string,
-            data=buf.getvalue(),
-            content_type=self._get_content_type(headers),
-            predefined_acl=self.POLICY,
+        return deferred_from_coro(
+            run_in_thread(
+                blob.upload_from_string,
+                data=buf.getvalue(),
+                content_type=self._get_content_type(headers),
+                predefined_acl=self.POLICY,
+            )
         )
 
 
@@ -377,15 +379,17 @@ class FTPFilesStore:
         headers: dict[str, str] | None = None,
     ) -> Deferred[Any]:
         path = f"{self.basedir}/{path}"
-        return deferToThread(
-            ftp_store_file,
-            path=path,
-            file=buf,
-            host=self.host,
-            port=self.port,
-            username=self.username,
-            password=self.password,
-            use_active_mode=self.USE_ACTIVE_MODE,
+        return deferred_from_coro(
+            run_in_thread(
+                ftp_store_file,
+                path=path,
+                file=buf,
+                host=self.host,
+                port=self.port,
+                username=self.username,
+                password=self.password,
+                use_active_mode=bool(self.USE_ACTIVE_MODE),
+            )
         )
 
     def stat_file(
@@ -407,7 +411,7 @@ class FTPFilesStore:
             except Exception:
                 return {}
 
-        return cast("Deferred[StatInfo]", deferToThread(_stat_file, path))
+        return deferred_from_coro(run_in_thread(_stat_file, path))
 
 
 class FilesPipeline(MediaPipeline):

scrapy/utils/asyncio.py

@@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar
 
 from twisted.internet.defer import Deferred
 from twisted.internet.task import LoopingCall
+from twisted.internet.threads import deferToThread
 
 from scrapy.utils.asyncgen import as_async_generator
 from scrapy.utils.reactor import is_asyncio_reactor_installed, is_reactor_installed
@@ -278,3 +279,23 @@ class CallLaterResult:
         elif self._delayed_call and self._delayed_call.active():
             self._delayed_call.cancel()
             self._delayed_call = None
+
+
+async def run_in_thread(
+    func: Callable[_P, _T], *args: _P.args, **kwargs: _P.kwargs
+) -> _T:
+    """Call a function in a thread and return its result as a coroutine.
+
+    This uses either :func:`asyncio.to_thread` or
+    :func:`twisted.internet.threads.deferToThread`, depending on whether
+    asyncio support is available.
+
+    .. versionadded:: VERSION
+    """
+    if is_asyncio_available():
+        return await asyncio.to_thread(func, *args, **kwargs)
+
+    # circular import
+    from scrapy.utils.defer import maybe_deferred_to_future  # noqa: PLC0415
+
+    return await maybe_deferred_to_future(deferToThread(func, *args, **kwargs))

scrapy/utils/decorators.py

@@ -6,9 +6,10 @@ from functools import wraps
 from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, overload
 
 from twisted.internet.defer import Deferred, maybeDeferred
-from twisted.internet.threads import deferToThread
 
 from scrapy.exceptions import ScrapyDeprecationWarning
+from scrapy.utils.asyncio import run_in_thread
+from scrapy.utils.defer import deferred_from_coro
 
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator, Callable, Coroutine
@@ -59,12 +60,15 @@ def defers(func: Callable[_P, _T]) -> Callable[_P, Deferred[_T]]:  # pragma: no
 
 def inthread(func: Callable[_P, _T]) -> Callable[_P, Deferred[_T]]:
     """Decorator to call a function in a thread and return a deferred with the
-    result
+    result.
+
+    .. versionchanged:: VERSION
+        Now uses :func:`asyncio.to_thread` if the asyncio support is available.
     """
 
     @wraps(func)
     def wrapped(*a: _P.args, **kw: _P.kwargs) -> Deferred[_T]:
-        return deferToThread(func, *a, **kw)
+        return deferred_from_coro(run_in_thread(func, *a, **kw))
 
     return wrapped
 

tests/test_feedexport.py

@@ -1151,7 +1151,6 @@ class TestFeedExport(TestFeedExportBase):
         data = await self.exported_no_data(settings)
         assert data["csv"] == b""
 
-    @pytest.mark.requires_reactor  # TODO: needs a reactor for BlockingFeedStorage
     @coroutine_test
     async def test_multiple_feeds_success_logs_blocking_feed_storage(self):
         settings = {

tests/test_feedexport_batch.py

@@ -381,7 +381,6 @@ class TestBatchDeliveries(TestFeedExportBase):
         assert "feedexport/success_count/FileFeedStorage" in crawler.stats.get_stats()
         assert crawler.stats.get_value("feedexport/success_count/FileFeedStorage") == 12
 
-    @pytest.mark.requires_reactor  # TODO: needs a reactor for BlockingFeedStorage
     @pytest.mark.requires_boto3
     @inline_callbacks_test
     def test_s3_export(self):

tests/test_feedexport_storages.py

@@ -116,7 +116,6 @@ class TestFileFeedStorage:
         assert storage.path == path
 
 
-@pytest.mark.requires_reactor  # TODO: needs a reactor for BlockingFeedStorage
 class TestFTPFeedStorage:
     def get_test_spider(self, settings=None):
         class TestSpider(scrapy.Spider):
@@ -231,7 +230,6 @@ class TestBlockingFeedStorage:
 
 
 @pytest.mark.requires_boto3
-@pytest.mark.requires_reactor  # TODO: needs a reactor for BlockingFeedStorage
 class TestS3FeedStorage:
     def test_parse_credentials(self):
         aws_credentials = {
@@ -461,7 +459,6 @@ class TestS3FeedStorage:
         assert "S3 does not support appending to files" in str(log)
 
 
-@pytest.mark.requires_reactor  # TODO: needs a reactor for BlockingFeedStorage
 class TestGCSFeedStorage:
     def test_parse_settings(self):
         try:

tests/test_pipeline_files.py

@@ -584,7 +584,6 @@ class TestFilesPipelineCustomSettings:
         assert fs_store.basedir == str(tmp_path)
 
 
-@pytest.mark.requires_reactor  # TODO: needs a reactor for S3FilesStore
 @pytest.mark.requires_botocore
 class TestS3FilesStore:
     @inline_callbacks_test
@@ -715,7 +714,6 @@ class TestGCSFilesStore:
             store.bucket.get_blob.assert_called_with(expected_blob_path)
 
 
-@pytest.mark.requires_reactor  # TODO: needs a reactor for FTPFilesStore
 class TestFTPFileStore:
     @inline_callbacks_test
     def test_persist(self):