diff --git a/docs/topics/coroutines.rst b/docs/topics/coroutines.rst index 9dcd9d69c..cf39f1487 100644 --- a/docs/topics/coroutines.rst +++ b/docs/topics/coroutines.rst @@ -52,6 +52,26 @@ hence use coroutine syntax (e.g. ``await``, ``async for``, ``async with``): .. versionadded:: 2.14 +- The following methods of :ref:`item exporters `, when the + exporter is used by the :ref:`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 ` (see + :setting:`FEED_STORAGES`): + + - :meth:`~scrapy.extensions.feedexport.FeedStorageProtocol.open` + + - :meth:`~scrapy.extensions.feedexport.FeedStorageProtocol.store` + + .. versionadded:: VERSION + .. _coroutine-deferred-apis: @@ -104,7 +124,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`: diff --git a/docs/topics/exporters.rst b/docs/topics/exporters.rst index ecd154122..3700e9308 100644 --- a/docs/topics/exporters.rst +++ b/docs/topics/exporters.rst @@ -164,9 +164,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 .. method:: serialize_field(field, name, value) @@ -189,19 +187,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 diff --git a/docs/topics/feed-exports.rst b/docs/topics/feed-exports.rst index 66768c97b..ab4c0780b 100644 --- a/docs/topics/feed-exports.rst +++ b/docs/topics/feed-exports.rst @@ -298,6 +298,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 +`) 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 diff --git a/scrapy/exporters.py b/scrapy/exporters.py index 5a11df833..f3374c5b8 100644 --- a/scrapy/exporters.py +++ b/scrapy/exporters.py @@ -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 `. + """ 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 `. + """ + + 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 `. + """ def _get_serialized_fields( self, item: Any, default_value: Any = None, include_empty: bool | None = None diff --git a/scrapy/extensions/feedexport.py b/scrapy/extensions/feedexport.py index 678a29e2e..c8cf1528a 100644 --- a/scrapy/extensions/feedexport.py +++ b/scrapy/extensions/feedexport.py @@ -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 @@ -29,7 +29,11 @@ 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.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 @@ -126,21 +130,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 ` 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(): @@ -153,6 +175,10 @@ 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 @@ -419,9 +445,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]", @@ -443,7 +471,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( @@ -453,10 +481,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: assert self.exporter - self.exporter.finish_exporting() + await ensure_awaitable(self.exporter.finish_exporting()) self._exporting = False @@ -476,6 +504,12 @@ class FeedExporter: self.slots: list[FeedSlot] = [] self.filters: dict[str, ItemFilter] = {} self._pending_close_coros: list[Coroutine[Any, Any, 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 @@ -568,11 +602,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 @@ -631,7 +665,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): @@ -640,9 +681,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 ( diff --git a/tests/test_feedexport.py b/tests/test_feedexport.py index 92c414bef..0b31c0af6 100644 --- a/tests/test_feedexport.py +++ b/tests/test_feedexport.py @@ -99,13 +99,13 @@ class InstrumentedFeedSlot(FeedSlot): """Instrumented FeedSlot subclass for keeping track of calls to start_exporting and finish_exporting.""" - 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): @@ -1261,7 +1261,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 diff --git a/tests/test_feedexport_async.py b/tests/test_feedexport_async.py new file mode 100644 index 000000000..ea6c70b81 --- /dev/null +++ b/tests/test_feedexport_async.py @@ -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