From bfe34492fa41d7d8284d671f15125a38dc4f8d92 Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Tue, 24 Mar 2026 19:41:15 +0500 Subject: [PATCH] asyncio alternative to deferToThread() (#7349) --- scrapy/extensions/feedexport.py | 5 +-- scrapy/pipelines/files.py | 70 ++++++++++++++++--------------- scrapy/utils/asyncio.py | 21 ++++++++++ scrapy/utils/decorators.py | 10 +++-- tests/test_feedexport.py | 1 - tests/test_feedexport_batch.py | 1 - tests/test_feedexport_storages.py | 3 -- tests/test_pipeline_files.py | 2 - 8 files changed, 67 insertions(+), 46 deletions(-) diff --git a/scrapy/extensions/feedexport.py b/scrapy/extensions/feedexport.py index 86593aba8..fa928678e 100644 --- a/scrapy/extensions/feedexport.py +++ b/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: diff --git a/scrapy/pipelines/files.py b/scrapy/pipelines/files.py index a0bb674d1..707207337 100644 --- a/scrapy/pipelines/files.py +++ b/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): diff --git a/scrapy/utils/asyncio.py b/scrapy/utils/asyncio.py index 9ec090995..7852a10fd 100644 --- a/scrapy/utils/asyncio.py +++ b/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)) diff --git a/scrapy/utils/decorators.py b/scrapy/utils/decorators.py index 044d1d4c7..f1e274190 100644 --- a/scrapy/utils/decorators.py +++ b/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 diff --git a/tests/test_feedexport.py b/tests/test_feedexport.py index e50b2be42..dcf412c0f 100644 --- a/tests/test_feedexport.py +++ b/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 = { diff --git a/tests/test_feedexport_batch.py b/tests/test_feedexport_batch.py index 16c4357bb..3b70e896c 100644 --- a/tests/test_feedexport_batch.py +++ b/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): diff --git a/tests/test_feedexport_storages.py b/tests/test_feedexport_storages.py index c5cd5c782..5102f8e37 100644 --- a/tests/test_feedexport_storages.py +++ b/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: diff --git a/tests/test_pipeline_files.py b/tests/test_pipeline_files.py index 201b29f83..c412645f4 100644 --- a/tests/test_pipeline_files.py +++ b/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):