diff --git a/docs/topics/coroutines.rst b/docs/topics/coroutines.rst index b7ddb0a57..9a0067dbc 100644 --- a/docs/topics/coroutines.rst +++ b/docs/topics/coroutines.rst @@ -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 `, 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: @@ -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`: diff --git a/docs/topics/exporters.rst b/docs/topics/exporters.rst index 56b995e18..528f749eb 100644 --- a/docs/topics/exporters.rst +++ b/docs/topics/exporters.rst @@ -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 diff --git a/docs/topics/feed-exports.rst b/docs/topics/feed-exports.rst index 467abc989..20ce922dc 100644 --- a/docs/topics/feed-exports.rst +++ b/docs/topics/feed-exports.rst @@ -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 +`) 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 -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 diff --git a/scrapy/exporters.py b/scrapy/exporters.py index 7f8aaf059..787109313 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 `. + """ @staticmethod def _get_populated_field_names(adapter: ItemAdapter) -> Iterable[str]: diff --git a/scrapy/extensions/feedexport.py b/scrapy/extensions/feedexport.py index 874023e25..728f1d5a7 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 @@ -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 ` 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 ` storage backend.""" + def __init__( self, uri: str, @@ -184,6 +214,12 @@ class StdoutFeedStorage: class FileFeedStorage: + """:ref:`Local filesystem ` 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 ` storage backend.""" + def __init__( self, uri: str, @@ -297,6 +335,8 @@ class S3FeedStorage(BlockingFeedStorage): class GCSFeedStorage(BlockingFeedStorage): + """:ref:`GCS ` storage backend.""" + def __init__( self, uri: str, @@ -348,6 +388,9 @@ class GCSFeedStorage(BlockingFeedStorage): class FTPFeedStorage(BlockingFeedStorage): + """:ref:`FTP ` storage backend, which also handles + :ref:`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 ( diff --git a/tests/test_exporters.py b/tests/test_exporters.py index 4359f4ff4..01359ba86 100644 --- a/tests/test_exporters.py +++ b/tests/test_exporters.py @@ -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): diff --git a/tests/test_feedexport.py b/tests/test_feedexport.py index 8560dd364..bb52b0a59 100644 --- a/tests/test_feedexport.py +++ b/tests/test_feedexport.py @@ -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 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