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

Merge 8bd27fa7b4 into ad43bf0c56

This commit is contained in:
Adrian 2026-08-15 11:31:49 -05:00 committed by GitHub
parent ad43bf0c56 8bd27fa7b4
commit f8a8a50e59
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 418 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
docs/topics/coroutines.rst
View File

@ -54,6 +54,26 @@ hence use coroutine syntax (e.g. ``await``, ``async for``, ``async with``):
.. versionadded:: 2.14
- The following methods of :ref:`item exporters <topics-exporters>`, when the
exporter is used by the :ref:`feed exports <topics-feed-exports>`:
- :meth:`~scrapy.exporters.BaseItemExporter.start_exporting`
- :meth:`~scrapy.exporters.BaseItemExporter.export_item`
- :meth:`~scrapy.exporters.BaseItemExporter.finish_exporting`
.. versionadded:: VERSION
- Methods of :ref:`custom feed storages <topics-feed-storage-custom>` (see
:setting:`FEED_STORAGES`):
- :meth:`~scrapy.extensions.feedexport.FeedStorageProtocol.open`
- :meth:`~scrapy.extensions.feedexport.FeedStorageProtocol.store`
.. versionadded:: VERSION
.. _coroutine-deferred-apis:
@ -106,7 +126,7 @@ return coroutines are listed in :ref:`coroutine-support`):
- Custom feed storages (see :setting:`FEED_STORAGES`):
- ``store()``
- :meth:`~scrapy.extensions.feedexport.FeedStorageProtocol.store`
- Subclasses of :class:`scrapy.pipelines.media.MediaPipeline`:

18
docs/topics/exporters.rst
View File

@ -228,9 +228,7 @@ BaseItemExporter
populate their respective instance attributes: :attr:`fields_to_export`,
:attr:`export_empty_fields`, :attr:`encoding`, :attr:`indent`.
.. method:: export_item(item)
Exports the given item. This method must be implemented in subclasses.
.. automethod:: BaseItemExporter.export_item
.. automethod:: BaseItemExporter.get_serialized_fields
@ -255,19 +253,9 @@ BaseItemExporter
:param value: the value being serialized
.. method:: start_exporting()
.. automethod:: BaseItemExporter.start_exporting
Signal the beginning of the exporting process. Some exporters may use
this to generate some required header (for example, the
:class:`XmlItemExporter`). You must call this method before exporting any
items.
.. method:: finish_exporting()
Signal the end of the exporting process. Some exporters may use this to
generate some required footer (for example, the
:class:`XmlItemExporter`). You must always call this method after you
have no more items to export.
.. automethod:: BaseItemExporter.finish_exporting
.. attribute:: fields_to_export

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

@ -326,6 +326,28 @@ soon as a file reaches the maximum item count, that file is delivered to the
feed URI, allowing item delivery to start way before the end of the crawl.
.. _topics-feed-storage-custom:
Custom storage backends
-----------------------
To write your own storage backend, define a class that follows
:class:`~scrapy.extensions.feedexport.FeedStorageProtocol` and assign it to a
URI scheme through the :setting:`FEED_STORAGES` setting.
.. autoclass:: scrapy.extensions.feedexport.FeedStorageProtocol(uri, *, feed_options=None)
:members:
If your storage backend blocks, subclass
:class:`~scrapy.extensions.feedexport.BlockingFeedStorage` instead: it writes
items into a temporary local file (see :ref:`delayed file delivery
<delayed-file-delivery>`) and calls your ``_store_in_thread()`` method in a
separate thread once the crawl is done, keeping the reactor free.
.. autoclass:: scrapy.extensions.feedexport.BlockingFeedStorage
:members: _store_in_thread
.. _item-filter:
Item filtering
@ -654,10 +676,11 @@ Default:
"ftps": "scrapy.extensions.feedexport.FTPFeedStorage",
}
A dict containing the built-in feed storage backends supported by Scrapy. You
can disable any of these backends by assigning ``None`` to their URI scheme in
:setting:`FEED_STORAGES`. E.g., to disable the built-in FTP storage backend
(without replacement), place this in your ``settings.py``:
A dict containing the built-in feed storage backends supported by Scrapy, see
:ref:`feed-storage-classes`. You can disable any of these backends by assigning
``None`` to their URI scheme in :setting:`FEED_STORAGES`. E.g., to disable the
built-in FTP storage backend (without replacement), place this in your
``settings.py``:
.. code-block:: python
@ -818,6 +841,25 @@ source spider in the feed URI:
scrapy crawl <spider_name> -o "%(spider_name)s.jsonl"
.. _feed-storage-classes:
Storage backend classes
=======================
These are the classes that :setting:`FEED_STORAGES_BASE` assigns to the
built-in URI schemes.
.. autoclass:: FileFeedStorage
.. autoclass:: FTPFeedStorage
.. autoclass:: GCSFeedStorage
.. autoclass:: S3FeedStorage
.. autoclass:: StdoutFeedStorage
.. _URIs: https://en.wikipedia.org/wiki/Uniform_Resource_Identifier
.. _Amazon S3: https://aws.amazon.com/s3/
.. _Canned ACL: https://docs.aws.amazon.com/AmazonS3/latest/userguide/acl-overview.html#canned-acl

37
scrapy/exporters.py
View File

@ -23,6 +23,7 @@ from scrapy.utils.python import is_listlike, to_bytes, to_unicode
from scrapy.utils.serialize import ScrapyJSONEncoder
if TYPE_CHECKING:
from collections.abc import Coroutine
from json import JSONEncoder
logger = logging.getLogger(__name__)
@ -59,7 +60,15 @@ class BaseItemExporter(ABC):
raise TypeError(f"Unexpected options: {', '.join(options.keys())}")
@abstractmethod
def export_item(self, item: Any) -> None:
def export_item(self, item: Any) -> Coroutine[Any, Any, None] | None:
"""Exports the given item. This method must be implemented in
subclasses.
.. versionchanged:: VERSION
This method may now be a coroutine function (``async def``). Scrapy
awaits its result when the exporter is used by the :ref:`feed
exports <topics-feed-exports>`.
"""
raise NotImplementedError
def serialize_field(
@ -68,11 +77,29 @@ class BaseItemExporter(ABC):
serializer: Callable[[Any], Any] = field.get("serializer", lambda x: x)
return serializer(value)
def start_exporting(self) -> None: # noqa: B027
pass
def start_exporting(self) -> Coroutine[Any, Any, None] | None: # noqa: B027
"""Signal the beginning of the exporting process. Some exporters may
use this to generate some required header (for example, the
:class:`XmlItemExporter`). You must call this method before exporting
any items.
def finish_exporting(self) -> None: # noqa: B027
pass
.. versionchanged:: VERSION
This method may now be a coroutine function (``async def``). Scrapy
awaits its result when the exporter is used by the :ref:`feed
exports <topics-feed-exports>`.
"""
def finish_exporting(self) -> Coroutine[Any, Any, None] | None: # noqa: B027
"""Signal the end of the exporting process. Some exporters may use this
to generate some required footer (for example, the
:class:`XmlItemExporter`). You must always call this method after you
have no more items to export.
.. versionchanged:: VERSION
This method may now be a coroutine function (``async def``). Scrapy
awaits its result when the exporter is used by the :ref:`feed
exports <topics-feed-exports>`.
"""
@staticmethod
def _get_populated_field_names(adapter: ItemAdapter) -> Iterable[str]:

102
scrapy/extensions/feedexport.py
View File

@ -20,7 +20,7 @@ from tempfile import NamedTemporaryFile
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.defer import Deferred, DeferredList, DeferredLock
from w3lib.url import file_uri_to_path
from zope.interface import Interface
@ -30,12 +30,18 @@ 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.conf import feed_complete_default_values_from_settings
from scrapy.utils.defer import deferred_from_coro, ensure_awaitable
from scrapy.utils.defer import (
deferred_from_coro,
ensure_awaitable,
maybe_deferred_to_future,
)
from scrapy.utils.ftp import ftp_store_file
from scrapy.utils.misc import build_from_crawler, load_object
from scrapy.utils.python import without_none_values
if TYPE_CHECKING:
from collections.abc import Coroutine
from _typeshed import OpenBinaryMode
# typing.Self requires Python 3.11
@ -127,21 +133,39 @@ class _IFeedStorage(Interface): # type: ignore[misc] # pragma: no cover
class FeedStorageProtocol(Protocol):
"""Protocol that all Feed Storages must follow."""
"""Protocol that all Feed Storages must follow.
def __init__(self, uri: str, *, feed_options: dict[str, Any] | None = None):
"""Initialize the storage with the parameters given in the URI and the
feed-specific options (see :setting:`FEEDS`)"""
:param uri: Feed URI.
def open(self, spider: Spider) -> IO[bytes]:
"""Open the storage for the given spider. It must return a file-like
object that will be used for the exporters"""
:param feed_options: :ref:`Feed options <feed-options>` of the feed.
def store(self, file: IO[bytes]) -> Deferred[None] | None:
"""Store the given file stream"""
A feed storage may define a ``from_crawler(cls, crawler, uri, *,
feed_options=None)`` class method instead of ``__init__``, to also get the
:class:`~scrapy.crawler.Crawler` object.
"""
def open(self, spider: Spider) -> IO[bytes] | Coroutine[Any, Any, IO[bytes]]:
"""Open the storage for *spider* and return the file-like object that
item exporters write into.
.. versionchanged:: VERSION
This method may now be a coroutine function (``async def``).
"""
def store(
self, file: IO[bytes]
) -> Coroutine[Any, Any, None] | Deferred[None] | None:
"""Store *file*, the file-like object returned by :meth:`open`, and
close it.
.. versionchanged:: VERSION
This method may now be a coroutine function (``async def``).
"""
class BlockingFeedStorage(ABC):
"""Base class for feed storages that store feeds using blocking code."""
def open(self, spider: Spider) -> IO[bytes]:
path = spider.crawler.settings["FEED_TEMPDIR"]
if path and not Path(path).is_dir():
@ -154,10 +178,16 @@ class BlockingFeedStorage(ABC):
@abstractmethod
def _store_in_thread(self, file: IO[bytes]) -> None:
"""Store *file* and close it.
This method runs in a separate thread, so it may use blocking code.
"""
raise NotImplementedError
class StdoutFeedStorage:
""":ref:`Standard output <topics-feed-storage-stdout>` storage backend."""
def __init__(
self,
uri: str,
@ -184,6 +214,12 @@ class StdoutFeedStorage:
class FileFeedStorage:
""":ref:`Local filesystem <topics-feed-storage-fs>` storage backend.
*uri* may be a ``file://`` URI or a plain path. Missing parent directories
are created when the feed is opened.
"""
def __init__(self, uri: str, *, feed_options: dict[str, Any] | None = None):
self.path: str = file_uri_to_path(uri) if uri.startswith("file:") else uri
feed_options = feed_options or {}
@ -203,6 +239,8 @@ class FileFeedStorage:
class S3FeedStorage(BlockingFeedStorage):
""":ref:`Amazon S3 <topics-feed-storage-s3>` storage backend."""
def __init__(
self,
uri: str,
@ -297,6 +335,8 @@ class S3FeedStorage(BlockingFeedStorage):
class GCSFeedStorage(BlockingFeedStorage):
""":ref:`GCS <topics-feed-storage-gcs>` storage backend."""
def __init__(
self,
uri: str,
@ -348,6 +388,9 @@ class GCSFeedStorage(BlockingFeedStorage):
class FTPFeedStorage(BlockingFeedStorage):
""":ref:`FTP <feed-storage-ftp>` storage backend, which also handles
:ref:`FTPS <feed-storage-ftps>` when *uri* uses the ``ftps`` scheme."""
def __init__(
self,
uri: str,
@ -432,9 +475,11 @@ class FeedSlot:
self._exporting: bool = False
self._fileloaded: bool = False
def start_exporting(self) -> None:
async def start_exporting(self) -> None:
if not self._fileloaded:
self.file = self.storage.open(self.spider)
self.file = cast(
"IO[bytes]", await ensure_awaitable(self.storage.open(self.spider))
)
if "postprocessing" in self.feed_options:
self.file = cast(
"IO[bytes]",
@ -456,7 +501,7 @@ class FeedSlot:
if not self._exporting:
assert self.exporter
self.exporter.start_exporting()
await ensure_awaitable(self.exporter.start_exporting())
self._exporting = True
def _get_exporter(
@ -466,10 +511,10 @@ class FeedSlot:
self.exporters[format_], self.crawler, file, *args, **kwargs
)
def finish_exporting(self) -> None:
async def finish_exporting(self) -> None:
if self._exporting: # pragma: no branch
assert self.exporter
self.exporter.finish_exporting()
await ensure_awaitable(self.exporter.finish_exporting())
self._exporting = False
@ -489,6 +534,12 @@ class FeedExporter:
self.slots: list[FeedSlot] = []
self.filters: dict[str, ItemFilter] = {}
self._pending_close_tasks: list[asyncio.Task[None] | Deferred[None]] = []
# Item export may await user-defined code (e.g. an async def
# export_item() method), while item_scraped signals are sent
# concurrently (see the CONCURRENT_ITEMS setting). This lock prevents
# overlapping item export calls, which item exporters, being stateful,
# cannot handle.
self._export_lock = DeferredLock()
if not self.settings["FEEDS"] and not self.settings["FEED_URI"]:
raise NotConfigured
@ -602,11 +653,11 @@ class FeedExporter:
if slot.itemcount:
# Normal case
slot.finish_exporting()
await slot.finish_exporting()
elif slot.store_empty and slot.batch_id == 1:
# Need to store the empty file
slot.start_exporting()
slot.finish_exporting()
await slot.start_exporting()
await slot.finish_exporting()
else:
# In this case, the file is not stored, so no processing is required.
return
@ -664,7 +715,14 @@ class FeedExporter:
crawler=self.crawler,
)
def item_scraped(self, item: Any, spider: Spider) -> None:
async def item_scraped(self, item: Any, spider: Spider) -> None:
await maybe_deferred_to_future(self._export_lock.acquire())
try:
await self._export_item(item, spider)
finally:
self._export_lock.release()
async def _export_item(self, item: Any, spider: Spider) -> None:
slots = []
for slot in self.slots:
if not slot.filter.accepts(item):
@ -673,9 +731,9 @@ class FeedExporter:
) # if slot doesn't accept item, continue with next slot
continue
slot.start_exporting()
await slot.start_exporting()
assert slot.exporter
slot.exporter.export_item(item)
await ensure_awaitable(slot.exporter.export_item(item))
slot.itemcount += 1
# create new slot for each slot with itemcount == FEED_EXPORT_BATCH_ITEM_COUNT and close the old one
if (

2
tests/test_exporters.py
View File

@ -143,7 +143,7 @@ class TestBaseItemExporter(ABC):
class TestPythonItemExporter(TestBaseItemExporter):
def _get_exporter(self, **kwargs: Any) -> BaseItemExporter:
def _get_exporter(self, **kwargs: Any) -> PythonItemExporter:
return PythonItemExporter(**kwargs)
def test_invalid_option(self):

10
tests/test_feedexport.py
View File

@ -103,13 +103,13 @@ class InstrumentedFeedSlot(FeedSlot):
update_listener: Callable[[str], None]
def start_exporting(self):
async def start_exporting(self):
self.update_listener("start")
super().start_exporting()
await super().start_exporting()
def finish_exporting(self):
async def finish_exporting(self):
self.update_listener("finish")
super().finish_exporting()
await super().finish_exporting()
@classmethod
def subscribe__listener(cls, listener: IsExportingListener) -> None:
@ -1274,7 +1274,7 @@ class TestFeedExporterSignals:
)
feed_exporter.open_spider(spider)
for item in self.items:
feed_exporter.item_scraped(item, spider)
await feed_exporter.item_scraped(item, spider)
await feed_exporter.close_spider(spider)
@coroutine_test

230
tests/test_feedexport_async.py Normal file
View File

@ -0,0 +1,230 @@
"""Tests for coroutine support in feed storages and item exporters."""
from __future__ import annotations
import json
from contextlib import asynccontextmanager
from pathlib import Path
from typing import IO, TYPE_CHECKING, Any
from twisted.internet.defer import Deferred
from w3lib.url import file_uri_to_path
from scrapy.exporters import JsonLinesItemExporter
from scrapy.extensions.feedexport import FileFeedStorage
from scrapy.utils.asyncio import call_later
from scrapy.utils.defer import maybe_deferred_to_future
from scrapy.utils.test import get_crawler
from tests.utils.bases.feedexport import TestFeedExportBase
from tests.utils.decorators import coroutine_test
from tests.utils.feedexport import path_to_url, printf_escape
if TYPE_CHECKING:
from collections.abc import AsyncIterator, Iterable
from scrapy import Spider
async def suspend() -> None:
"""Let other coroutines run before resuming the calling one.
Unlike :func:`asyncio.sleep`, this works with any value of the
:setting:`TWISTED_REACTOR` setting.
"""
d: Deferred[None] = Deferred()
call_later(0, d.callback, None)
await maybe_deferred_to_future(d)
class CallTracker:
"""Base class for feed components that record their calls, and whether any
of those calls overlap."""
calls: dict[str, int] = {}
active = 0
max_active = 0
@classmethod
def reset(cls) -> None:
CallTracker.calls = {}
CallTracker.active = 0
CallTracker.max_active = 0
@asynccontextmanager
async def tracked(self, name: str) -> AsyncIterator[None]:
key = f"{type(self).__name__}.{name}"
CallTracker.calls[key] = CallTracker.calls.get(key, 0) + 1
CallTracker.active += 1
CallTracker.max_active = max(CallTracker.max_active, CallTracker.active)
try:
await suspend()
yield
finally:
CallTracker.active -= 1
class AsyncFeedStorage(CallTracker):
"""Feed storage that follows
:class:`~scrapy.extensions.feedexport.FeedStorageProtocol` with coroutine
``open()`` and ``store()`` methods."""
def __init__(self, uri: str, *, feed_options: dict[str, Any] | None = None):
self.path: Path = Path(file_uri_to_path(uri))
async def open(self, spider: Spider) -> IO[bytes]:
async with self.tracked("open"):
return self.path.open("wb")
async def store(self, file: IO[bytes]) -> None:
async with self.tracked("store"):
file.close()
class DeferredFeedStorage(FileFeedStorage):
"""Feed storage whose ``store()`` method returns a
:class:`~twisted.internet.defer.Deferred` object."""
def store(self, file: IO[bytes]) -> Deferred[None]:
d: Deferred[None] = Deferred()
call_later(0, d.callback, None)
return d.addCallback(lambda _: file.close())
class AsyncJsonLinesItemExporter(CallTracker, JsonLinesItemExporter):
"""Item exporter with coroutine ``start_exporting()``, ``export_item()``
and ``finish_exporting()`` methods."""
async def start_exporting(self) -> None:
async with self.tracked("start_exporting"):
super().start_exporting()
async def export_item(self, item: Any) -> None: # type: ignore[override]
async with self.tracked("export_item"):
super().export_item(item)
async def finish_exporting(self) -> None:
async with self.tracked("finish_exporting"):
super().finish_exporting()
class TestAsyncFeedExport(TestFeedExportBase):
items: list[dict[str, Any]] = [{"foo": f"bar{index}"} for index in range(10)]
async def run_and_export(
self, spider_cls: type[Spider], settings: dict[str, Any]
) -> dict[str, bytes | None]:
"""Run spider with specified settings; return exported data by path."""
feeds = settings["FEEDS"]
settings["FEEDS"] = {
printf_escape(path_to_url(file_path)): feed_options
for file_path, feed_options in feeds.items()
}
try:
spider_cls.start_urls = [self.mockserver.url("/")]
crawler = get_crawler(spider_cls, settings)
await crawler.crawl_async()
return {
str(file_path): (
Path(file_path).read_bytes() if Path(file_path).exists() else None
)
for file_path in feeds
}
finally:
for file_path in feeds:
Path(file_path).unlink(missing_ok=True)
async def _export(
self, items: list[dict[str, Any]], settings: dict[str, Any]
) -> bytes | None:
path = self._random_temp_filename()
settings["FEEDS"] = {path: {"format": "jl"}}
data: dict[str, bytes | None] = await self.exported_data(items, settings)
return data[str(path)]
@staticmethod
def _sorted(items: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
return sorted(items, key=lambda item: item["foo"])
def _parse_jsonlines(self, data: bytes | None) -> list[dict[str, Any]]:
assert data is not None
# Items are exported in scraping completion order, which is arbitrary.
return self._sorted(json.loads(line) for line in data.splitlines())
@coroutine_test
async def test_storage(self) -> None:
CallTracker.reset()
data = await self._export(
self.items, {"FEED_STORAGES": {"file": AsyncFeedStorage}}
)
assert CallTracker.calls == {
"AsyncFeedStorage.open": 1,
"AsyncFeedStorage.store": 1,
}
assert self._parse_jsonlines(data) == self._sorted(self.items)
@coroutine_test
async def test_storage_no_items(self) -> None:
CallTracker.reset()
data = await self._export(
[],
{
"FEED_STORAGES": {"file": AsyncFeedStorage},
"FEED_STORE_EMPTY": True,
},
)
assert CallTracker.calls == {
"AsyncFeedStorage.open": 1,
"AsyncFeedStorage.store": 1,
}
assert data == b""
@coroutine_test
async def test_storage_deferred_store(self) -> None:
data = await self._export(
self.items, {"FEED_STORAGES": {"file": DeferredFeedStorage}}
)
assert self._parse_jsonlines(data) == self._sorted(self.items)
@coroutine_test
async def test_exporter(self) -> None:
CallTracker.reset()
data = await self._export(
self.items, {"FEED_EXPORTERS": {"jl": AsyncJsonLinesItemExporter}}
)
assert CallTracker.calls == {
"AsyncJsonLinesItemExporter.start_exporting": 1,
"AsyncJsonLinesItemExporter.export_item": len(self.items),
"AsyncJsonLinesItemExporter.finish_exporting": 1,
}
assert self._parse_jsonlines(data) == self._sorted(self.items)
@coroutine_test
async def test_exporter_no_items(self) -> None:
CallTracker.reset()
data = await self._export(
[],
{
"FEED_EXPORTERS": {"jl": AsyncJsonLinesItemExporter},
"FEED_STORE_EMPTY": True,
},
)
assert CallTracker.calls == {
"AsyncJsonLinesItemExporter.start_exporting": 1,
"AsyncJsonLinesItemExporter.finish_exporting": 1,
}
assert data == b""
@coroutine_test
async def test_calls_are_serialized(self) -> None:
"""Item export calls never overlap, even though items are scraped
concurrently, so that components that keep state do not need to support
concurrent calls."""
CallTracker.reset()
await self._export(
self.items,
{
"FEED_STORAGES": {"file": AsyncFeedStorage},
"FEED_EXPORTERS": {"jl": AsyncJsonLinesItemExporter},
},
)
assert CallTracker.max_active == 1