diff --git a/docs/topics/asyncio.rst b/docs/topics/asyncio.rst index 63bde8279..8efa2559e 100644 --- a/docs/topics/asyncio.rst +++ b/docs/topics/asyncio.rst @@ -177,7 +177,6 @@ in future Scrapy versions. The following features are not available: :class:`~scrapy.core.downloader.handlers._httpx.HttpxDownloadHandler`) * :class:`~scrapy.core.downloader.handlers.ftp.FTPDownloadHandler` * :class:`~scrapy.core.downloader.handlers.http2.H2DownloadHandler` -* :ref:`topics-shell` * :ref:`topics-telnetconsole` * :class:`~scrapy.crawler.CrawlerRunner` and :class:`~scrapy.crawler.CrawlerProcess` diff --git a/docs/topics/shell.rst b/docs/topics/shell.rst index ca5457b2a..8ae8ff512 100644 --- a/docs/topics/shell.rst +++ b/docs/topics/shell.rst @@ -17,9 +17,6 @@ spider, without having to run the spider to test every change. Once you get familiarized with the Scrapy shell, you'll see that it's an invaluable tool for developing and debugging your spiders. -.. note:: - This feature is not supported when :setting:`TWISTED_REACTOR_ENABLED` is ``False``. - Configuring the shell ===================== diff --git a/scrapy/cmdline.py b/scrapy/cmdline.py index c43aa626d..6c306afdb 100644 --- a/scrapy/cmdline.py +++ b/scrapy/cmdline.py @@ -204,9 +204,10 @@ def execute(argv: list[str] | None = None, settings: Settings | None = None) -> _run_print_help(parser, cmd.process_options, args, opts) if cmd.requires_crawler_process: - if settings[ - "TWISTED_REACTOR" - ] == _asyncio_reactor_path and not settings.getbool("FORCE_CRAWLER_PROCESS"): + if ( + settings["TWISTED_REACTOR"] == _asyncio_reactor_path + and not settings.getbool("FORCE_CRAWLER_PROCESS") + ) or not settings.getbool("TWISTED_REACTOR_ENABLED"): cmd.crawler_process = AsyncCrawlerProcess(settings) else: cmd.crawler_process = CrawlerProcess(settings) diff --git a/scrapy/commands/shell.py b/scrapy/commands/shell.py index 8db5f4b49..19138ffd0 100644 --- a/scrapy/commands/shell.py +++ b/scrapy/commands/shell.py @@ -6,10 +6,12 @@ See documentation in docs/topics/shell.rst from __future__ import annotations +import asyncio from threading import Thread from typing import TYPE_CHECKING, Any, ClassVar from scrapy.commands import ScrapyCommand +from scrapy.crawler import AsyncCrawlerProcess, Crawler from scrapy.http import Request from scrapy.shell import Shell from scrapy.utils.defer import _schedule_coro @@ -83,20 +85,47 @@ class Command(ScrapyCommand): # crawling engine, so the set up in the crawl method won't work crawler = self.crawler_process._create_crawler(spidercls) crawler._apply_settings() - if not crawler.settings.getbool("TWISTED_REACTOR_ENABLED"): - raise RuntimeError( - "scrapy shell currently doesn't support TWISTED_REACTOR_ENABLED=False" - ) - # The Shell class needs a persistent engine in the crawler - crawler.engine = crawler._create_engine() - _schedule_coro(crawler.engine.start_async(_start_request_processing=False)) - - self._start_crawler_thread() - - shell = Shell(crawler, update_vars=self.update_vars, code=opts.code) + loop: asyncio.AbstractEventLoop | None = None + if crawler.settings.getbool("TWISTED_REACTOR_ENABLED"): + self._init_with_reactor(crawler) + else: + self._init_without_reactor(crawler) + loop = self._get_reactorless_loop() + shell = Shell(crawler, update_vars=self.update_vars, code=opts.code, loop=loop) shell.start(url=url, redirect=not opts.no_redirect) + def _init_with_reactor(self, crawler: Crawler) -> None: + # Create the engine and run start_async() in the main thread + crawler.engine = crawler._create_engine() + _schedule_coro(crawler.engine.start_async(_start_request_processing=False)) + self._start_crawler_thread() + + def _init_without_reactor(self, crawler: Crawler) -> None: + # Create the engine and run start_async() in the event loop thread + loop = self._get_reactorless_loop() + self._start_crawler_thread() + + async def _init_engine() -> None: + # We may need to wait until some parts of start_async() have + # finished, which may need a special event in the engine and may + # wait until https://github.com/scrapy/scrapy/issues/6916 + crawler.engine = crawler._create_engine() + loop.create_task( + crawler.engine.start_async(_start_request_processing=False) + ) + + future = asyncio.run_coroutine_threadsafe(_init_engine(), loop) + future.result() + + def _get_reactorless_loop(self) -> asyncio.AbstractEventLoop: + assert self.crawler_process + assert isinstance(self.crawler_process, AsyncCrawlerProcess) + loop = self.crawler_process._reactorless_loop + assert loop + return loop + def _start_crawler_thread(self) -> None: + """Run self.crawler_process.start() in a separate thread.""" assert self.crawler_process t = Thread( target=self.crawler_process.start, diff --git a/scrapy/shell.py b/scrapy/shell.py index 6a9464d74..0966d9f55 100644 --- a/scrapy/shell.py +++ b/scrapy/shell.py @@ -6,34 +6,95 @@ See documentation in docs/topics/shell.rst from __future__ import annotations +import asyncio import contextlib import os import signal +import warnings from typing import TYPE_CHECKING, Any from itemadapter import is_item -from twisted.internet import defer, threads +from twisted.internet import threads +from twisted.internet.defer import Deferred from twisted.python import threadable from w3lib.url import any_to_uri import scrapy from scrapy.crawler import Crawler -from scrapy.exceptions import IgnoreRequest +from scrapy.exceptions import IgnoreRequest, ScrapyDeprecationWarning from scrapy.http import Request, Response from scrapy.settings import Settings from scrapy.spiders import Spider from scrapy.utils.conf import get_config from scrapy.utils.console import DEFAULT_PYTHON_SHELLS, start_python_console from scrapy.utils.datatypes import SequenceExclude -from scrapy.utils.defer import _schedule_coro, deferred_f_from_coro_f +from scrapy.utils.defer import ( + _schedule_coro, + deferred_f_from_coro_f, + maybe_deferred_to_future, +) from scrapy.utils.misc import load_object -from scrapy.utils.python import global_object_name from scrapy.utils.reactor import is_asyncio_reactor_installed, set_asyncio_event_loop from scrapy.utils.response import open_in_browser if TYPE_CHECKING: from collections.abc import Callable +# Hopefully temporary architecture notes +# +# The Shell class is always instantiated in the "main" thread. There are two +# official ways to use it: +# 1. scrapy.commands.shell, which makes a secondary thread and calls +# CrawlerProcess.start() in it, which runs a reactor there. +# 2. scrapy.shell.inspect_response(), which just creates Shell() in the current +# thread. +# +# Shell._inthread is True when this class is run in a thread separate from the +# reactor, e.g. the 1st way (in other words, the reactor is in a secondary +# thread). +# Shell._inthread is False when this class is run in the same thread as the +# reactor, e.g. the 2nd way. +# The only thing that differs is availability of fetch() (it needs the +# reactor to be in a separate thread: the shell sends the request to +# the reactor and waits for the result synchronously). +# +# Thus the only thing Shell needs an event loop for is fetch(). More machinery +# is used for it to work. In chronological order: +# 1. scrapy.commands.shell.Command.run() creates a crawler and an engine, then +# calls +# _schedule_coro(crawler.engine.start_async(_start_request_processing=False)), +# which initializes the engine but doesn't start processing of requests. +# 2. scrapy.commands.shell.Command.run() calls crawler_process.start() in a +# thread which starts a reactor in that thread. +# 3. When fetch() is called, it prepares a request and calls Shell._schedule() +# in the reactor thread (via threads.blockingCallFromThread()). +# 4. Shell._schedule() calls Shell._open_spider() (on the first call). +# 5. Shell._open_spider() calls engine.open_spider_async(close_if_idle=False) +# and engine._start_request_processing(). +# 6. Shell._schedule() calls engine.crawl(request), scheduling the request. +# 7. Shell._schedule() via _request_deferred() waits until the request callback +# is called. When it's called, the response becomes available. +# +# In the reactorless mode this is slightly different, the engine initialization +# happens in the event loop thread as many things need either a reactor or a +# running event loop. +# +# Side note: it should be possible to remove _request_deferred() by using +# engine.download() instead of engine.schedule(), losing the usual stuff like +# spider middlewares (none of which should be important). +# +# Other architecture problems: +# * scrapy.cmdline.execute() creates an AsyncCrawlerProcess instance which +# immediately installs a reactor (which is maybe not thread-specific?) or an +# event loop (which *is* thread-specific, so the main thread will always have +# a (not running) loop installed. +# * scrapy.commands.shell.Command.run() calls _schedule_coro() in the main +# thread, and various engine init code also calls similar things, +# conceptually this shouldn't work (and doesn't in the reactorless mode, so +# there the initialization is moved to the event loop thread). +# * The engine has several code paths specifically for the shell, and the shell +# uses several private members of the engine and of AsyncCrawlerProcess. + class Shell: relevant_classes: tuple[type, ...] = (Crawler, Spider, Request, Response, Settings) @@ -43,21 +104,47 @@ class Shell: crawler: Crawler, update_vars: Callable[[dict[str, Any]], None] | None = None, code: str | None = None, + *, + loop: asyncio.AbstractEventLoop | None = None, ): - self.crawler: Crawler = crawler - if not crawler.settings.getbool("TWISTED_REACTOR_ENABLED"): # pragma: no cover + self._use_reactor = crawler.settings.getbool("TWISTED_REACTOR_ENABLED") + if not self._use_reactor and not loop: # pragma: no cover raise RuntimeError( - f"{global_object_name(self.__class__)} currently doesn't support TWISTED_REACTOR_ENABLED=False." + "Shell needs the crawler loop reference when TWISTED_REACTOR_ENABLED=False." ) + self._loop = loop + self.crawler: Crawler = crawler self.update_vars: Callable[[dict[str, Any]], None] = update_vars or ( lambda x: None ) self.item_class: type = load_object(crawler.settings["DEFAULT_ITEM_CLASS"]) self.spider: Spider | None = None - self.inthread: bool = not threadable.isInIOThread() + if self._use_reactor: + self._inthread: bool = not threadable.isInIOThread() + else: + try: + # in case there is also a running loop in the main thread + current_loop = asyncio.get_running_loop() + self._inthread = current_loop is not self._loop + except RuntimeError: + self._inthread = True self.code: str | None = code self.vars: dict[str, Any] = {} + @property + def inthread(self) -> bool: # pragma: no cover + warnings.warn( + "Shell.inthread is deprecated, use Shell.fetch_available instead.", + ScrapyDeprecationWarning, + stacklevel=2, + ) + return self._inthread + + @property + def fetch_available(self) -> bool: + """Whether fetch() can be used.""" + return self._inthread + def start( self, url: str | None = None, @@ -103,28 +190,24 @@ class Shell: self.vars, shells=shells, banner=self.vars.pop("banner", "") ) - def _schedule(self, request: Request, spider: Spider | None) -> defer.Deferred[Any]: - if is_asyncio_reactor_installed(): + async def _schedule(self, request: Request, spider: Spider | None) -> Response: + """Send the request to the engine, wait for the result. + + Runs in the reactor thread. + """ + if self._use_reactor and is_asyncio_reactor_installed(): # set the asyncio event loop for the current thread event_loop_path = self.crawler.settings["ASYNCIO_EVENT_LOOP"] set_asyncio_event_loop(event_loop_path) + if not self.spider: + await self._open_spider(spider) + assert self.crawler.engine is not None + # send the request to the engine + self.crawler.engine.crawl(request) + # this will fire when the request callback runs (via the callback hijacking in _request_deferred()) + return await maybe_deferred_to_future(_request_deferred(request)) - def crawl_request(_: None) -> None: - assert self.crawler.engine is not None - self.crawler.engine.crawl(request) - - d2 = self._open_spider(request, spider) - d2.addCallback(crawl_request) - - d = _request_deferred(request) - d.addCallback(lambda x: (x, spider)) - return d - - @deferred_f_from_coro_f - async def _open_spider(self, request: Request, spider: Spider | None) -> None: - if self.spider: - return - + async def _open_spider(self, spider: Spider | None) -> None: if spider is None: spider = self.crawler.spider or self.crawler._create_spider() @@ -141,8 +224,6 @@ class Shell: redirect: bool = True, **kwargs: Any, ) -> None: - from twisted.internet import reactor - if isinstance(request_or_url, Request): request = request_or_url else: @@ -154,12 +235,22 @@ class Shell: ) else: request.meta["handle_httpstatus_all"] = True - response = None - with contextlib.suppress(IgnoreRequest): - response, spider = threads.blockingCallFromThread( - reactor, self._schedule, request, spider - ) - self.populate_vars(response, request, spider) + response: Response | None = None + if self._use_reactor: + from twisted.internet import reactor + + with contextlib.suppress(IgnoreRequest): + response = threads.blockingCallFromThread( + reactor, deferred_f_from_coro_f(self._schedule), request, spider + ) + else: + assert self._loop + with contextlib.suppress(IgnoreRequest): + future = asyncio.run_coroutine_threadsafe( + self._schedule(request, spider), self._loop + ) + response = future.result() + self.populate_vars(response, request, self.spider) def populate_vars( self, @@ -174,7 +265,7 @@ class Shell: self.vars["spider"] = spider self.vars["request"] = request self.vars["response"] = response - if self.inthread: + if self.fetch_available: self.vars["fetch"] = self.fetch self.vars["view"] = open_in_browser self.vars["shelp"] = self.print_help @@ -195,7 +286,7 @@ class Shell: if self._is_relevant(v): b.append(f" {k:<10} {v}") b.append("Useful shortcuts:") - if self.inthread: + if self.fetch_available: b.append( " fetch(url[, redirect=True]) " "Fetch URL and update local objects (by default, redirects are followed)" @@ -218,11 +309,15 @@ def inspect_response(response: Response, spider: Spider) -> None: # Shell.start removes the SIGINT handler, so save it and re-add it after # the shell has closed sigint_handler = signal.getsignal(signal.SIGINT) - Shell(spider.crawler).start(response=response, spider=spider) + if not spider.crawler.settings.getbool("TWISTED_REACTOR_ENABLED"): + loop = asyncio.get_running_loop() + else: + loop = None + Shell(spider.crawler, loop=loop).start(response=response, spider=spider) signal.signal(signal.SIGINT, sigint_handler) -def _request_deferred(request: Request) -> defer.Deferred[Any]: +def _request_deferred(request: Request) -> Deferred[Any]: """Wrap a request inside a Deferred. This function is harmful, do not use it until you know what you are doing. @@ -241,7 +336,7 @@ def _request_deferred(request: Request) -> defer.Deferred[Any]: request.errback = request_errback return result - d: defer.Deferred[Any] = defer.Deferred() + d: Deferred[Any] = Deferred() d.addBoth(_restore_callbacks) if request.callback: d.addCallback(request.callback) diff --git a/tests/test_command_shell.py b/tests/test_command_shell.py index 6a12f28bb..b407211dd 100644 --- a/tests/test_command_shell.py +++ b/tests/test_command_shell.py @@ -22,6 +22,12 @@ class TestShellCommand: _, out, _ = proc("shell", "-c", "item") assert "{}" in out + def test_empty_no_reactor(self) -> None: + _, out, _ = proc( + "shell", "-c", "item", "--set", "TWISTED_REACTOR_ENABLED=False" + ) + assert "{}" in out + def test_response_body(self, mockserver: MockServer) -> None: _, out, _ = proc("shell", mockserver.url("/text"), "-c", "response.body") assert "Works" in out @@ -125,7 +131,6 @@ class TestShellCommand: assert ret == 0, err assert "RuntimeError: There is no current event loop in thread" not in err - @pytest.mark.xfail(reason="Not implemented yet", strict=True) def test_shell_fetch_no_reactor(self, mockserver: MockServer) -> None: url = mockserver.url("/html") code = f"fetch('{url}')" @@ -134,17 +139,6 @@ class TestShellCommand: ) assert ret == 0, err - def test_no_reactor_unsupported(self) -> None: - # to be removed when it's supported - ret, out, err = proc( - "shell", "-c", "item", "--set", "TWISTED_REACTOR_ENABLED=False" - ) - assert ret == 1, out or err - assert ( - "RuntimeError: scrapy shell currently doesn't support TWISTED_REACTOR_ENABLED=False" - in err - ) - class TestInteractiveShell: def test_fetch(self, mockserver: MockServer) -> None: