RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Support aioboto3

This commit is contained in:
Adrian Chaves 2026-07-24 14:25:24 +02:00
parent abbc024bbe
commit 580068a779
11 changed files with 385 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
conftest.py
View File

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

8
docs/topics/feed-exports.rst
View File

@ -233,6 +233,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:
@ -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/

8
docs/topics/media-pipeline.rst
View File

@ -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 <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/

6
pyproject.toml
View File

@ -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",
]

42
scrapy/extensions/feedexport.py
View File

@ -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:

121
scrapy/pipelines/files.py
View File

@ -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(

8
scrapy/utils/boto.py
View File

@ -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

59
tests/test_feedexport_batch.py
View File

@ -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

51
tests/test_feedexport_storages.py
View File

@ -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(

123
tests/test_pipeline_files.py
View File

@ -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(

4
tox.ini
View File

@ -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}