Merge 654571a72d into ad43bf0c56

conftest.py

@@ -137,6 +137,8 @@ def pytest_runtest_setup(item):
         "uvloop",
         "botocore",
         "boto3",
+        "aiobotocore",
+        "aioboto3",
     ]
 
     for module in optional_deps:

docs/topics/feed-exports.rst

@@ -261,6 +261,12 @@ storage backend is: ``True``.
 
 This storage backend uses :ref:`delayed file delivery <delayed-file-delivery>`.
 
+.. versionchanged:: VERSION
+    When :ref:`asyncio support is available <using-asyncio>` and aioboto3_ is
+    installed (it is part of the :ref:`s3 <extras>` extra), feeds are uploaded
+    using genuinely-asynchronous I/O. Otherwise, the blocking boto3_ client is
+    run in a separate thread.
+
 
 .. _topics-feed-storage-gcs:
 
@@ -819,6 +825,8 @@ source spider in the feed URI:
 
 
 .. _URIs: https://en.wikipedia.org/wiki/Uniform_Resource_Identifier
+.. _aioboto3: https://github.com/terricain/aioboto3
 .. _Amazon S3: https://aws.amazon.com/s3/
+.. _boto3: https://github.com/boto/boto3
 .. _Canned ACL: https://docs.aws.amazon.com/AmazonS3/latest/userguide/acl-overview.html#canned-acl
 .. _Google Cloud Storage: https://cloud.google.com/storage/

docs/topics/media-pipeline.rst

@@ -302,6 +302,15 @@ For self-hosting you also might feel the need not to use SSL and not to verify S
 To reuse connections for as many files as you check or upload in parallel, set
 :setting:`AWS_MAX_POOL_CONNECTIONS` accordingly.
 
+.. versionchanged:: VERSION
+    When :ref:`asyncio support is available <using-asyncio>` and aiobotocore_ is
+    installed (it is part of the :ref:`s3 <extras>` extra), files are stat'ed and
+    uploaded using genuinely-asynchronous I/O. Otherwise, the blocking botocore_
+    client is run in a separate thread.
+
+.. _aiobotocore: https://github.com/aio-libs/aiobotocore
+.. _botocore: https://github.com/boto/botocore
+
 .. _canned ACLs: https://docs.aws.amazon.com/AmazonS3/latest/userguide/acl-overview.html#canned-acl
 .. _Minio: https://github.com/minio/minio
 .. _Zenko CloudServer: https://www.zenko.io/cloudserver/

pyproject.toml

@@ -70,7 +70,7 @@ images = ["Pillow>=8.3.2"]
 ipython = ["ipython>=8.15.0"]
 ptpython = ["ptpython>=3.0.23"]
 robotparser = ["robotexclusionrulesparser>=1.6.2"]
-s3 = ["boto3>=1.20.0"]
+s3 = ["aioboto3>=9.0.0", "boto3>=1.20.0"]
 twisted-http2 = ["Twisted[http2]>=21.7.0"]
 uvloop = [
     "uvloop>=0.16.0; platform_system != 'Windows' and implementation_name != 'pypy'",
@@ -169,6 +169,8 @@ ignore_errors = true
 # usually no type hints
 [[tool.mypy.overrides]]
 module = [
+    "aioboto3",
+    "aiobotocore.*",
     "bpython",
     "brotli",
     "brotlicffi",
@@ -314,6 +316,8 @@ markers = [
     "requires_uvloop: marks tests as only enabled when uvloop is known to be working",
     "requires_botocore: marks tests that need botocore (but not boto3)",
     "requires_boto3: marks tests that need botocore and boto3",
+    "requires_aiobotocore: marks tests that need aiobotocore (but not aioboto3)",
+    "requires_aioboto3: marks tests that need aiobotocore and aioboto3",
     "requires_mitmproxy: marks tests that need a mitmdump executable",
     "requires_internet: marks tests that need real Internet access",
 ]

scrapy/extensions/feedexport.py

@@ -28,7 +28,7 @@ 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, run_in_thread
-from scrapy.utils.boto import _get_max_pool_connections
+from scrapy.utils.boto import _get_max_pool_connections, is_aioboto3_available
 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
@@ -234,20 +234,23 @@ class S3FeedStorage(BlockingFeedStorage):
         self.region_name: str | None = region_name
         self.max_pool_connections: int | None = max_pool_connections
 
-        boto3_session = boto3.session.Session()
+        self._client_kwargs: dict[str, Any] = {
-        self.s3_client = boto3_session.client(
+            "aws_access_key_id": self.access_key,
-            "s3",
+            "aws_secret_access_key": self.secret_key,
-            aws_access_key_id=self.access_key,
+            "aws_session_token": self.session_token,
-            aws_secret_access_key=self.secret_key,
+            "endpoint_url": self.endpoint_url,
-            aws_session_token=self.session_token,
+            "region_name": self.region_name,
-            endpoint_url=self.endpoint_url,
+            "config": (
-            region_name=self.region_name,
-            config=(
                 Config(max_pool_connections=self.max_pool_connections)
                 if self.max_pool_connections is not None
                 else None
             ),
-        )
+        }
+
+        # Synchronous boto3 client, used when asyncio support or aioboto3 is not
+        # available (its calls are then run in a thread).
+        boto3_session = boto3.session.Session()
+        self.s3_client = boto3_session.client("s3", **self._client_kwargs)
 
         if feed_options and feed_options.get("overwrite", True) is False:
             logger.warning(
@@ -276,6 +279,28 @@ class S3FeedStorage(BlockingFeedStorage):
             feed_options=feed_options,
         )
 
+    def store(self, file: IO[bytes]) -> Deferred[None] | None:
+        if is_asyncio_available() and is_aioboto3_available():
+            return deferred_from_coro(self._store_async(file))
+        return super().store(file)
+
+    async def _store_async(self, file: IO[bytes]) -> None:
+        import aioboto3  # noqa: PLC0415
+
+        file.seek(0)
+        extra_args = {"ACL": self.acl} if self.acl else {}
+        session = aioboto3.Session()
+        try:
+            async with session.client("s3", **self._client_kwargs) as client:
+                await client.upload_fileobj(
+                    Fileobj=file,
+                    Bucket=self.bucketname,
+                    Key=self.keyname,
+                    ExtraArgs=extra_args,
+                )
+        finally:
+            file.close()
+
     def _store_in_thread(self, file: IO[bytes]) -> None:
         file.seek(0)
         try:

scrapy/pipelines/files.py

@@ -6,6 +6,7 @@ See documentation in topics/media-pipeline.rst
 
 from __future__ import annotations
 
+import asyncio
 import base64
 import functools
 import hashlib
@@ -34,9 +35,14 @@ from scrapy.pipelines.media import (
     MediaPipeline,
     _MediaRequestFiltered,
 )
-from scrapy.utils.asyncio import run_in_thread
+from scrapy.utils.asyncio import is_asyncio_available, run_in_thread
-from scrapy.utils.boto import _get_max_pool_connections, is_botocore_available
+from scrapy.utils.boto import (
+    _get_max_pool_connections,
+    is_aiobotocore_available,
+    is_botocore_available,
+)
 from scrapy.utils.datatypes import CaseInsensitiveDict
+from scrapy.utils.decorators import _warn_spider_arg
 from scrapy.utils.defer import deferred_from_coro, ensure_awaitable
 from scrapy.utils.ftp import ftp_store_file
 from scrapy.utils.httpobj import urlparse_cached
@@ -53,6 +59,7 @@ if TYPE_CHECKING:
     # typing.Self requires Python 3.11
     from typing_extensions import Self
 
+    from scrapy import Spider
     from scrapy.crawler import Crawler
     from scrapy.settings import BaseSettings
 
@@ -174,30 +181,64 @@ class S3FilesStore:
     def __init__(self, uri: str):
         if not is_botocore_available():
             raise NotConfigured("missing botocore library")
-        import botocore.session  # noqa: PLC0415
-        from botocore.config import Config  # noqa: PLC0415
-
-        config = (
-            Config(max_pool_connections=self.AWS_MAX_POOL_CONNECTIONS)
-            if self.AWS_MAX_POOL_CONNECTIONS is not None
-            else None
-        )
-        session = botocore.session.get_session()
-        self.s3_client = session.create_client(
-            "s3",
-            aws_access_key_id=self.AWS_ACCESS_KEY_ID,
-            aws_secret_access_key=self.AWS_SECRET_ACCESS_KEY,
-            aws_session_token=self.AWS_SESSION_TOKEN,
-            endpoint_url=self.AWS_ENDPOINT_URL,
-            region_name=self.AWS_REGION_NAME,
-            use_ssl=self.AWS_USE_SSL,
-            verify=self.AWS_VERIFY,
-            config=config,
-        )
         if not uri.startswith("s3://"):
             raise ValueError(f"Incorrect URI scheme in {uri}, expected 's3'")
         self.bucket, self.prefix = uri[5:].split("/", 1)
 
+        from botocore.config import Config  # noqa: PLC0415
+
+        self._client_kwargs: dict[str, Any] = {
+            "aws_access_key_id": self.AWS_ACCESS_KEY_ID,
+            "aws_secret_access_key": self.AWS_SECRET_ACCESS_KEY,
+            "aws_session_token": self.AWS_SESSION_TOKEN,
+            "endpoint_url": self.AWS_ENDPOINT_URL,
+            "region_name": self.AWS_REGION_NAME,
+            "use_ssl": self.AWS_USE_SSL,
+            "verify": self.AWS_VERIFY,
+            "config": (
+                Config(max_pool_connections=self.AWS_MAX_POOL_CONNECTIONS)
+                if self.AWS_MAX_POOL_CONNECTIONS is not None
+                else None
+            ),
+        }
+
+        # Synchronous botocore client, used when asyncio support or aiobotocore
+        # is not available (its calls are then run in a thread).
+        import botocore.session  # noqa: PLC0415
+
+        session = botocore.session.get_session()
+        self.s3_client = session.create_client("s3", **self._client_kwargs)
+
+        # Asynchronous aiobotocore client. It's created lazily (it must be
+        # instantiated from within the running event loop), reused across calls
+        # and closed by close().
+        self._aio_client: Any = None
+        self._aio_client_cm: Any = None
+        self._aio_client_lock = asyncio.Lock()
+
+    def _use_async(self) -> bool:
+        """Whether to use the genuinely-asynchronous aiobotocore client instead
+        of running the blocking botocore client in a thread."""
+        return is_asyncio_available() and is_aiobotocore_available()
+
+    async def _get_aio_client(self) -> Any:
+        async with self._aio_client_lock:
+            if self._aio_client is None:
+                from aiobotocore.session import get_session  # noqa: PLC0415
+
+                self._aio_client_cm = get_session().create_client(
+                    "s3", **self._client_kwargs
+                )
+                # pylint: disable-next=unnecessary-dunder-call
+                self._aio_client = await self._aio_client_cm.__aenter__()
+        return self._aio_client
+
+    async def close(self) -> None:
+        """Close the underlying aiobotocore client, if one was created."""
+        if self._aio_client is not None:
+            await self._aio_client_cm.__aexit__(None, None, None)
+            self._aio_client = self._aio_client_cm = None
+
     @staticmethod
     def _onsuccess(boto_key: dict[str, Any]) -> StatInfo:
         checksum = boto_key["ETag"].strip('"')
@@ -208,18 +249,20 @@ class S3FilesStore:
     def stat_file(
         self, path: str, info: MediaPipeline.SpiderInfo
     ) -> Deferred[StatInfo]:
+        return deferred_from_coro(self._stat_file(path))
 
-        return self._get_boto_key(path).addCallback(self._onsuccess)
+    async def _stat_file(self, path: str) -> StatInfo:
-
-    def _get_boto_key(self, path: str) -> Deferred[dict[str, Any]]:
         key_name = f"{self.prefix}{path}"
-        return deferred_from_coro(
+        if self._use_async():
-            run_in_thread(
+            client = await self._get_aio_client()
+            boto_key = await client.head_object(Bucket=self.bucket, Key=key_name)
+        else:
+            boto_key = await run_in_thread(
                 self.s3_client.head_object,  # type: ignore[attr-defined]
                 Bucket=self.bucket,
                 Key=key_name,
             )
-        )
+        return self._onsuccess(boto_key)
 
     def persist_file(
         self,
@@ -230,21 +273,34 @@ class S3FilesStore:
         headers: dict[str, str] | None = None,
     ) -> Deferred[Any]:
         """Upload file to S3 storage"""
+        return deferred_from_coro(self._persist_file(path, buf, meta, headers))
+
+    async def _persist_file(
+        self,
+        path: str,
+        buf: BytesIO,
+        meta: dict[str, Any] | None,
+        headers: dict[str, str] | None,
+    ) -> Any:
         key_name = f"{self.prefix}{path}"
         buf.seek(0)
         extra = self._headers_to_botocore_kwargs(self.HEADERS)
         if headers:
             extra.update(self._headers_to_botocore_kwargs(headers))
-        return deferred_from_coro(
+        kwargs: dict[str, Any] = {
-            run_in_thread(
+            "Bucket": self.bucket,
-                self.s3_client.put_object,  # type: ignore[attr-defined]
+            "Key": key_name,
-                Bucket=self.bucket,
+            "Body": buf,
-                Key=key_name,
+            "Metadata": {k: str(v) for k, v in meta.items()} if meta else {},
-                Body=buf,
+            "ACL": self.POLICY,
-                Metadata={k: str(v) for k, v in meta.items()} if meta else {},
+            **extra,
-                ACL=self.POLICY,
+        }
-                **extra,
+        if self._use_async():
-            )
+            client = await self._get_aio_client()
+            return await client.put_object(**kwargs)
+        return await run_in_thread(
+            self.s3_client.put_object,  # type: ignore[attr-defined]
+            **kwargs,
         )
 
     def _headers_to_botocore_kwargs(self, headers: dict[str, Any]) -> dict[str, Any]:
@@ -518,6 +574,12 @@ class FilesPipeline(MediaPipeline):
         store_uri = settings["FILES_STORE"]
         return cls(store_uri, crawler=crawler)
 
+    @_warn_spider_arg
+    async def close_spider(self, spider: Spider | None = None) -> None:
+        close = getattr(self.store, "close", None)
+        if close is not None:
+            await ensure_awaitable(close())
+
     @classmethod
     def _update_stores(cls, settings: BaseSettings) -> None:
         s3store: type[S3FilesStore] = cast(

scrapy/utils/boto.py

@@ -13,6 +13,14 @@ def is_botocore_available() -> bool:
     return find_spec("botocore") is not None
 
 
+def is_aiobotocore_available() -> bool:
+    return find_spec("aiobotocore") is not None
+
+
+def is_aioboto3_available() -> bool:
+    return find_spec("aioboto3") is not None
+
+
 def _get_max_pool_connections(settings: BaseSettings) -> int:
     """Return the maximum number of connections that AWS clients may keep in
     their connection pool.

tests/test_feedexport_batch.py

@@ -7,6 +7,7 @@ import pickle
 from collections import defaultdict
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
+from unittest import mock
 from urllib.parse import urljoin
 
 import lxml.etree
@@ -426,7 +427,12 @@ class TestBatchDeliveries(TestFeedExportBase):
 
     @pytest.mark.requires_boto3
     @inline_callbacks_test
-    def test_s3_export(self):
+    def test_s3_export(self, monkeypatch):
+        # Force the blocking boto3 code path so that the botocore Stubber below
+        # intercepts the uploads.
+        monkeypatch.setattr(
+            "scrapy.extensions.feedexport.is_aioboto3_available", lambda: False
+        )
         bucket = "mybucket"
         items = [
             MyItem({"foo": "bar1", "egg": "spam1"}),
@@ -498,3 +504,54 @@ class TestBatchDeliveries(TestFeedExportBase):
         assert (
             crawler.stats.get_value("feedexport/success_count/CustomS3FeedStorage") == 3
         )
+
+    @pytest.mark.requires_aioboto3
+    @pytest.mark.only_asyncio
+    @inline_callbacks_test
+    def test_s3_export_async(self, monkeypatch):
+        """One batch per item is uploaded through the aioboto3 code path."""
+        import aioboto3  # noqa: PLC0415
+
+        bucket = "mybucket"
+        items = [
+            MyItem({"foo": "bar1", "egg": "spam1"}),
+            MyItem({"foo": "bar2", "egg": "spam2", "baz": "quux2"}),
+            MyItem({"foo": "bar3", "baz": "quux3"}),
+        ]
+
+        upload_fileobj = mock.AsyncMock()
+
+        def make_client(self, *args, **kwargs):
+            client = mock.MagicMock()
+            client.upload_fileobj = upload_fileobj
+            client_cm = mock.MagicMock()
+            client_cm.__aenter__ = mock.AsyncMock(return_value=client)
+            client_cm.__aexit__ = mock.AsyncMock(return_value=False)
+            return client_cm
+
+        monkeypatch.setattr(aioboto3.Session, "client", make_client)
+
+        key = "export.csv"
+        uri = f"s3://{bucket}/{key}/%(batch_id)d.json"
+        settings = {
+            "AWS_ACCESS_KEY_ID": "access_key",
+            "AWS_SECRET_ACCESS_KEY": "secret_key",
+            "FEED_EXPORT_BATCH_ITEM_COUNT": 1,
+            "FEEDS": {uri: {"format": "json"}},
+        }
+
+        class TestSpider(scrapy.Spider):
+            name = "testspider"
+
+            def parse(self, response):
+                yield from items
+
+        TestSpider.start_urls = [self.mockserver.url("/")]
+        crawler = get_crawler(TestSpider, settings)
+        yield crawler.crawl()
+
+        assert upload_fileobj.await_count == len(items)
+        for call in upload_fileobj.await_args_list:
+            assert call.kwargs["Bucket"] == bucket
+        assert crawler.stats
+        assert crawler.stats.get_value("feedexport/success_count/S3FeedStorage") == 3

tests/test_feedexport_storages.py

@@ -279,7 +279,12 @@ class TestS3FeedStorage:
         assert storage.secret_key == "uri_secret"
 
     @coroutine_test
-    async def test_store(self):
+    async def test_store(self, monkeypatch):
+        """The blocking boto3 client is used when asyncio/aioboto3 support is
+        not available."""
+        monkeypatch.setattr(
+            "scrapy.extensions.feedexport.is_aioboto3_available", lambda: False
+        )
         settings = {
             "AWS_ACCESS_KEY_ID": "access_key",
             "AWS_SECRET_ACCESS_KEY": "secret_key",
@@ -458,7 +463,10 @@ class TestS3FeedStorage:
         assert config.max_pool_connections == expected
 
     @coroutine_test
-    async def test_store_without_acl(self):
+    async def test_store_without_acl(self, monkeypatch):
+        monkeypatch.setattr(
+            "scrapy.extensions.feedexport.is_aioboto3_available", lambda: False
+        )
         storage = S3FeedStorage(
             "s3://mybucket/export.csv",
             "access_key",
@@ -478,7 +486,10 @@ class TestS3FeedStorage:
         assert acl is None
 
     @coroutine_test
-    async def test_store_with_acl(self):
+    async def test_store_with_acl(self, monkeypatch):
+        monkeypatch.setattr(
+            "scrapy.extensions.feedexport.is_aioboto3_available", lambda: False
+        )
         storage = S3FeedStorage(
             "s3://mybucket/export.csv", "access_key", "secret_key", "custom-acl"
         )
@@ -491,6 +502,40 @@ class TestS3FeedStorage:
         acl = storage.s3_client.upload_fileobj.call_args[1]["ExtraArgs"]["ACL"]
         assert acl == "custom-acl"
 
+    @pytest.mark.requires_aioboto3
+    @pytest.mark.only_asyncio
+    @coroutine_test
+    async def test_store_async(self, monkeypatch):
+        """The genuinely-asynchronous aioboto3 client is used when
+        asyncio/aioboto3 support is available."""
+        import aioboto3  # noqa: PLC0415
+
+        storage = S3FeedStorage(
+            "s3://mybucket/export.csv", "access_key", "secret_key", "custom-acl"
+        )
+
+        client = mock.MagicMock()
+        client.upload_fileobj = mock.AsyncMock()
+        client_cm = mock.MagicMock()
+        client_cm.__aenter__ = mock.AsyncMock(return_value=client)
+        client_cm.__aexit__ = mock.AsyncMock(return_value=False)
+
+        file = BytesIO(b"test file")
+        with mock.patch.object(
+            aioboto3.Session, "client", return_value=client_cm
+        ) as client_call:
+            await maybe_deferred_to_future(storage.store(file))
+
+        client_call.assert_called_once_with("s3", **storage._client_kwargs)
+        client.upload_fileobj.assert_awaited_once_with(
+            Fileobj=file,
+            Bucket="mybucket",
+            Key="export.csv",
+            ExtraArgs={"ACL": "custom-acl"},
+        )
+        client_cm.__aexit__.assert_awaited_once()
+        assert file.closed
+
     def test_overwrite_default(self, caplog: pytest.LogCaptureFixture) -> None:
         S3FeedStorage(
             "s3://mybucket/export.csv", "access_key", "secret_key", "custom-acl"

tests/test_pipeline_files.py

@@ -112,6 +112,20 @@ class TestFilesPipeline:
         pipeline.open_spider()
         return pipeline
 
+    @coroutine_test
+    async def test_close_spider_closes_store(self):
+        """close_spider() closes the store when it exposes a close() method."""
+        self.pipeline.store = mock.MagicMock()
+        self.pipeline.store.close = mock.AsyncMock()
+        await self.pipeline.close_spider()
+        self.pipeline.store.close.assert_awaited_once_with()
+
+    @coroutine_test
+    async def test_close_spider_without_store_close(self):
+        """close_spider() is a no-op for stores without a close() method."""
+        assert not hasattr(self.pipeline.store, "close")
+        await self.pipeline.close_spider()
+
     def test_file_path_query_parameters(self):
         file_path = self.pipeline.file_path
 
@@ -829,7 +843,9 @@ class TestFSFilesStore:
 @pytest.mark.requires_botocore
 class TestS3FilesStore:
     @inline_callbacks_test
-    def test_persist(self):
+    def test_persist(self, monkeypatch):
+        """The blocking botocore client is used when asyncio/aiobotocore support
+        is not available."""
         bucket = "mybucket"
         key = "export.csv"
         uri = f"s3://{bucket}/{key}"
@@ -839,6 +855,7 @@ class TestS3FilesStore:
         content_type = "image/png"
 
         store = S3FilesStore(uri)
+        monkeypatch.setattr(store, "_use_async", lambda: False)
         from botocore.stub import Stubber  # noqa: PLC0415
 
         with Stubber(store.s3_client) as stub:
@@ -869,13 +886,14 @@ class TestS3FilesStore:
             assert buffer.method_calls == [mock.call.seek(0)]
 
     @inline_callbacks_test
-    def test_persist_without_headers(self):
+    def test_persist_without_headers(self, monkeypatch):
         """Without custom headers only the default ones are sent."""
         bucket = "mybucket"
         key = "export.csv"
         buffer = mock.MagicMock()
 
         store = S3FilesStore(f"s3://{bucket}/{key}")
+        monkeypatch.setattr(store, "_use_async", lambda: False)
         from botocore.stub import Stubber  # noqa: PLC0415
 
         with Stubber(store.s3_client) as stub:
@@ -922,7 +940,9 @@ class TestS3FilesStore:
             store._headers_to_botocore_kwargs({"X-Custom": "value"})
 
     @inline_callbacks_test
-    def test_stat(self):
+    def test_stat(self, monkeypatch):
+        """The blocking botocore client is used when asyncio/aiobotocore support
+        is not available."""
         bucket = "mybucket"
         key = "export.csv"
         uri = f"s3://{bucket}/{key}"
@@ -930,6 +950,7 @@ class TestS3FilesStore:
         last_modified = datetime(2019, 12, 1)
 
         store = S3FilesStore(uri)
+        monkeypatch.setattr(store, "_use_async", lambda: False)
         from botocore.stub import Stubber  # noqa: PLC0415
 
         with Stubber(store.s3_client) as stub:
@@ -981,6 +1002,113 @@ class TestS3FilesStore:
         assert config.max_pool_connections == expected
 
 
+@pytest.mark.requires_aiobotocore
+@pytest.mark.only_asyncio
+class TestS3FilesStoreAsync:
+    """Tests for the genuinely-asynchronous aiobotocore code path of
+    :class:`~scrapy.pipelines.files.S3FilesStore`."""
+
+    @coroutine_test
+    async def test_persist(self):
+        bucket = "mybucket"
+        key = "export.csv"
+        uri = f"s3://{bucket}/{key}"
+        buffer = mock.MagicMock()
+        meta = {"foo": "bar"}
+        path = ""
+        content_type = "image/png"
+
+        store = S3FilesStore(uri)
+        assert store._use_async()
+        client = await store._get_aio_client()
+        from aiobotocore.stub import Stubber  # noqa: PLC0415
+
+        with Stubber(client) as stub:
+            stub.add_response(
+                "put_object",
+                expected_params={
+                    "ACL": S3FilesStore.POLICY,
+                    "Body": buffer,
+                    "Bucket": bucket,
+                    "CacheControl": S3FilesStore.HEADERS["Cache-Control"],
+                    "ContentType": content_type,
+                    "Key": key,
+                    "Metadata": meta,
+                },
+                service_response={},
+            )
+
+            await maybe_deferred_to_future(
+                store.persist_file(
+                    path,
+                    buffer,
+                    info=DUMMY_SPIDER_INFO,
+                    meta=meta,
+                    headers={"Content-Type": content_type},
+                )
+            )
+
+            stub.assert_no_pending_responses()
+            assert buffer.method_calls == [mock.call.seek(0)]
+
+        await store.close()
+
+    @coroutine_test
+    async def test_max_pool_connections(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(S3FilesStore, "AWS_MAX_POOL_CONNECTIONS", 30)
+        store = S3FilesStore("s3://mybucket/prefix/")
+        client = await store._get_aio_client()
+        assert client.meta.config.max_pool_connections == 30
+        await store.close()
+
+    @coroutine_test
+    async def test_stat(self):
+        bucket = "mybucket"
+        key = "export.csv"
+        uri = f"s3://{bucket}/{key}"
+        checksum = "3187896a9657a28163abb31667df64c8"
+        last_modified = datetime(2019, 12, 1)
+
+        store = S3FilesStore(uri)
+        client = await store._get_aio_client()
+        from aiobotocore.stub import Stubber  # noqa: PLC0415
+
+        with Stubber(client) as stub:
+            stub.add_response(
+                "head_object",
+                expected_params={"Bucket": bucket, "Key": key},
+                service_response={
+                    "ETag": f'"{checksum}"',
+                    "LastModified": last_modified,
+                },
+            )
+
+            file_stats = await maybe_deferred_to_future(
+                store.stat_file("", info=DUMMY_SPIDER_INFO)
+            )
+            assert file_stats == {
+                "checksum": checksum,
+                "last_modified": last_modified.timestamp(),
+            }
+
+            stub.assert_no_pending_responses()
+
+        await store.close()
+
+    @coroutine_test
+    async def test_client_reused_and_closed(self):
+        store = S3FilesStore("s3://mybucket/export.csv")
+        assert store._aio_client is None
+        client = await store._get_aio_client()
+        # The client is created lazily and reused across calls.
+        assert store._aio_client is client
+        assert await store._get_aio_client() is client
+        await store.close()
+        assert store._aio_client is None
+        # close() is a no-op if there is no open client.
+        await store.close()
+
+
 class TestGCSFilesStore:
     @staticmethod
     def build_gcs_files_store(

tox.ini

@@ -172,6 +172,7 @@ deps =
     {[testenv]deps}
     Pillow
     Twisted[http2]
+    aioboto3
     boto3
     bpython  # optional for shell wrapper tests
     google-cloud-storage
@@ -350,9 +351,10 @@ commands =
 [testenv:botocore]
 deps =
     {[testenv]deps}
+    aiobotocore
     botocore>=1.13.45
 commands =
-    pytest {posargs:--cov-config=pyproject.toml --cov=scrapy --cov-report=xml --cov-report= tests --junitxml=botocore.junit.xml -o junit_family=legacy} -m requires_botocore
+    pytest {posargs:--cov-config=pyproject.toml --cov=scrapy --cov-report=xml --cov-report= tests --junitxml=botocore.junit.xml -o junit_family=legacy} -m "requires_botocore or requires_aiobotocore"
 
 [testenv:min-botocore]
 basepython = {[min]basepython}