From 580068a779d69313fbbbd83ae0429e2d1c71ff18 Mon Sep 17 00:00:00 2001 From: Adrian Chaves Date: Fri, 24 Jul 2026 14:25:24 +0200 Subject: [PATCH] Support aioboto3 --- conftest.py | 2 + docs/topics/feed-exports.rst | 8 ++ docs/topics/media-pipeline.rst | 8 ++ pyproject.toml | 6 +- scrapy/extensions/feedexport.py | 42 ++++++++-- scrapy/pipelines/files.py | 121 +++++++++++++++++++++-------- scrapy/utils/boto.py | 8 ++ tests/test_feedexport_batch.py | 59 +++++++++++++- tests/test_feedexport_storages.py | 51 ++++++++++++- tests/test_pipeline_files.py | 123 +++++++++++++++++++++++++++++- tox.ini | 4 +- 11 files changed, 385 insertions(+), 47 deletions(-) diff --git a/conftest.py b/conftest.py index b8a9dc19e..b1e5ad7bb 100644 --- a/conftest.py +++ b/conftest.py @@ -125,6 +125,8 @@ def pytest_runtest_setup(item): "uvloop", "botocore", "boto3", + "aiobotocore", + "aioboto3", ] for module in optional_deps: diff --git a/docs/topics/feed-exports.rst b/docs/topics/feed-exports.rst index 66768c97b..bd265423b 100644 --- a/docs/topics/feed-exports.rst +++ b/docs/topics/feed-exports.rst @@ -233,6 +233,12 @@ storage backend is: ``True``. This storage backend uses :ref:`delayed file delivery `. +.. versionchanged:: VERSION + When :ref:`asyncio support is available ` and aioboto3_ is + installed (it is part of the :ref:`s3 ` extra), feeds are uploaded + using genuinely-asynchronous I/O. Otherwise, the blocking boto3_ client is + run in a separate thread. + .. _topics-feed-storage-gcs: @@ -790,6 +796,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/ diff --git a/docs/topics/media-pipeline.rst b/docs/topics/media-pipeline.rst index 4ceb4732a..efbf04d39 100644 --- a/docs/topics/media-pipeline.rst +++ b/docs/topics/media-pipeline.rst @@ -268,6 +268,14 @@ For self-hosting you also might feel the need not to use SSL and not to verify S AWS_USE_SSL = False # or True (None by default) AWS_VERIFY = False # or True (None by default) +.. versionchanged:: VERSION + When :ref:`asyncio support is available ` and aiobotocore_ is + installed (it is part of the :ref:`s3 ` 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/ diff --git a/pyproject.toml b/pyproject.toml index 4220b12ed..7bd64daac 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -72,7 +72,7 @@ images = ["Pillow>=8.3.2"] ipython = ["ipython>=7.1.0"] ptpython = ["ptpython>=2.0.1"] 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'", @@ -217,6 +217,8 @@ ignore_errors = true # usually no type hints [[tool.mypy.overrides]] module = [ + "aioboto3", + "aiobotocore.*", "bpython", "brotli", "brotlicffi", @@ -363,6 +365,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", ] diff --git a/scrapy/extensions/feedexport.py b/scrapy/extensions/feedexport.py index 678a29e2e..c198677d0 100644 --- a/scrapy/extensions/feedexport.py +++ b/scrapy/extensions/feedexport.py @@ -28,6 +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 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 @@ -229,15 +230,18 @@ class S3FeedStorage(BlockingFeedStorage): self.endpoint_url: str | None = endpoint_url self.region_name: str | None = region_name + self._client_kwargs: dict[str, Any] = { + "aws_access_key_id": self.access_key, + "aws_secret_access_key": self.secret_key, + "aws_session_token": self.session_token, + "endpoint_url": self.endpoint_url, + "region_name": self.region_name, + } + + # 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", - aws_access_key_id=self.access_key, - aws_secret_access_key=self.secret_key, - aws_session_token=self.session_token, - endpoint_url=self.endpoint_url, - region_name=self.region_name, - ) + self.s3_client = boto3_session.client("s3", **self._client_kwargs) if feed_options and feed_options.get("overwrite", True) is False: logger.warning( @@ -265,6 +269,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: diff --git a/scrapy/pipelines/files.py b/scrapy/pipelines/files.py index 70dc36e35..e793d5c58 100644 --- a/scrapy/pipelines/files.py +++ b/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 @@ -28,9 +29,10 @@ from scrapy.exceptions import IgnoreRequest, NotConfigured, ScrapyDeprecationWar 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.asyncio import is_asyncio_available, run_in_thread +from scrapy.utils.boto import 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 @@ -47,6 +49,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 @@ -169,23 +172,58 @@ class S3FilesStore: def __init__(self, uri: str): if not is_botocore_available(): raise NotConfigured("missing botocore library") - import botocore.session # noqa: PLC0415 - - 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, - ) if not uri.startswith("s3://"): raise ValueError(f"Incorrect URI scheme in {uri}, expected 's3'") self.bucket, self.prefix = uri[5:].split("/", 1) + 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, + } + + # 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: + if self._aio_client is None: + async with self._aio_client_lock: + # Another concurrent call may have created it in the meantime. + 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 + ) + 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('"') @@ -196,18 +234,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) - - def _get_boto_key(self, path: str) -> Deferred[dict[str, Any]]: + async def _stat_file(self, path: str) -> StatInfo: key_name = f"{self.prefix}{path}" - return deferred_from_coro( - run_in_thread( + if self._use_async(): + 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, @@ -218,21 +258,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( - 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.items()} if meta else {}, - ACL=self.POLICY, - **extra, - ) + kwargs: dict[str, Any] = { + "Bucket": self.bucket, + "Key": key_name, + "Body": buf, + "Metadata": {k: str(v) for k, v in meta.items()} if meta else {}, + "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]: @@ -506,6 +559,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( diff --git a/scrapy/utils/boto.py b/scrapy/utils/boto.py index 2a77ee2ac..e1278e3c3 100644 --- a/scrapy/utils/boto.py +++ b/scrapy/utils/boto.py @@ -5,3 +5,11 @@ from importlib.util import find_spec 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 diff --git a/tests/test_feedexport_batch.py b/tests/test_feedexport_batch.py index 3b50cd492..70f921308 100644 --- a/tests/test_feedexport_batch.py +++ b/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 @@ -384,7 +385,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"}), @@ -456,3 +462,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 diff --git a/tests/test_feedexport_storages.py b/tests/test_feedexport_storages.py index bdd0d7614..511e8e2bd 100644 --- a/tests/test_feedexport_storages.py +++ b/tests/test_feedexport_storages.py @@ -242,7 +242,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", @@ -382,7 +387,10 @@ class TestS3FeedStorage: assert storage.s3_client._client_config.region_name == region_name @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", @@ -402,7 +410,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" ) @@ -415,6 +426,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): with LogCapture() as log: S3FeedStorage( diff --git a/tests/test_pipeline_files.py b/tests/test_pipeline_files.py index 0d9aac830..a5e8f2607 100644 --- a/tests/test_pipeline_files.py +++ b/tests/test_pipeline_files.py @@ -101,6 +101,20 @@ class TestFilesPipeline: def teardown_method(self): rmtree(self.tempdir) + @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 @@ -588,7 +602,9 @@ class TestFilesPipelineCustomSettings: @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}" @@ -598,6 +614,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: @@ -628,7 +645,9 @@ class TestS3FilesStore: assert buffer.method_calls == [mock.call.seek(0)] @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}" @@ -636,6 +655,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: @@ -660,6 +680,105 @@ class TestS3FilesStore: stub.assert_no_pending_responses() +@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_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( diff --git a/tox.ini b/tox.ini index d35fe91db..f355f400f 100644 --- a/tox.ini +++ b/tox.ini @@ -166,6 +166,7 @@ deps = {[testenv]deps} Pillow Twisted[http2] + aioboto3 boto3 bpython # optional for shell wrapper tests brotli >= 1.2.0; implementation_name != "pypy" # optional for HTTP compress downloader middleware tests @@ -303,9 +304,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}