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

Allow item exporter and feed storage methods to be coroutines

This commit is contained in:
Adrian Chaves 2026-07-28 17:19:26 +02:00
parent ad816d2b3a
commit 099bcd0386
7 changed files with 376 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
docs/topics/coroutines.rst
View File

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

18
docs/topics/exporters.rst
View File

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

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

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

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>`.
"""
def _get_serialized_fields(
self, item: Any, default_value: Any = None, include_empty: bool | None = None

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

10
tests/test_feedexport.py
View File

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

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