Repository: scrapy/scrapy
Pull request #5876 (2023-03-27 14:41:26)

Add feed_slot_closed and feed_exporter_closed signals to work with FeedExporter
The signals will be sent when a slot and the FeedExporter are closed respectively. 

This allows for the use of extensions related to feeds (like email feeds or slack notifications) that may require them to be used without the need for a custom FeedExporter.


Patch:
diff --git a/.bandit.yml b/.bandit.yml
index c8e84cc2e..2aae8a0aa 100644
--- a/.bandit.yml
+++ b/.bandit.yml
@@ -1,5 +1,6 @@
 skips:
 - B101
+- B113  # https://github.com/PyCQA/bandit/issues/1010
 - B105
 - B301
 - B303
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 4b90233e5..faf8808f2 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -1,6 +1,6 @@
 repos:
 - repo: https://github.com/PyCQA/bandit
-  rev: 1.7.4
+  rev: 1.7.5
   hooks:
   - id: bandit
     args: [-r, -c, .bandit.yml]
@@ -9,7 +9,7 @@ repos:
   hooks:
   - id: flake8
 - repo: https://github.com/psf/black.git
-  rev: 23.1.0
+  rev: 23.3.0
   hooks:
   - id: black
 - repo: https://github.com/pycqa/isort
@@ -21,4 +21,4 @@ repos:
   hooks:
   - id: blacken-docs
     additional_dependencies:
-    - black==23.1.0
+    - black==23.3.0
diff --git a/docs/Makefile b/docs/Makefile
index 596cb6cef..48401bac8 100644
--- a/docs/Makefile
+++ b/docs/Makefile
@@ -87,7 +87,7 @@ coverage: build
 
 htmlview: html
 	 $(PYTHON) -c "import webbrowser; from pathlib import Path; \
-	 webbrowser.open('file://' + Path('build/html/index.html').resolve())"
+	 webbrowser.open(Path('build/html/index.html').resolve().as_uri())"
 
 clean:
 	-rm -rf build/*
diff --git a/docs/_templates/layout.html b/docs/_templates/layout.html
deleted file mode 100644
index 18a5231ee..000000000
--- a/docs/_templates/layout.html
+++ /dev/null
@@ -1,11 +0,0 @@
-{% extends "!layout.html" %}
-
-{% block footer %}
-{{ super() }}
-<script type="text/javascript">
-analytics.ready(function () {
-    ga('require', 'linker');
-    ga('linker:autoLink', ['zyte.com']);
-});
-</script>
-{% endblock %}
diff --git a/docs/contributing.rst b/docs/contributing.rst
index 6b1a41339..eef92e148 100644
--- a/docs/contributing.rst
+++ b/docs/contributing.rst
@@ -11,10 +11,6 @@ Contributing to Scrapy
 
 There are many ways to contribute to Scrapy. Here are some of them:
 
-* Blog about Scrapy. Tell the world how you're using Scrapy. This will help
-  newcomers with more examples and will help the Scrapy project to increase its
-  visibility.
-
 * Report bugs and request features in the `issue tracker`_, trying to follow
   the guidelines detailed in `Reporting bugs`_ below.
 
@@ -22,13 +18,16 @@ There are many ways to contribute to Scrapy. Here are some of them:
   :ref:`writing-patches` and `Submitting patches`_ below for details on how to
   write and submit a patch.
 
+* Blog about Scrapy. Tell the world how you're using Scrapy. This will help
+  newcomers with more examples and will help the Scrapy project to increase its
+  visibility.
+
 * Join the `Scrapy subreddit`_ and share your ideas on how to
   improve Scrapy. We're always open to suggestions.
 
 * Answer Scrapy questions at
   `Stack Overflow <https://stackoverflow.com/questions/tagged/scrapy>`__.
 
-
 Reporting bugs
 ==============
 
@@ -80,6 +79,13 @@ guidelines when you're going to report a new bug.
 Writing patches
 ===============
 
+Scrapy has a list of `good first issues`_ and `help wanted issues`_ that you
+can work on. These issues are a great way to get started with contributing to
+Scrapy. If you're new to the codebase, you may want to focus on documentation
+or testing-related issues, as they are always useful and can help you get
+more familiar with the project. You can also check Scrapy's `test coverage`_
+to see which areas may benefit from more tests.
+
 The better a patch is written, the higher the chances that it'll get accepted and the sooner it will be merged.
 
 Well-written patches should:
@@ -314,3 +320,6 @@ And their unit-tests are in::
 .. _PEP 257: https://www.python.org/dev/peps/pep-0257/
 .. _pull request: https://help.github.com/en/github/collaborating-with-issues-and-pull-requests/creating-a-pull-request
 .. _pytest-xdist: https://github.com/pytest-dev/pytest-xdist
+.. _good first issues: https://github.com/scrapy/scrapy/issues?q=is%3Aissue+is%3Aopen+label%3A%22good+first+issue%22
+.. _help wanted issues: https://github.com/scrapy/scrapy/issues?q=is%3Aissue+is%3Aopen+label%3A%22help+wanted%22
+.. _test coverage: https://app.codecov.io/gh/scrapy/scrapy
\ No newline at end of file
diff --git a/docs/faq.rst b/docs/faq.rst
index 031f4b942..20dd814df 100644
--- a/docs/faq.rst
+++ b/docs/faq.rst
@@ -231,8 +231,8 @@ Can I return (Twisted) deferreds from signal handlers?
 Some signals support returning deferreds from their handlers, others don't. See
 the :ref:`topics-signals-ref` to know which ones.
 
-What does the response status code 999 means?
----------------------------------------------
+What does the response status code 999 mean?
+--------------------------------------------
 
 999 is a custom response status code used by Yahoo sites to throttle requests.
 Try slowing down the crawling speed by using a download delay of ``2`` (or
diff --git a/docs/intro/tutorial.rst b/docs/intro/tutorial.rst
index 064ce05f8..19a76fc16 100644
--- a/docs/intro/tutorial.rst
+++ b/docs/intro/tutorial.rst
@@ -252,7 +252,7 @@ object:
 .. code-block:: pycon
 
     >>> response.css("title")
-    [<Selector xpath='descendant-or-self::title' data='<title>Quotes to Scrape</title>'>]
+    [<Selector query='descendant-or-self::title' data='<title>Quotes to Scrape</title>'>]
 
 The result of running ``response.css('title')`` is a list-like object called
 :class:`~scrapy.selector.SelectorList`, which represents a list of
@@ -329,7 +329,7 @@ the :meth:`~scrapy.selector.SelectorList.re` method to extract using
     >>> response.css("title::text").re(r"(\w+) to (\w+)")
     ['Quotes', 'Scrape']
 
-In order to find the proper CSS selectors to use, you might find useful opening
+In order to find the proper CSS selectors to use, you might find it useful to open
 the response page from the shell in your web browser using ``view(response)``.
 You can use your browser's developer tools to inspect the HTML and come up
 with a selector (see :ref:`topics-developer-tools`).
@@ -348,7 +348,7 @@ Besides `CSS`_, Scrapy selectors also support using `XPath`_ expressions:
 .. code-block:: pycon
 
     >>> response.xpath("//title")
-    [<Selector xpath='//title' data='<title>Quotes to Scrape</title>'>]
+    [<Selector query='//title' data='<title>Quotes to Scrape</title>'>]
     >>> response.xpath("//title/text()").get()
     'Quotes to Scrape'
 
@@ -410,8 +410,8 @@ We get a list of selectors for the quote HTML elements with:
 .. code-block:: pycon
 
     >>> response.css("div.quote")
-    [<Selector xpath="descendant-or-self::div[@class and contains(concat(' ', normalize-space(@class), ' '), ' quote ')]" data='<div class="quote" itemscope itemtype...'>,
-    <Selector xpath="descendant-or-self::div[@class and contains(concat(' ', normalize-space(@class), ' '), ' quote ')]" data='<div class="quote" itemscope itemtype...'>,
+    [<Selector query="descendant-or-self::div[@class and contains(concat(' ', normalize-space(@class), ' '), ' quote ')]" data='<div class="quote" itemscope itemtype...'>,
+    <Selector query="descendant-or-self::div[@class and contains(concat(' ', normalize-space(@class), ' '), ' quote ')]" data='<div class="quote" itemscope itemtype...'>,
     ...]
 
 Each of the selectors returned by the query above allows us to run further
diff --git a/docs/topics/commands.rst b/docs/topics/commands.rst
index 106045fc0..1d37895c2 100644
--- a/docs/topics/commands.rst
+++ b/docs/topics/commands.rst
@@ -285,13 +285,13 @@ Usage examples::
     $ scrapy crawl myspider
     [ ... myspider starts crawling ... ]
 
-    $ scrapy -o myfile:csv myspider
+    $ scrapy crawl -o myfile:csv myspider
     [ ... myspider starts crawling and appends the result to the file myfile in csv format ... ]
 
-    $ scrapy -O myfile:json myspider
+    $ scrapy crawl -O myfile:json myspider
     [ ... myspider starts crawling and saves the result in myfile in json format overwriting the original content... ]
 
-    $ scrapy -o myfile -t csv myspider
+    $ scrapy crawl -o myfile -t csv myspider
     [ ... myspider starts crawling and appends the result to the file myfile in csv format ... ]
 
 .. command:: check
diff --git a/docs/topics/dynamic-content.rst b/docs/topics/dynamic-content.rst
index d01e0a8d4..a0f4b4411 100644
--- a/docs/topics/dynamic-content.rst
+++ b/docs/topics/dynamic-content.rst
@@ -276,7 +276,7 @@ The following is a simple snippet to illustrate its usage within a Scrapy spider
             async with async_playwright() as pw:
                 browser = await pw.chromium.launch()
                 page = await browser.new_page()
-                await page.goto("https:/example.org")
+                await page.goto("https://example.org")
                 title = await page.title()
                 return {"title": title}
 
diff --git a/docs/topics/signals.rst b/docs/topics/signals.rst
index 3400a205a..9bfd1761c 100644
--- a/docs/topics/signals.rst
+++ b/docs/topics/signals.rst
@@ -307,6 +307,33 @@ spider_error
     :param spider: the spider which raised the exception
     :type spider: :class:`~scrapy.Spider` object
 
+feed_slot_closed
+~~~~~~~~~~~~~~~~
+
+.. signal:: feed_slot_closed
+.. function:: feed_slot_closed(slot)
+
+    Sent when a :ref:`feed exports <topics-feed-exports>` slot is closed.
+
+    This signal supports returning deferreds from its handlers.
+
+    :param slot: the slot closed
+    :type slot: scrapy.extensions.feedexport.FeedSlot
+
+
+feed_exporter_closed
+~~~~~~~~~~~~~~~~~~~~
+
+.. signal:: feed_exporter_closed
+.. function:: feed_exporter_closed()
+
+    Sent when the :ref:`feed exports <topics-feed-exports>` extension is closed,
+    during the handling of the :signal:`spider_closed` signal by the extension,
+    after all feed exporting has been handled.
+
+    This signal supports returning deferreds from its handlers.
+
+
 Request signals
 ---------------
 
diff --git a/scrapy/core/downloader/__init__.py b/scrapy/core/downloader/__init__.py
index 1b83c3a8a..426056dc8 100644
--- a/scrapy/core/downloader/__init__.py
+++ b/scrapy/core/downloader/__init__.py
@@ -2,44 +2,52 @@ import random
 from collections import deque
 from datetime import datetime
 from time import time
+from typing import TYPE_CHECKING, Any, Deque, Dict, Set, Tuple, cast
 
-from twisted.internet import defer, task
+from twisted.internet import task
+from twisted.internet.defer import Deferred
 
-from scrapy import signals
+from scrapy import Request, Spider, signals
 from scrapy.core.downloader.handlers import DownloadHandlers
 from scrapy.core.downloader.middleware import DownloaderMiddlewareManager
+from scrapy.http import Response
 from scrapy.resolver import dnscache
+from scrapy.settings import BaseSettings
+from scrapy.signalmanager import SignalManager
 from scrapy.utils.defer import mustbe_deferred
 from scrapy.utils.httpobj import urlparse_cached
 
+if TYPE_CHECKING:
+    from scrapy.crawler import Crawler
+
 
 class Slot:
     """Downloader slot"""
 
-    def __init__(self, concurrency, delay, randomize_delay):
-        self.concurrency = concurrency
-        self.delay = delay
-        self.randomize_delay = randomize_delay
+    def __init__(self, concurrency: int, delay: float, randomize_delay: bool):
+        self.concurrency: int = concurrency
+        self.delay: float = delay
+        self.randomize_delay: bool = randomize_delay
 
-        self.active = set()
-        self.queue = deque()
-        self.transferring = set()
-        self.lastseen = 0
+        self.active: Set[Request] = set()
+        self.queue: Deque[Tuple[Request, Deferred]] = deque()
+        self.transferring: Set[Request] = set()
+        self.lastseen: float = 0
         self.latercall = None
 
-    def free_transfer_slots(self):
+    def free_transfer_slots(self) -> int:
         return self.concurrency - len(self.transferring)
 
-    def download_delay(self):
+    def download_delay(self) -> float:
         if self.randomize_delay:
             return random.uniform(0.5 * self.delay, 1.5 * self.delay)
         return self.delay
 
-    def close(self):
+    def close(self) -> None:
         if self.latercall and self.latercall.active():
             self.latercall.cancel()
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         cls_name = self.__class__.__name__
         return (
             f"{cls_name}(concurrency={self.concurrency!r}, "
@@ -47,7 +55,7 @@ class Slot:
             f"randomize_delay={self.randomize_delay!r})"
         )
 
-    def __str__(self):
+    def __str__(self) -> str:
         return (
             f"<downloader.Slot concurrency={self.concurrency!r} "
             f"delay={self.delay:.2f} randomize_delay={self.randomize_delay!r} "
@@ -57,8 +65,10 @@ class Slot:
         )
 
 
-def _get_concurrency_delay(concurrency, spider, settings):
-    delay = settings.getfloat("DOWNLOAD_DELAY")
+def _get_concurrency_delay(
+    concurrency: int, spider: Spider, settings: BaseSettings
+) -> Tuple[int, float]:
+    delay: float = settings.getfloat("DOWNLOAD_DELAY")
     if hasattr(spider, "download_delay"):
         delay = spider.download_delay
 
@@ -71,23 +81,29 @@ def _get_concurrency_delay(concurrency, spider, settings):
 class Downloader:
     DOWNLOAD_SLOT = "download_slot"
 
-    def __init__(self, crawler):
-        self.settings = crawler.settings
-        self.signals = crawler.signals
-        self.slots = {}
-        self.active = set()
-        self.handlers = DownloadHandlers(crawler)
-        self.total_concurrency = self.settings.getint("CONCURRENT_REQUESTS")
-        self.domain_concurrency = self.settings.getint("CONCURRENT_REQUESTS_PER_DOMAIN")
-        self.ip_concurrency = self.settings.getint("CONCURRENT_REQUESTS_PER_IP")
-        self.randomize_delay = self.settings.getbool("RANDOMIZE_DOWNLOAD_DELAY")
-        self.middleware = DownloaderMiddlewareManager.from_crawler(crawler)
-        self._slot_gc_loop = task.LoopingCall(self._slot_gc)
+    def __init__(self, crawler: "Crawler"):
+        self.settings: BaseSettings = crawler.settings
+        self.signals: SignalManager = crawler.signals
+        self.slots: Dict[str, Slot] = {}
+        self.active: Set[Request] = set()
+        self.handlers: DownloadHandlers = DownloadHandlers(crawler)
+        self.total_concurrency: int = self.settings.getint("CONCURRENT_REQUESTS")
+        self.domain_concurrency: int = self.settings.getint(
+            "CONCURRENT_REQUESTS_PER_DOMAIN"
+        )
+        self.ip_concurrency: int = self.settings.getint("CONCURRENT_REQUESTS_PER_IP")
+        self.randomize_delay: bool = self.settings.getbool("RANDOMIZE_DOWNLOAD_DELAY")
+        self.middleware: DownloaderMiddlewareManager = (
+            DownloaderMiddlewareManager.from_crawler(crawler)
+        )
+        self._slot_gc_loop: task.LoopingCall = task.LoopingCall(self._slot_gc)
         self._slot_gc_loop.start(60)
-        self.per_slot_settings = self.settings.getdict("DOWNLOAD_SLOTS", {})
+        self.per_slot_settings: Dict[str, Dict[str, Any]] = self.settings.getdict(
+            "DOWNLOAD_SLOTS", {}
+        )
 
-    def fetch(self, request, spider):
-        def _deactivate(response):
+    def fetch(self, request: Request, spider: Spider) -> Deferred:
+        def _deactivate(response: Response) -> Response:
             self.active.remove(request)
             return response
 
@@ -98,7 +114,7 @@ class Downloader:
     def needs_backout(self) -> bool:
         return len(self.active) >= self.total_concurrency
 
-    def _get_slot(self, request, spider):
+    def _get_slot(self, request: Request, spider: Spider) -> Tuple[str, Slot]:
         key = self._get_slot_key(request, spider)
         if key not in self.slots:
             slot_settings = self.per_slot_settings.get(key, {})
@@ -116,9 +132,9 @@ class Downloader:
 
         return key, self.slots[key]
 
-    def _get_slot_key(self, request, spider):
+    def _get_slot_key(self, request: Request, spider: Spider) -> str:
         if self.DOWNLOAD_SLOT in request.meta:
-            return request.meta[self.DOWNLOAD_SLOT]
+            return cast(str, request.meta[self.DOWNLOAD_SLOT])
 
         key = urlparse_cached(request).hostname or ""
         if self.ip_concurrency:
@@ -126,11 +142,11 @@ class Downloader:
 
         return key
 
-    def _enqueue_request(self, request, spider):
+    def _enqueue_request(self, request: Request, spider: Spider) -> Deferred:
         key, slot = self._get_slot(request, spider)
         request.meta[self.DOWNLOAD_SLOT] = key
 
-        def _deactivate(response):
+        def _deactivate(response: Response) -> Response:
             slot.active.remove(request)
             return response
 
@@ -138,12 +154,12 @@ class Downloader:
         self.signals.send_catch_log(
             signal=signals.request_reached_downloader, request=request, spider=spider
         )
-        deferred = defer.Deferred().addBoth(_deactivate)
+        deferred = Deferred().addBoth(_deactivate)
         slot.queue.append((request, deferred))
         self._process_queue(spider, slot)
         return deferred
 
-    def _process_queue(self, spider, slot):
+    def _process_queue(self, spider: Spider, slot: Slot) -> None:
         from twisted.internet import reactor
 
         if slot.latercall and slot.latercall.active():
@@ -171,7 +187,7 @@ class Downloader:
                 self._process_queue(spider, slot)
                 break
 
-    def _download(self, slot, request, spider):
+    def _download(self, slot: Slot, request: Request, spider: Spider) -> Deferred:
         # The order is very important for the following deferreds. Do not change!
 
         # 1. Create the download deferred
@@ -179,7 +195,7 @@ class Downloader:
 
         # 2. Notify response_downloaded listeners about the recent download
         # before querying queue for next request
-        def _downloaded(response):
+        def _downloaded(response: Response) -> Response:
             self.signals.send_catch_log(
                 signal=signals.response_downloaded,
                 response=response,
@@ -196,7 +212,7 @@ class Downloader:
         # middleware itself)
         slot.transferring.add(request)
 
-        def finish_transferring(_):
+        def finish_transferring(_: Any) -> Any:
             slot.transferring.remove(request)
             self._process_queue(spider, slot)
             self.signals.send_catch_log(
@@ -206,12 +222,12 @@ class Downloader:
 
         return dfd.addBoth(finish_transferring)
 
-    def close(self):
+    def close(self) -> None:
         self._slot_gc_loop.stop()
         for slot in self.slots.values():
             slot.close()
 
-    def _slot_gc(self, age=60):
+    def _slot_gc(self, age: float = 60) -> None:
         mintime = time() - age
         for key, slot in list(self.slots.items()):
             if not slot.active and slot.lastseen + slot.delay < mintime:
diff --git a/scrapy/core/downloader/contextfactory.py b/scrapy/core/downloader/contextfactory.py
index 53ae78918..909cc273f 100644
--- a/scrapy/core/downloader/contextfactory.py
+++ b/scrapy/core/downloader/contextfactory.py
@@ -1,4 +1,5 @@
 import warnings
+from typing import TYPE_CHECKING, Any, List, Optional
 
 from OpenSSL import SSL
 from twisted.internet._sslverify import _setAcceptableProtocols
@@ -18,8 +19,12 @@ from scrapy.core.downloader.tls import (
     ScrapyClientTLSOptions,
     openssl_methods,
 )
+from scrapy.settings import BaseSettings
 from scrapy.utils.misc import create_instance, load_object
 
+if TYPE_CHECKING:
+    from twisted.internet._sslverify import ClientTLSOptions
+
 
 @implementer(IPolicyForHTTPS)
 class ScrapyClientContextFactory(BrowserLikePolicyForHTTPS):
@@ -35,25 +40,34 @@ class ScrapyClientContextFactory(BrowserLikePolicyForHTTPS):
 
     def __init__(
         self,
-        method=SSL.SSLv23_METHOD,
-        tls_verbose_logging=False,
-        tls_ciphers=None,
-        *args,
-        **kwargs,
+        method: int = SSL.SSLv23_METHOD,
+        tls_verbose_logging: bool = False,
+        tls_ciphers: Optional[str] = None,
+        *args: Any,
+        **kwargs: Any,
     ):
         super().__init__(*args, **kwargs)
-        self._ssl_method = method
-        self.tls_verbose_logging = tls_verbose_logging
+        self._ssl_method: int = method
+        self.tls_verbose_logging: bool = tls_verbose_logging
+        self.tls_ciphers: AcceptableCiphers
         if tls_ciphers:
             self.tls_ciphers = AcceptableCiphers.fromOpenSSLCipherString(tls_ciphers)
         else:
             self.tls_ciphers = DEFAULT_CIPHERS
 
     @classmethod
-    def from_settings(cls, settings, method=SSL.SSLv23_METHOD, *args, **kwargs):
-        tls_verbose_logging = settings.getbool("DOWNLOADER_CLIENT_TLS_VERBOSE_LOGGING")
-        tls_ciphers = settings["DOWNLOADER_CLIENT_TLS_CIPHERS"]
-        return cls(
+    def from_settings(
+        cls,
+        settings: BaseSettings,
+        method: int = SSL.SSLv23_METHOD,
+        *args: Any,
+        **kwargs: Any,
+    ):
+        tls_verbose_logging: bool = settings.getbool(
+            "DOWNLOADER_CLIENT_TLS_VERBOSE_LOGGING"
+        )
+        tls_ciphers: Optional[str] = settings["DOWNLOADER_CLIENT_TLS_CIPHERS"]
+        return cls(  # type: ignore[misc]
             method=method,
             tls_verbose_logging=tls_verbose_logging,
             tls_ciphers=tls_ciphers,
@@ -61,7 +75,7 @@ class ScrapyClientContextFactory(BrowserLikePolicyForHTTPS):
             **kwargs,
         )
 
-    def getCertificateOptions(self):
+    def getCertificateOptions(self) -> CertificateOptions:
         # setting verify=True will require you to provide CAs
         # to verify against; in other words: it's not that simple
 
@@ -82,12 +96,12 @@ class ScrapyClientContextFactory(BrowserLikePolicyForHTTPS):
 
     # kept for old-style HTTP/1.0 downloader context twisted calls,
     # e.g. connectSSL()
-    def getContext(self, hostname=None, port=None):
+    def getContext(self, hostname: Any = None, port: Any = None) -> SSL.Context:
         ctx = self.getCertificateOptions().getContext()
         ctx.set_options(0x4)  # OP_LEGACY_SERVER_CONNECT
         return ctx
 
-    def creatorForNetloc(self, hostname, port):
+    def creatorForNetloc(self, hostname: bytes, port: int) -> "ClientTLSOptions":
         return ScrapyClientTLSOptions(
             hostname.decode("ascii"),
             self.getContext(),
@@ -114,7 +128,7 @@ class BrowserLikeContextFactory(ScrapyClientContextFactory):
     ``SSLv23_METHOD``) which allows TLS protocol negotiation.
     """
 
-    def creatorForNetloc(self, hostname, port):
+    def creatorForNetloc(self, hostname: bytes, port: int) -> "ClientTLSOptions":
         # trustRoot set to platformTrust() will use the platform's root CAs.
         #
         # This means that a website like https://www.cacert.org will be rejected
@@ -133,13 +147,15 @@ class AcceptableProtocolsContextFactory:
     negotiation.
     """
 
-    def __init__(self, context_factory, acceptable_protocols):
+    def __init__(self, context_factory: Any, acceptable_protocols: List[bytes]):
         verifyObject(IPolicyForHTTPS, context_factory)
-        self._wrapped_context_factory = context_factory
-        self._acceptable_protocols = acceptable_protocols
+        self._wrapped_context_factory: Any = context_factory
+        self._acceptable_protocols: List[bytes] = acceptable_protocols
 
-    def creatorForNetloc(self, hostname, port):
-        options = self._wrapped_context_factory.creatorForNetloc(hostname, port)
+    def creatorForNetloc(self, hostname: bytes, port: int) -> "ClientTLSOptions":
+        options: "ClientTLSOptions" = self._wrapped_context_factory.creatorForNetloc(
+            hostname, port
+        )
         _setAcceptableProtocols(options._ctx, self._acceptable_protocols)
         return options
 
diff --git a/scrapy/core/downloader/handlers/__init__.py b/scrapy/core/downloader/handlers/__init__.py
index 39155efe9..6a211aafa 100644
--- a/scrapy/core/downloader/handlers/__init__.py
+++ b/scrapy/core/downloader/handlers/__init__.py
@@ -1,25 +1,32 @@
 """Download handlers for different schemes"""
 
 import logging
+from typing import TYPE_CHECKING, Any, Callable, Dict, Generator, Union, cast
 
 from twisted.internet import defer
+from twisted.internet.defer import Deferred
 
-from scrapy import signals
+from scrapy import Request, Spider, signals
 from scrapy.exceptions import NotConfigured, NotSupported
 from scrapy.utils.httpobj import urlparse_cached
 from scrapy.utils.misc import create_instance, load_object
 from scrapy.utils.python import without_none_values
 
+if TYPE_CHECKING:
+    from scrapy.crawler import Crawler
+
 logger = logging.getLogger(__name__)
 
 
 class DownloadHandlers:
-    def __init__(self, crawler):
-        self._crawler = crawler
-        self._schemes = {}  # stores acceptable schemes on instancing
-        self._handlers = {}  # stores instanced handlers for schemes
-        self._notconfigured = {}  # remembers failed handlers
-        handlers = without_none_values(
+    def __init__(self, crawler: "Crawler"):
+        self._crawler: "Crawler" = crawler
+        self._schemes: Dict[
+            str, Union[str, Callable]
+        ] = {}  # stores acceptable schemes on instancing
+        self._handlers: Dict[str, Any] = {}  # stores instanced handlers for schemes
+        self._notconfigured: Dict[str, str] = {}  # remembers failed handlers
+        handlers: Dict[str, Union[str, Callable]] = without_none_values(
             crawler.settings.getwithbase("DOWNLOAD_HANDLERS")
         )
         for scheme, clspath in handlers.items():
@@ -28,7 +35,7 @@ class DownloadHandlers:
 
         crawler.signals.connect(self._close, signals.engine_stopped)
 
-    def _get_handler(self, scheme):
+    def _get_handler(self, scheme: str) -> Any:
         """Lazy-load the downloadhandler for a scheme
         only on the first request for that scheme.
         """
@@ -42,7 +49,7 @@ class DownloadHandlers:
 
         return self._load_handler(scheme)
 
-    def _load_handler(self, scheme, skip_lazy=False):
+    def _load_handler(self, scheme: str, skip_lazy: bool = False) -> Any:
         path = self._schemes[scheme]
         try:
             dhcls = load_object(path)
@@ -69,17 +76,17 @@ class DownloadHandlers:
             self._handlers[scheme] = dh
             return dh
 
-    def download_request(self, request, spider):
+    def download_request(self, request: Request, spider: Spider) -> Deferred:
         scheme = urlparse_cached(request).scheme
         handler = self._get_handler(scheme)
         if not handler:
             raise NotSupported(
                 f"Unsupported URL scheme '{scheme}': {self._notconfigured[scheme]}"
             )
-        return handler.download_request(request, spider)
+        return cast(Deferred, handler.download_request(request, spider))
 
     @defer.inlineCallbacks
-    def _close(self, *_a, **_kw):
+    def _close(self, *_a: Any, **_kw: Any) -> Generator[Deferred, Any, None]:
         for dh in self._handlers.values():
             if hasattr(dh, "close"):
                 yield dh.close()
diff --git a/scrapy/core/downloader/middleware.py b/scrapy/core/downloader/middleware.py
index 56df48b2e..dca13c01e 100644
--- a/scrapy/core/downloader/middleware.py
+++ b/scrapy/core/downloader/middleware.py
@@ -3,15 +3,16 @@ Downloader Middleware manager
 
 See documentation in docs/topics/downloader-middleware.rst
 """
-from typing import Callable, Union, cast
+from typing import Any, Callable, Generator, List, Union, cast
 
-from twisted.internet import defer
+from twisted.internet.defer import Deferred, inlineCallbacks
 from twisted.python.failure import Failure
 
 from scrapy import Spider
 from scrapy.exceptions import _InvalidOutput
 from scrapy.http import Request, Response
 from scrapy.middleware import MiddlewareManager
+from scrapy.settings import BaseSettings
 from scrapy.utils.conf import build_component_list
 from scrapy.utils.defer import deferred_from_coro, mustbe_deferred
 
@@ -20,10 +21,10 @@ class DownloaderMiddlewareManager(MiddlewareManager):
     component_name = "downloader middleware"
 
     @classmethod
-    def _get_mwlist_from_settings(cls, settings):
+    def _get_mwlist_from_settings(cls, settings: BaseSettings) -> List[Any]:
         return build_component_list(settings.getwithbase("DOWNLOADER_MIDDLEWARES"))
 
-    def _add_middleware(self, mw):
+    def _add_middleware(self, mw: Any) -> None:
         if hasattr(mw, "process_request"):
             self.methods["process_request"].append(mw.process_request)
         if hasattr(mw, "process_response"):
@@ -31,9 +32,11 @@ class DownloaderMiddlewareManager(MiddlewareManager):
         if hasattr(mw, "process_exception"):
             self.methods["process_exception"].appendleft(mw.process_exception)
 
-    def download(self, download_func: Callable, request: Request, spider: Spider):
-        @defer.inlineCallbacks
-        def process_request(request: Request):
+    def download(
+        self, download_func: Callable, request: Request, spider: Spider
+    ) -> Deferred:
+        @inlineCallbacks
+        def process_request(request: Request) -> Generator[Deferred, Any, Any]:
             for method in self.methods["process_request"]:
                 method = cast(Callable, method)
                 response = yield deferred_from_coro(
@@ -50,8 +53,10 @@ class DownloaderMiddlewareManager(MiddlewareManager):
                     return response
             return (yield download_func(request=request, spider=spider))
 
-        @defer.inlineCallbacks
-        def process_response(response: Union[Response, Request]):
+        @inlineCallbacks
+        def process_response(
+            response: Union[Response, Request]
+        ) -> Generator[Deferred, Any, Union[Response, Request]]:
             if response is None:
                 raise TypeError("Received None in process_response")
             elif isinstance(response, Request):
@@ -71,8 +76,10 @@ class DownloaderMiddlewareManager(MiddlewareManager):
                     return response
             return response
 
-        @defer.inlineCallbacks
-        def process_exception(failure: Failure):
+        @inlineCallbacks
+        def process_exception(
+            failure: Failure,
+        ) -> Generator[Deferred, Any, Union[Failure, Response, Request]]:
             exception = failure.value
             for method in self.methods["process_exception"]:
                 method = cast(Callable, method)
diff --git a/scrapy/core/downloader/tls.py b/scrapy/core/downloader/tls.py
index 025575fe1..33cea7263 100644
--- a/scrapy/core/downloader/tls.py
+++ b/scrapy/core/downloader/tls.py
@@ -1,4 +1,5 @@
 import logging
+from typing import Any, Dict
 
 from OpenSSL import SSL
 from service_identity.exceptions import CertificateError
@@ -20,7 +21,7 @@ METHOD_TLSv11 = "TLSv1.1"
 METHOD_TLSv12 = "TLSv1.2"
 
 
-openssl_methods = {
+openssl_methods: Dict[str, int] = {
     METHOD_TLS: SSL.SSLv23_METHOD,  # protocol negotiation (recommended)
     METHOD_TLSv10: SSL.TLSv1_METHOD,  # TLS 1.0 only
     METHOD_TLSv11: SSL.TLSv1_1_METHOD,  # TLS 1.1 only
@@ -39,11 +40,13 @@ class ScrapyClientTLSOptions(ClientTLSOptions):
     logging warnings. Also, HTTPS connection parameters logging is added.
     """
 
-    def __init__(self, hostname, ctx, verbose_logging=False):
+    def __init__(self, hostname: str, ctx: SSL.Context, verbose_logging: bool = False):
         super().__init__(hostname, ctx)
-        self.verbose_logging = verbose_logging
+        self.verbose_logging: bool = verbose_logging
 
-    def _identityVerifyingInfoCallback(self, connection, where, ret):
+    def _identityVerifyingInfoCallback(
+        self, connection: SSL.Connection, where: int, ret: Any
+    ) -> None:
         if where & SSL.SSL_CB_HANDSHAKE_START:
             connection.set_tlsext_host_name(self._hostnameBytes)
         elif where & SSL.SSL_CB_HANDSHAKE_DONE:
@@ -55,11 +58,12 @@ class ScrapyClientTLSOptions(ClientTLSOptions):
                     connection.get_cipher_name(),
                 )
                 server_cert = connection.get_peer_certificate()
-                logger.debug(
-                    'SSL connection certificate: issuer "%s", subject "%s"',
-                    x509name_to_string(server_cert.get_issuer()),
-                    x509name_to_string(server_cert.get_subject()),
-                )
+                if server_cert:
+                    logger.debug(
+                        'SSL connection certificate: issuer "%s", subject "%s"',
+                        x509name_to_string(server_cert.get_issuer()),
+                        x509name_to_string(server_cert.get_subject()),
+                    )
                 key_info = get_temp_key_info(connection._ssl)
                 if key_info:
                     logger.debug("SSL temp key: %s", key_info)
@@ -82,4 +86,6 @@ class ScrapyClientTLSOptions(ClientTLSOptions):
                 )
 
 
-DEFAULT_CIPHERS = AcceptableCiphers.fromOpenSSLCipherString("DEFAULT")
+DEFAULT_CIPHERS: AcceptableCiphers = AcceptableCiphers.fromOpenSSLCipherString(
+    "DEFAULT"
+)
diff --git a/scrapy/core/downloader/webclient.py b/scrapy/core/downloader/webclient.py
index 3d103652b..bb1f73805 100644
--- a/scrapy/core/downloader/webclient.py
+++ b/scrapy/core/downloader/webclient.py
@@ -1,22 +1,25 @@
 import re
 from time import time
-from urllib.parse import urldefrag, urlparse, urlunparse
+from typing import Optional, Tuple
+from urllib.parse import ParseResult, urldefrag, urlparse, urlunparse
 
 from twisted.internet import defer
 from twisted.internet.protocol import ClientFactory
 from twisted.web.http import HTTPClient
 
+from scrapy import Request
 from scrapy.http import Headers
 from scrapy.responsetypes import responsetypes
 from scrapy.utils.httpobj import urlparse_cached
 from scrapy.utils.python import to_bytes, to_unicode
 
 
-def _parsed_url_args(parsed):
+def _parsed_url_args(parsed: ParseResult) -> Tuple[bytes, bytes, bytes, int, bytes]:
     # Assume parsed is urlparse-d from Request.url,
     # which was passed via safe_url_string and is ascii-only.
-    path = urlunparse(("", "", parsed.path or "/", parsed.params, parsed.query, ""))
-    path = to_bytes(path, encoding="ascii")
+    path_str = urlunparse(("", "", parsed.path or "/", parsed.params, parsed.query, ""))
+    path = to_bytes(path_str, encoding="ascii")
+    assert parsed.hostname is not None
     host = to_bytes(parsed.hostname, encoding="ascii")
     port = parsed.port
     scheme = to_bytes(parsed.scheme, encoding="ascii")
@@ -26,7 +29,7 @@ def _parsed_url_args(parsed):
     return scheme, netloc, host, port, path
 
 
-def _parse(url):
+def _parse(url: str) -> Tuple[bytes, bytes, bytes, int, bytes]:
     """Return tuple of (scheme, netloc, host, port, path),
     all in bytes except for port which is int.
     Assume url is from Request.url, which was passed via safe_url_string
@@ -132,17 +135,19 @@ class ScrapyHTTPClientFactory(ClientFactory):
             self.scheme, _, self.host, self.port, _ = _parse(proxy)
             self.path = self.url
 
-    def __init__(self, request, timeout=180):
-        self._url = urldefrag(request.url)[0]
+    def __init__(self, request: Request, timeout: float = 180):
+        self._url: str = urldefrag(request.url)[0]
         # converting to bytes to comply to Twisted interface
-        self.url = to_bytes(self._url, encoding="ascii")
-        self.method = to_bytes(request.method, encoding="ascii")
-        self.body = request.body or None
-        self.headers = Headers(request.headers)
-        self.response_headers = None
-        self.timeout = request.meta.get("download_timeout") or timeout
-        self.start_time = time()
-        self.deferred = defer.Deferred().addCallback(self._build_response, request)
+        self.url: bytes = to_bytes(self._url, encoding="ascii")
+        self.method: bytes = to_bytes(request.method, encoding="ascii")
+        self.body: Optional[bytes] = request.body or None
+        self.headers: Headers = Headers(request.headers)
+        self.response_headers: Optional[Headers] = None
+        self.timeout: float = request.meta.get("download_timeout") or timeout
+        self.start_time: float = time()
+        self.deferred: defer.Deferred = defer.Deferred().addCallback(
+            self._build_response, request
+        )
 
         # Fixes Twisted 11.1.0+ support as HTTPClientFactory is expected
         # to have _disconnectedDeferred. See Twisted r32329.
@@ -150,7 +155,7 @@ class ScrapyHTTPClientFactory(ClientFactory):
         # needed to add the callback _waitForDisconnect.
         # Specifically this avoids the AttributeError exception when
         # clientConnectionFailed method is called.
-        self._disconnectedDeferred = defer.Deferred()
+        self._disconnectedDeferred: defer.Deferred = defer.Deferred()
 
         self._set_connection_attributes(request)
 
@@ -166,8 +171,8 @@ class ScrapyHTTPClientFactory(ClientFactory):
         elif self.method == b"POST":
             self.headers["Content-Length"] = 0
 
-    def __repr__(self):
-        return f"<{self.__class__.__name__}: {self.url}>"
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__}: {self._url}>"
 
     def _cancelTimeout(self, result, timeoutCall):
         if timeoutCall.active():
diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py
index c6738b531..3e5a281b2 100644
--- a/scrapy/core/engine.py
+++ b/scrapy/core/engine.py
@@ -8,13 +8,16 @@ import logging
 import warnings
 from time import time
 from typing import (
+    TYPE_CHECKING,
     Any,
     Callable,
     Generator,
     Iterable,
     Iterator,
+    List,
     Optional,
     Set,
+    Type,
     Union,
     cast,
 )
@@ -28,30 +31,36 @@ from scrapy.core.downloader import Downloader
 from scrapy.core.scraper import Scraper
 from scrapy.exceptions import CloseSpider, DontCloseSpider, ScrapyDeprecationWarning
 from scrapy.http import Request, Response
-from scrapy.settings import BaseSettings
+from scrapy.logformatter import LogFormatter
+from scrapy.settings import BaseSettings, Settings
+from scrapy.signalmanager import SignalManager
 from scrapy.spiders import Spider
 from scrapy.utils.log import failure_to_exc_info, logformatter_adapter
 from scrapy.utils.misc import create_instance, load_object
 from scrapy.utils.reactor import CallLaterOnce
 
+if TYPE_CHECKING:
+    from scrapy.core.scheduler import BaseScheduler
+    from scrapy.crawler import Crawler
+
 logger = logging.getLogger(__name__)
 
 
 class Slot:
     def __init__(
         self,
-        start_requests: Iterable,
+        start_requests: Iterable[Request],
         close_if_idle: bool,
         nextcall: CallLaterOnce,
-        scheduler,
+        scheduler: "BaseScheduler",
     ) -> None:
         self.closing: Optional[Deferred] = None
         self.inprogress: Set[Request] = set()
-        self.start_requests: Optional[Iterator] = iter(start_requests)
-        self.close_if_idle = close_if_idle
-        self.nextcall = nextcall
-        self.scheduler = scheduler
-        self.heartbeat = LoopingCall(nextcall.schedule)
+        self.start_requests: Optional[Iterator[Request]] = iter(start_requests)
+        self.close_if_idle: bool = close_if_idle
+        self.nextcall: CallLaterOnce = nextcall
+        self.scheduler: "BaseScheduler" = scheduler
+        self.heartbeat: LoopingCall = LoopingCall(nextcall.schedule)
 
     def add_request(self, request: Request) -> None:
         self.inprogress.add(request)
@@ -75,25 +84,28 @@ class Slot:
 
 
 class ExecutionEngine:
-    def __init__(self, crawler, spider_closed_callback: Callable) -> None:
-        self.crawler = crawler
-        self.settings = crawler.settings
-        self.signals = crawler.signals
-        self.logformatter = crawler.logformatter
+    def __init__(self, crawler: "Crawler", spider_closed_callback: Callable) -> None:
+        self.crawler: "Crawler" = crawler
+        self.settings: Settings = crawler.settings
+        self.signals: SignalManager = crawler.signals
+        self.logformatter: LogFormatter = crawler.logformatter
         self.slot: Optional[Slot] = None
         self.spider: Optional[Spider] = None
-        self.running = False
-        self.paused = False
-        self.scheduler_cls = self._get_scheduler_class(crawler.settings)
-        downloader_cls = load_object(self.settings["DOWNLOADER"])
+        self.running: bool = False
+        self.paused: bool = False
+        self.scheduler_cls: Type["BaseScheduler"] = self._get_scheduler_class(
+            crawler.settings
+        )
+        downloader_cls: Type[Downloader] = load_object(self.settings["DOWNLOADER"])
         self.downloader: Downloader = downloader_cls(crawler)
         self.scraper = Scraper(crawler)
-        self._spider_closed_callback = spider_closed_callback
+        self._spider_closed_callback: Callable = spider_closed_callback
+        self.start_time: Optional[float] = None
 
-    def _get_scheduler_class(self, settings: BaseSettings) -> type:
+    def _get_scheduler_class(self, settings: BaseSettings) -> Type["BaseScheduler"]:
         from scrapy.core.scheduler import BaseScheduler
 
-        scheduler_cls = load_object(settings["SCHEDULER"])
+        scheduler_cls: Type = load_object(settings["SCHEDULER"])
         if not issubclass(scheduler_cls, BaseScheduler):
             raise TypeError(
                 f"The provided scheduler class ({settings['SCHEDULER']})"
@@ -115,7 +127,7 @@ class ExecutionEngine:
         """Gracefully stop the execution engine"""
 
         @inlineCallbacks
-        def _finish_stopping_engine(_) -> Generator[Deferred, Any, None]:
+        def _finish_stopping_engine(_: Any) -> Generator[Deferred, Any, None]:
             yield self.signals.send_catch_log_deferred(signal=signals.engine_stopped)
             self._closewait.callback(None)
 
@@ -141,7 +153,8 @@ class ExecutionEngine:
             return self.close_spider(
                 self.spider, reason="shutdown"
             )  # will also close downloader
-        return succeed(self.downloader.close())
+        self.downloader.close()
+        return succeed(None)
 
     def pause(self) -> None:
         self.paused = True
@@ -209,7 +222,7 @@ class ExecutionEngine:
                 extra={"spider": self.spider},
             )
         )
-        d.addBoth(lambda _: cast(Slot, self.slot).remove_request(request))
+        d.addBoth(lambda _: cast(Slot, self.slot).remove_request(request))  # type: ignore[arg-type]
         d.addErrback(
             lambda f: logger.info(
                 "Error while removing request from slot",
@@ -339,6 +352,7 @@ class ExecutionEngine:
             if isinstance(result, Response):
                 if result.request is None:
                     result.request = request
+                assert spider is not None
                 logkws = self.logformatter.crawled(result.request, result, spider)
                 if logkws is not None:
                     logger.log(*logformatter_adapter(logkws), extra={"spider": spider})
@@ -350,10 +364,12 @@ class ExecutionEngine:
                 )
             return result
 
-        def _on_complete(_):
+        def _on_complete(_: Any) -> Any:
+            assert self.slot is not None
             self.slot.nextcall.schedule()
             return _
 
+        assert spider is not None
         dwld = self.downloader.fetch(request, spider)
         dwld.addCallbacks(_on_success)
         dwld.addBoth(_on_complete)
@@ -362,7 +378,7 @@ class ExecutionEngine:
     @inlineCallbacks
     def open_spider(
         self, spider: Spider, start_requests: Iterable = (), close_if_idle: bool = True
-    ):
+    ) -> Generator[Deferred, Any, None]:
         if self.slot is not None:
             raise RuntimeError(f"No free spider slot when opening {spider.name!r}")
         logger.info("Spider opened", extra={"spider": spider})
@@ -471,7 +487,7 @@ class ExecutionEngine:
         return dfd
 
     @property
-    def open_spiders(self) -> list:
+    def open_spiders(self) -> List[Spider]:
         warnings.warn(
             "ExecutionEngine.open_spiders is deprecated, please use ExecutionEngine.spider instead",
             category=ScrapyDeprecationWarning,
diff --git a/scrapy/core/scheduler.py b/scrapy/core/scheduler.py
index 3c46e3a5f..3fb0bbaff 100644
--- a/scrapy/core/scheduler.py
+++ b/scrapy/core/scheduler.py
@@ -2,13 +2,15 @@ import json
 import logging
 from abc import abstractmethod
 from pathlib import Path
-from typing import Optional, Type, TypeVar
+from typing import Any, Optional, Type, TypeVar, cast
 
 from twisted.internet.defer import Deferred
 
 from scrapy.crawler import Crawler
+from scrapy.dupefilters import BaseDupeFilter
 from scrapy.http.request import Request
 from scrapy.spiders import Spider
+from scrapy.statscollectors import StatsCollector
 from scrapy.utils.job import job_dir
 from scrapy.utils.misc import create_instance, load_object
 
@@ -20,10 +22,10 @@ class BaseSchedulerMeta(type):
     Metaclass to check scheduler classes against the necessary interface
     """
 
-    def __instancecheck__(cls, instance):
+    def __instancecheck__(cls, instance: Any) -> bool:
         return cls.__subclasscheck__(type(instance))
 
-    def __subclasscheck__(cls, subclass):
+    def __subclasscheck__(cls, subclass: type) -> bool:
         return (
             hasattr(subclass, "has_pending_requests")
             and callable(subclass.has_pending_requests)
@@ -168,26 +170,26 @@ class Scheduler(BaseScheduler):
 
     def __init__(
         self,
-        dupefilter,
+        dupefilter: BaseDupeFilter,
         jobdir: Optional[str] = None,
         dqclass=None,
         mqclass=None,
         logunser: bool = False,
-        stats=None,
+        stats: Optional[StatsCollector] = None,
         pqclass=None,
         crawler: Optional[Crawler] = None,
     ):
-        self.df = dupefilter
-        self.dqdir = self._dqdir(jobdir)
+        self.df: BaseDupeFilter = dupefilter
+        self.dqdir: Optional[str] = self._dqdir(jobdir)
         self.pqclass = pqclass
         self.dqclass = dqclass
         self.mqclass = mqclass
-        self.logunser = logunser
-        self.stats = stats
-        self.crawler = crawler
+        self.logunser: bool = logunser
+        self.stats: Optional[StatsCollector] = stats
+        self.crawler: Optional[Crawler] = crawler
 
     @classmethod
-    def from_crawler(cls: Type[SchedulerTV], crawler) -> SchedulerTV:
+    def from_crawler(cls: Type[SchedulerTV], crawler: Crawler) -> SchedulerTV:
         """
         Factory method, initializes the scheduler with arguments taken from the crawl settings
         """
@@ -242,6 +244,7 @@ class Scheduler(BaseScheduler):
             self.df.log(request, self.spider)
             return False
         dqok = self._dqpush(request)
+        assert self.stats is not None
         if dqok:
             self.stats.inc_value("scheduler/enqueued/disk", spider=self.spider)
         else:
@@ -259,7 +262,8 @@ class Scheduler(BaseScheduler):
         Increment the appropriate stats, such as: ``scheduler/dequeued``,
         ``scheduler/dequeued/disk``, ``scheduler/dequeued/memory``.
         """
-        request = self.mqs.pop()
+        request: Optional[Request] = self.mqs.pop()
+        assert self.stats is not None
         if request is not None:
             self.stats.inc_value("scheduler/dequeued/memory", spider=self.spider)
         else:
@@ -295,6 +299,7 @@ class Scheduler(BaseScheduler):
                     extra={"spider": self.spider},
                 )
                 self.logunser = False
+            assert self.stats is not None
             self.stats.inc_value("scheduler/unserializable", spider=self.spider)
             return False
         else:
@@ -351,7 +356,7 @@ class Scheduler(BaseScheduler):
         if not path.exists():
             return []
         with path.open(encoding="utf-8") as f:
-            return json.load(f)
+            return cast(list, json.load(f))
 
     def _write_dqs_state(self, dqdir: str, state: list) -> None:
         with Path(dqdir, "active.json").open("w", encoding="utf-8") as f:
diff --git a/scrapy/core/scraper.py b/scrapy/core/scraper.py
index 8468b8419..a85f6a661 100644
--- a/scrapy/core/scraper.py
+++ b/scrapy/core/scraper.py
@@ -15,6 +15,7 @@ from typing import (
     Optional,
     Set,
     Tuple,
+    Type,
     Union,
 )
 
@@ -26,6 +27,9 @@ from scrapy import Spider, signals
 from scrapy.core.spidermw import SpiderMiddlewareManager
 from scrapy.exceptions import CloseSpider, DropItem, IgnoreRequest
 from scrapy.http import Request, Response
+from scrapy.logformatter import LogFormatter
+from scrapy.pipelines import ItemPipelineManager
+from scrapy.signalmanager import SignalManager
 from scrapy.utils.defer import (
     aiter_errback,
     defer_fail,
@@ -96,16 +100,20 @@ class Slot:
 class Scraper:
     def __init__(self, crawler: Crawler) -> None:
         self.slot: Optional[Slot] = None
-        self.spidermw = SpiderMiddlewareManager.from_crawler(crawler)
-        itemproc_cls = load_object(crawler.settings["ITEM_PROCESSOR"])
-        self.itemproc = itemproc_cls.from_crawler(crawler)
-        self.concurrent_items = crawler.settings.getint("CONCURRENT_ITEMS")
-        self.crawler = crawler
-        self.signals = crawler.signals
-        self.logformatter = crawler.logformatter
+        self.spidermw: SpiderMiddlewareManager = SpiderMiddlewareManager.from_crawler(
+            crawler
+        )
+        itemproc_cls: Type[ItemPipelineManager] = load_object(
+            crawler.settings["ITEM_PROCESSOR"]
+        )
+        self.itemproc: ItemPipelineManager = itemproc_cls.from_crawler(crawler)
+        self.concurrent_items: int = crawler.settings.getint("CONCURRENT_ITEMS")
+        self.crawler: Crawler = crawler
+        self.signals: SignalManager = crawler.signals
+        self.logformatter: LogFormatter = crawler.logformatter
 
     @inlineCallbacks
-    def open_spider(self, spider: Spider):
+    def open_spider(self, spider: Spider) -> Generator[Deferred, Any, None]:
         """Open the given spider for scraping and allocate resources for it"""
         self.slot = Slot(self.crawler.settings.getint("SCRAPER_SLOT_MAX_ACTIVE_SIZE"))
         yield self.itemproc.open_spider(spider)
@@ -135,7 +143,8 @@ class Scraper:
             raise RuntimeError("Scraper slot not assigned")
         dfd = self.slot.add_response_request(result, request)
 
-        def finish_scraping(_):
+        def finish_scraping(_: Any) -> Any:
+            assert self.slot is not None
             self.slot.finish_response(result, request)
             self._check_if_closing(spider)
             self._scrape_next(spider)
@@ -205,9 +214,9 @@ class Scraper:
         else:  # result is a Failure
             # TODO: properly type adding this attribute to a Failure
             result.request = request  # type: ignore[attr-defined]
-            warn_on_generator_with_return_value(spider, request.errback)
             dfd = defer_fail(result)
             if request.errback:
+                warn_on_generator_with_return_value(spider, request.errback)
                 dfd.addErrback(request.errback)
         return dfd.addCallback(iterate_spider_output)
 
@@ -338,7 +347,7 @@ class Scraper:
 
     def _itemproc_finished(
         self, output: Any, item: Any, response: Response, spider: Spider
-    ) -> None:
+    ) -> Deferred:
         """ItemProcessor finished for the given ``item`` and returned ``output``"""
         assert self.slot is not None  # typing
         self.slot.itemproc_size -= 1
diff --git a/scrapy/core/spidermw.py b/scrapy/core/spidermw.py
index c21985b18..dcf1a6dbc 100644
--- a/scrapy/core/spidermw.py
+++ b/scrapy/core/spidermw.py
@@ -13,6 +13,8 @@ from typing import (
     Callable,
     Generator,
     Iterable,
+    List,
+    Optional,
     Tuple,
     Union,
     cast,
@@ -25,6 +27,7 @@ from scrapy import Request, Spider
 from scrapy.exceptions import _InvalidOutput
 from scrapy.http import Response
 from scrapy.middleware import MiddlewareManager
+from scrapy.settings import BaseSettings
 from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
 from scrapy.utils.conf import build_component_list
 from scrapy.utils.defer import (
@@ -41,22 +44,22 @@ logger = logging.getLogger(__name__)
 ScrapeFunc = Callable[[Union[Response, Failure], Request, Spider], Any]
 
 
-def _isiterable(o) -> bool:
+def _isiterable(o: Any) -> bool:
     return isinstance(o, (Iterable, AsyncIterable))
 
 
 class SpiderMiddlewareManager(MiddlewareManager):
     component_name = "spider middleware"
 
-    def __init__(self, *middlewares):
+    def __init__(self, *middlewares: Any):
         super().__init__(*middlewares)
         self.downgrade_warning_done = False
 
     @classmethod
-    def _get_mwlist_from_settings(cls, settings):
+    def _get_mwlist_from_settings(cls, settings: BaseSettings) -> List[Any]:
         return build_component_list(settings.getwithbase("SPIDER_MIDDLEWARES"))
 
-    def _add_middleware(self, mw):
+    def _add_middleware(self, mw: Any) -> None:
         super()._add_middleware(mw)
         if hasattr(mw, "process_spider_input"):
             self.methods["process_spider_input"].append(mw.process_spider_input)
@@ -98,7 +101,7 @@ class SpiderMiddlewareManager(MiddlewareManager):
         exception_processor_index: int,
         recover_to: Union[MutableChain, MutableAsyncChain],
     ) -> Union[Generator, AsyncGenerator]:
-        def process_sync(iterable: Iterable):
+        def process_sync(iterable: Iterable) -> Generator:
             try:
                 for r in iterable:
                     yield r
@@ -110,7 +113,7 @@ class SpiderMiddlewareManager(MiddlewareManager):
                     raise
                 recover_to.extend(exception_result)
 
-        async def process_async(iterable: AsyncIterable):
+        async def process_async(iterable: AsyncIterable) -> AsyncGenerator:
             try:
                 async for r in iterable:
                     yield r
@@ -280,7 +283,7 @@ class SpiderMiddlewareManager(MiddlewareManager):
         if isinstance(recovered, AsyncIterable):
             recovered_collected = await collect_asyncgen(recovered)
             recovered = MutableChain(recovered_collected)
-        return MutableChain(result, recovered)  # type: ignore[arg-type]
+        return MutableChain(result, recovered)
 
     def scrape_response(
         self,
@@ -306,7 +309,9 @@ class SpiderMiddlewareManager(MiddlewareManager):
         )
         return dfd
 
-    def process_start_requests(self, start_requests, spider: Spider) -> Deferred:
+    def process_start_requests(
+        self, start_requests: Iterable[Request], spider: Spider
+    ) -> Deferred:
         return self._process_chain("process_start_requests", start_requests, spider)
 
     # This method is only needed until _async compatibility methods are removed.
@@ -314,9 +319,9 @@ class SpiderMiddlewareManager(MiddlewareManager):
     def _get_async_method_pair(
         mw: Any, methodname: str
     ) -> Union[None, Callable, Tuple[Callable, Callable]]:
-        normal_method = getattr(mw, methodname, None)
+        normal_method: Optional[Callable] = getattr(mw, methodname, None)
         methodname_async = methodname + "_async"
-        async_method = getattr(mw, methodname_async, None)
+        async_method: Optional[Callable] = getattr(mw, methodname_async, None)
         if not async_method:
             return normal_method
         if not normal_method:
diff --git a/scrapy/crawler.py b/scrapy/crawler.py
index 397817d6f..256f6e2c5 100644
--- a/scrapy/crawler.py
+++ b/scrapy/crawler.py
@@ -4,7 +4,7 @@ import logging
 import pprint
 import signal
 import warnings
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Optional, Type, Union
 
 from twisted.internet import defer
 from zope.interface.exceptions import DoesNotImplement
@@ -22,8 +22,10 @@ from scrapy.core.engine import ExecutionEngine
 from scrapy.exceptions import ScrapyDeprecationWarning
 from scrapy.extension import ExtensionManager
 from scrapy.interfaces import ISpiderLoader
+from scrapy.logformatter import LogFormatter
 from scrapy.settings import Settings, overridden_settings
 from scrapy.signalmanager import SignalManager
+from scrapy.statscollectors import StatsCollector
 from scrapy.utils.log import (
     LogCounterHandler,
     configure_logging,
@@ -49,20 +51,25 @@ logger = logging.getLogger(__name__)
 
 
 class Crawler:
-    def __init__(self, spidercls, settings=None, init_reactor: bool = False):
+    def __init__(
+        self,
+        spidercls: Type[Spider],
+        settings: Union[None, dict, Settings] = None,
+        init_reactor: bool = False,
+    ):
         if isinstance(spidercls, Spider):
             raise ValueError("The spidercls argument must be a class, not an object")
 
         if isinstance(settings, dict) or settings is None:
             settings = Settings(settings)
 
-        self.spidercls = spidercls
-        self.settings = settings.copy()
+        self.spidercls: Type[Spider] = spidercls
+        self.settings: Settings = settings.copy()
         self.spidercls.update_settings(self.settings)
 
-        self.signals = SignalManager(self)
+        self.signals: SignalManager = SignalManager(self)
 
-        self.stats = load_object(self.settings["STATS_CLASS"])(self)
+        self.stats: StatsCollector = load_object(self.settings["STATS_CLASS"])(self)
 
         handler = LogCounterHandler(self, level=self.settings.get("LOG_LEVEL"))
         logging.root.addHandler(handler)
@@ -80,8 +87,8 @@ class Crawler:
         self.__remove_handler = lambda: logging.root.removeHandler(handler)
         self.signals.connect(self.__remove_handler, signals.engine_stopped)
 
-        lf_cls = load_object(self.settings["LOG_FORMATTER"])
-        self.logformatter = lf_cls.from_crawler(self)
+        lf_cls: Type[LogFormatter] = load_object(self.settings["LOG_FORMATTER"])
+        self.logformatter: LogFormatter = lf_cls.from_crawler(self)
 
         self.request_fingerprinter: RequestFingerprinter = create_instance(
             load_object(self.settings["REQUEST_FINGERPRINTER_CLASS"]),
@@ -89,8 +96,8 @@ class Crawler:
             crawler=self,
         )
 
-        reactor_class = self.settings["TWISTED_REACTOR"]
-        event_loop = self.settings["ASYNCIO_EVENT_LOOP"]
+        reactor_class: str = self.settings["TWISTED_REACTOR"]
+        event_loop: str = self.settings["ASYNCIO_EVENT_LOOP"]
         if init_reactor:
             # this needs to be done after the spider settings are merged,
             # but before something imports twisted.internet.reactor
@@ -104,11 +111,11 @@ class Crawler:
             if is_asyncio_reactor_installed() and event_loop:
                 verify_installed_asyncio_event_loop(event_loop)
 
-        self.extensions = ExtensionManager.from_crawler(self)
+        self.extensions: ExtensionManager = ExtensionManager.from_crawler(self)
 
         self.settings.freeze()
-        self.crawling = False
-        self.spider = None
+        self.crawling: bool = False
+        self.spider: Optional[Spider] = None
         self.engine: Optional[ExecutionEngine] = None
 
     @defer.inlineCallbacks
diff --git a/scrapy/exceptions.py b/scrapy/exceptions.py
index 7f202b8b8..fedd02805 100644
--- a/scrapy/exceptions.py
+++ b/scrapy/exceptions.py
@@ -39,7 +39,7 @@ class DontCloseSpider(Exception):
 class CloseSpider(Exception):
     """Raise this from callbacks to request the spider to be closed"""
 
-    def __init__(self, reason="cancelled"):
+    def __init__(self, reason: str = "cancelled"):
         super().__init__()
         self.reason = reason
 
diff --git a/scrapy/extensions/feedexport.py b/scrapy/extensions/feedexport.py
index cd26b5778..bcf0b779a 100644
--- a/scrapy/extensions/feedexport.py
+++ b/scrapy/extensions/feedexport.py
@@ -11,10 +11,11 @@ import warnings
 from datetime import datetime
 from pathlib import Path
 from tempfile import NamedTemporaryFile
-from typing import IO, Any, Callable, Optional, Tuple, Union
+from typing import IO, Any, Callable, List, Optional, Tuple, Union
 from urllib.parse import unquote, urlparse
 
 from twisted.internet import defer, threads
+from twisted.internet.defer import DeferredList
 from w3lib.url import file_uri_to_path
 from zope.interface import Interface, implementer
 
@@ -23,6 +24,8 @@ from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
 from scrapy.extensions.postprocessing import PostProcessingManager
 from scrapy.utils.boto import is_botocore_available
 from scrapy.utils.conf import feed_complete_default_values_from_settings
+from scrapy.utils.defer import maybe_deferred_to_future
+from scrapy.utils.deprecate import create_deprecated_class
 from scrapy.utils.ftp import ftp_store_file
 from scrapy.utils.log import failure_to_exc_info
 from scrapy.utils.misc import create_instance, load_object
@@ -271,7 +274,7 @@ class FTPFeedStorage(BlockingFeedStorage):
         )
 
 
-class _FeedSlot:
+class FeedSlot:
     def __init__(
         self,
         file,
@@ -309,7 +312,15 @@ class _FeedSlot:
             self._exporting = False
 
 
+_FeedSlot = create_deprecated_class(
+    name="_FeedSlot",
+    new_class=FeedSlot,
+)
+
+
 class FeedExporter:
+    _pending_deferreds: List[defer.Deferred] = []
+
     @classmethod
     def from_crawler(cls, crawler):
         exporter = cls(crawler)
@@ -375,21 +386,34 @@ class FeedExporter:
                 )
             )
 
-    def close_spider(self, spider):
-        deferred_list = []
+    async def close_spider(self, spider):
         for slot in self.slots:
-            d = self._close_slot(slot, spider)
-            deferred_list.append(d)
-        return defer.DeferredList(deferred_list) if deferred_list else None
+            self._close_slot(slot, spider)
+
+        # Await all deferreds
+        if self._pending_deferreds:
+            await maybe_deferred_to_future(DeferredList(self._pending_deferreds))
+
+        # Send FEED_EXPORTER_CLOSED signal
+        await maybe_deferred_to_future(
+            self.crawler.signals.send_catch_log_deferred(signals.feed_exporter_closed)
+        )
 
     def _close_slot(self, slot, spider):
+        def get_file(slot_):
+            if isinstance(slot_.file, PostProcessingManager):
+                slot_.file.close()
+                return slot_.file.file
+            return slot_.file
+
         slot.finish_exporting()
         if not slot.itemcount and not slot.store_empty:
             # We need to call slot.storage.store nonetheless to get the file
             # properly closed.
-            return defer.maybeDeferred(slot.storage.store, slot.file)
+            return defer.maybeDeferred(slot.storage.store, get_file(slot))
+
         logmsg = f"{slot.format} feed ({slot.itemcount} items) in: {slot.uri}"
-        d = defer.maybeDeferred(slot.storage.store, slot.file)
+        d = defer.maybeDeferred(slot.storage.store, get_file(slot))
 
         d.addCallback(
             self._handle_store_success, logmsg, spider, type(slot.storage).__name__
@@ -397,6 +421,14 @@ class FeedExporter:
         d.addErrback(
             self._handle_store_error, logmsg, spider, type(slot.storage).__name__
         )
+        self._pending_deferreds.append(d)
+        d.addCallback(
+            lambda _: self.crawler.signals.send_catch_log_deferred(
+                signals.feed_slot_closed, slot=slot
+            )
+        )
+        d.addBoth(lambda _: self._pending_deferreds.remove(d))
+
         return d
 
     def _handle_store_error(self, f, logmsg, spider, slot_type):
@@ -437,7 +469,7 @@ class FeedExporter:
             indent=feed_options["indent"],
             **feed_options["item_export_kwargs"],
         )
-        slot = _FeedSlot(
+        slot = FeedSlot(
             file=file,
             exporter=exporter,
             storage=storage,
@@ -572,7 +604,7 @@ class FeedExporter:
         self,
         spider: Spider,
         uri_params_function: Optional[Union[str, Callable[[dict, Spider], dict]]],
-        slot: Optional[_FeedSlot] = None,
+        slot: Optional[FeedSlot] = None,
     ) -> dict:
         params = {}
         for k in dir(spider):
diff --git a/scrapy/linkextractors/lxmlhtml.py b/scrapy/linkextractors/lxmlhtml.py
index dd8dcdf7c..23cbd0116 100644
--- a/scrapy/linkextractors/lxmlhtml.py
+++ b/scrapy/linkextractors/lxmlhtml.py
@@ -1,6 +1,7 @@
 """
 Link extractor based on lxml.html
 """
+import logging
 import operator
 from functools import partial
 from urllib.parse import urljoin, urlparse
@@ -23,6 +24,8 @@ from scrapy.utils.python import unique as unique_list
 from scrapy.utils.response import get_base_url
 from scrapy.utils.url import url_has_any_extension, url_is_from_any_domain
 
+logger = logging.getLogger(__name__)
+
 # from lxml/src/lxml/html/__init__.py
 XHTML_NAMESPACE = "http://www.w3.org/1999/xhtml"
 
@@ -88,7 +91,12 @@ class LxmlParserLinkExtractor:
                 url = self.process_attr(attr_val)
                 if url is None:
                     continue
-            url = safe_url_string(url, encoding=response_encoding)
+            try:
+                url = safe_url_string(url, encoding=response_encoding)
+            except ValueError:
+                logger.debug(f"Skipping extraction of link with bad URL {url!r}")
+                continue
+
             # to fix relative links after process_value
             url = urljoin(response_url, url)
             link = Link(
diff --git a/scrapy/logformatter.py b/scrapy/logformatter.py
index 560006c95..7cb379b46 100644
--- a/scrapy/logformatter.py
+++ b/scrapy/logformatter.py
@@ -1,8 +1,11 @@
 import logging
 import os
+from typing import Any, Dict, Optional, Union
 
 from twisted.python.failure import Failure
 
+from scrapy import Request, Spider
+from scrapy.http import Response
 from scrapy.utils.request import referer_str
 
 SCRAPEDMSG = "Scraped from %(src)s" + os.linesep + "%(item)s"
@@ -52,7 +55,7 @@ class LogFormatter:
                     }
     """
 
-    def crawled(self, request, response, spider):
+    def crawled(self, request: Request, response: Response, spider: Spider) -> dict:
         """Logs a message when the crawler finds a webpage."""
         request_flags = f" {str(request.flags)}" if request.flags else ""
         response_flags = f" {str(response.flags)}" if response.flags else ""
@@ -70,8 +73,11 @@ class LogFormatter:
             },
         }
 
-    def scraped(self, item, response, spider):
+    def scraped(
+        self, item: Any, response: Union[Response, Failure], spider: Spider
+    ) -> dict:
         """Logs a message when an item is scraped by a spider."""
+        src: Any
         if isinstance(response, Failure):
             src = response.getErrorMessage()
         else:
@@ -85,7 +91,9 @@ class LogFormatter:
             },
         }
 
-    def dropped(self, item, exception, response, spider):
+    def dropped(
+        self, item: Any, exception: BaseException, response: Response, spider: Spider
+    ) -> dict:
         """Logs a message when an item is dropped while it is passing through the item pipeline."""
         return {
             "level": logging.WARNING,
@@ -96,7 +104,9 @@ class LogFormatter:
             },
         }
 
-    def item_error(self, item, exception, response, spider):
+    def item_error(
+        self, item: Any, exception, response: Response, spider: Spider
+    ) -> dict:
         """Logs a message when an item causes an error while it is passing
         through the item pipeline.
 
@@ -110,7 +120,9 @@ class LogFormatter:
             },
         }
 
-    def spider_error(self, failure, request, response, spider):
+    def spider_error(
+        self, failure: Failure, request: Request, response: Response, spider: Spider
+    ) -> dict:
         """Logs an error message from a spider.
 
         .. versionadded:: 2.0
@@ -124,13 +136,19 @@ class LogFormatter:
             },
         }
 
-    def download_error(self, failure, request, spider, errmsg=None):
+    def download_error(
+        self,
+        failure: Failure,
+        request: Request,
+        spider: Spider,
+        errmsg: Optional[str] = None,
+    ) -> dict:
         """Logs a download error message from a spider (typically coming from
         the engine).
 
         .. versionadded:: 2.0
         """
-        args = {"request": request}
+        args: Dict[str, Any] = {"request": request}
         if errmsg:
             msg = DOWNLOADERRORMSG_LONG
             args["errmsg"] = errmsg
diff --git a/scrapy/mail.py b/scrapy/mail.py
index 43115c53e..c11f3898d 100644
--- a/scrapy/mail.py
+++ b/scrapy/mail.py
@@ -164,6 +164,7 @@ class MailSender:
                 "mailerr": errstr,
             },
         )
+        return failure
 
     def _sendmail(self, to_addrs, msg):
         from twisted.internet import reactor
diff --git a/scrapy/middleware.py b/scrapy/middleware.py
index f82d722fa..03e92b565 100644
--- a/scrapy/middleware.py
+++ b/scrapy/middleware.py
@@ -1,7 +1,7 @@
 import logging
 import pprint
 from collections import defaultdict, deque
-from typing import Any, Callable, Deque, Dict, Iterable, Tuple, Union, cast
+from typing import Any, Callable, Deque, Dict, Iterable, List, Tuple, Union, cast
 
 from twisted.internet.defer import Deferred
 
@@ -30,7 +30,7 @@ class MiddlewareManager:
             self._add_middleware(mw)
 
     @classmethod
-    def _get_mwlist_from_settings(cls, settings: Settings) -> list:
+    def _get_mwlist_from_settings(cls, settings: Settings) -> List[Any]:
         raise NotImplementedError
 
     @classmethod
@@ -67,17 +67,17 @@ class MiddlewareManager:
     def from_crawler(cls, crawler):
         return cls.from_settings(crawler.settings, crawler)
 
-    def _add_middleware(self, mw) -> None:
+    def _add_middleware(self, mw: Any) -> None:
         if hasattr(mw, "open_spider"):
             self.methods["open_spider"].append(mw.open_spider)
         if hasattr(mw, "close_spider"):
             self.methods["close_spider"].appendleft(mw.close_spider)
 
-    def _process_parallel(self, methodname: str, obj, *args) -> Deferred:
+    def _process_parallel(self, methodname: str, obj: Any, *args: Any) -> Deferred:
         methods = cast(Iterable[Callable], self.methods[methodname])
         return process_parallel(methods, obj, *args)
 
-    def _process_chain(self, methodname: str, obj, *args) -> Deferred:
+    def _process_chain(self, methodname: str, obj: Any, *args: Any) -> Deferred:
         methods = cast(Iterable[Callable], self.methods[methodname])
         return process_chain(methods, obj, *args)
 
diff --git a/scrapy/pipelines/__init__.py b/scrapy/pipelines/__init__.py
index df574a0a1..c97d71fb6 100644
--- a/scrapy/pipelines/__init__.py
+++ b/scrapy/pipelines/__init__.py
@@ -3,7 +3,11 @@ Item pipeline
 
 See documentation in docs/item-pipeline.rst
 """
+from typing import Any, List
 
+from twisted.internet.defer import Deferred
+
+from scrapy import Spider
 from scrapy.middleware import MiddlewareManager
 from scrapy.utils.conf import build_component_list
 from scrapy.utils.defer import deferred_f_from_coro_f
@@ -13,15 +17,15 @@ class ItemPipelineManager(MiddlewareManager):
     component_name = "item pipeline"
 
     @classmethod
-    def _get_mwlist_from_settings(cls, settings):
+    def _get_mwlist_from_settings(cls, settings) -> List[Any]:
         return build_component_list(settings.getwithbase("ITEM_PIPELINES"))
 
-    def _add_middleware(self, pipe):
+    def _add_middleware(self, pipe: Any) -> None:
         super()._add_middleware(pipe)
         if hasattr(pipe, "process_item"):
             self.methods["process_item"].append(
                 deferred_f_from_coro_f(pipe.process_item)
             )
 
-    def process_item(self, item, spider):
+    def process_item(self, item: Any, spider: Spider) -> Deferred:
         return self._process_chain("process_item", item, spider)
diff --git a/scrapy/pipelines/files.py b/scrapy/pipelines/files.py
index 6e9f661e5..4b594ccb7 100644
--- a/scrapy/pipelines/files.py
+++ b/scrapy/pipelines/files.py
@@ -3,6 +3,7 @@ Files Pipeline
 
 See documentation in topics/media-pipeline.rst
 """
+import base64
 import functools
 import hashlib
 import logging
@@ -228,7 +229,7 @@ class GCSFilesStore:
     def stat_file(self, path, info):
         def _onsuccess(blob):
             if blob:
-                checksum = blob.md5_hash
+                checksum = base64.b64decode(blob.md5_hash).hex()
                 last_modified = time.mktime(blob.updated.timetuple())
                 return {"checksum": checksum, "last_modified": last_modified}
             return {}
diff --git a/scrapy/signalmanager.py b/scrapy/signalmanager.py
index d7e3bce91..f6df191d8 100644
--- a/scrapy/signalmanager.py
+++ b/scrapy/signalmanager.py
@@ -1,13 +1,16 @@
+from typing import Any, List, Tuple
+
 from pydispatch import dispatcher
+from twisted.internet.defer import Deferred
 
 from scrapy.utils import signal as _signal
 
 
 class SignalManager:
-    def __init__(self, sender=dispatcher.Anonymous):
-        self.sender = sender
+    def __init__(self, sender: Any = dispatcher.Anonymous):
+        self.sender: Any = sender
 
-    def connect(self, receiver, signal, **kwargs):
+    def connect(self, receiver: Any, signal: Any, **kwargs: Any) -> None:
         """
         Connect a receiver function to a signal.
 
@@ -22,18 +25,18 @@ class SignalManager:
         :type signal: object
         """
         kwargs.setdefault("sender", self.sender)
-        return dispatcher.connect(receiver, signal, **kwargs)
+        dispatcher.connect(receiver, signal, **kwargs)
 
-    def disconnect(self, receiver, signal, **kwargs):
+    def disconnect(self, receiver: Any, signal: Any, **kwargs: Any) -> None:
         """
         Disconnect a receiver function from a signal. This has the
         opposite effect of the :meth:`connect` method, and the arguments
         are the same.
         """
         kwargs.setdefault("sender", self.sender)
-        return dispatcher.disconnect(receiver, signal, **kwargs)
+        dispatcher.disconnect(receiver, signal, **kwargs)
 
-    def send_catch_log(self, signal, **kwargs):
+    def send_catch_log(self, signal: Any, **kwargs: Any) -> List[Tuple[Any, Any]]:
         """
         Send a signal, catch exceptions and log them.
 
@@ -43,7 +46,7 @@ class SignalManager:
         kwargs.setdefault("sender", self.sender)
         return _signal.send_catch_log(signal, **kwargs)
 
-    def send_catch_log_deferred(self, signal, **kwargs):
+    def send_catch_log_deferred(self, signal: Any, **kwargs: Any) -> Deferred:
         """
         Like :meth:`send_catch_log` but supports returning
         :class:`~twisted.internet.defer.Deferred` objects from signal handlers.
@@ -57,7 +60,7 @@ class SignalManager:
         kwargs.setdefault("sender", self.sender)
         return _signal.send_catch_log_deferred(signal, **kwargs)
 
-    def disconnect_all(self, signal, **kwargs):
+    def disconnect_all(self, signal: Any, **kwargs: Any) -> None:
         """
         Disconnect all receivers from the given signal.
 
@@ -65,4 +68,4 @@ class SignalManager:
         :type signal: object
         """
         kwargs.setdefault("sender", self.sender)
-        return _signal.disconnect_all(signal, **kwargs)
+        _signal.disconnect_all(signal, **kwargs)
diff --git a/scrapy/signals.py b/scrapy/signals.py
index 8cf2a4d93..0090f1c8b 100644
--- a/scrapy/signals.py
+++ b/scrapy/signals.py
@@ -22,6 +22,8 @@ bytes_received = object()
 item_scraped = object()
 item_dropped = object()
 item_error = object()
+feed_slot_closed = object()
+feed_exporter_closed = object()
 
 # for backward compatibility
 stats_spider_opened = spider_opened
diff --git a/scrapy/statscollectors.py b/scrapy/statscollectors.py
index dd3c32737..15193aac5 100644
--- a/scrapy/statscollectors.py
+++ b/scrapy/statscollectors.py
@@ -3,44 +3,57 @@ Scrapy extension for collecting scraping stats
 """
 import logging
 import pprint
+from typing import TYPE_CHECKING, Any, Dict, Optional
+
+from scrapy import Spider
+
+if TYPE_CHECKING:
+    from scrapy.crawler import Crawler
 
 logger = logging.getLogger(__name__)
 
 
+StatsT = Dict[str, Any]
+
+
 class StatsCollector:
-    def __init__(self, crawler):
-        self._dump = crawler.settings.getbool("STATS_DUMP")
-        self._stats = {}
+    def __init__(self, crawler: "Crawler"):
+        self._dump: bool = crawler.settings.getbool("STATS_DUMP")
+        self._stats: StatsT = {}
 
-    def get_value(self, key, default=None, spider=None):
+    def get_value(
+        self, key: str, default: Any = None, spider: Optional[Spider] = None
+    ) -> Any:
         return self._stats.get(key, default)
 
-    def get_stats(self, spider=None):
+    def get_stats(self, spider: Optional[Spider] = None) -> StatsT:
         return self._stats
 
-    def set_value(self, key, value, spider=None):
+    def set_value(self, key: str, value: Any, spider: Optional[Spider] = None) -> None:
         self._stats[key] = value
 
-    def set_stats(self, stats, spider=None):
+    def set_stats(self, stats: StatsT, spider: Optional[Spider] = None) -> None:
         self._stats = stats
 
-    def inc_value(self, key, count=1, start=0, spider=None):
+    def inc_value(
+        self, key: str, count: int = 1, start: int = 0, spider: Optional[Spider] = None
+    ) -> None:
         d = self._stats
         d[key] = d.setdefault(key, start) + count
 
-    def max_value(self, key, value, spider=None):
+    def max_value(self, key: str, value: Any, spider: Optional[Spider] = None) -> None:
         self._stats[key] = max(self._stats.setdefault(key, value), value)
 
-    def min_value(self, key, value, spider=None):
+    def min_value(self, key: str, value: Any, spider: Optional[Spider] = None) -> None:
         self._stats[key] = min(self._stats.setdefault(key, value), value)
 
-    def clear_stats(self, spider=None):
+    def clear_stats(self, spider: Optional[Spider] = None) -> None:
         self._stats.clear()
 
-    def open_spider(self, spider):
+    def open_spider(self, spider: Spider) -> None:
         pass
 
-    def close_spider(self, spider, reason):
+    def close_spider(self, spider: Spider, reason: str) -> None:
         if self._dump:
             logger.info(
                 "Dumping Scrapy stats:\n" + pprint.pformat(self._stats),
@@ -48,34 +61,38 @@ class StatsCollector:
             )
         self._persist_stats(self._stats, spider)
 
-    def _persist_stats(self, stats, spider):
+    def _persist_stats(self, stats: StatsT, spider: Spider) -> None:
         pass
 
 
 class MemoryStatsCollector(StatsCollector):
-    def __init__(self, crawler):
+    def __init__(self, crawler: "Crawler"):
         super().__init__(crawler)
-        self.spider_stats = {}
+        self.spider_stats: Dict[str, StatsT] = {}
 
-    def _persist_stats(self, stats, spider):
+    def _persist_stats(self, stats: StatsT, spider: Spider) -> None:
         self.spider_stats[spider.name] = stats
 
 
 class DummyStatsCollector(StatsCollector):
-    def get_value(self, key, default=None, spider=None):
+    def get_value(
+        self, key: str, default: Any = None, spider: Optional[Spider] = None
+    ) -> Any:
         return default
 
-    def set_value(self, key, value, spider=None):
+    def set_value(self, key: str, value: Any, spider: Optional[Spider] = None) -> None:
         pass
 
-    def set_stats(self, stats, spider=None):
+    def set_stats(self, stats: StatsT, spider: Optional[Spider] = None) -> None:
         pass
 
-    def inc_value(self, key, count=1, start=0, spider=None):
+    def inc_value(
+        self, key: str, count: int = 1, start: int = 0, spider: Optional[Spider] = None
+    ) -> None:
         pass
 
-    def max_value(self, key, value, spider=None):
+    def max_value(self, key: str, value: Any, spider: Optional[Spider] = None) -> None:
         pass
 
-    def min_value(self, key, value, spider=None):
+    def min_value(self, key: str, value: Any, spider: Optional[Spider] = None) -> None:
         pass
diff --git a/scrapy/utils/curl.py b/scrapy/utils/curl.py
index a2243ae2e..f5dbbd64e 100644
--- a/scrapy/utils/curl.py
+++ b/scrapy/utils/curl.py
@@ -7,6 +7,14 @@ from urllib.parse import urlparse
 from w3lib.http import basic_auth_header
 
 
+class DataAction(argparse.Action):
+    def __call__(self, parser, namespace, values, option_string=None):
+        value = str(values)
+        if value.startswith("$"):
+            value = value[1:]
+        setattr(namespace, self.dest, value)
+
+
 class CurlParser(argparse.ArgumentParser):
     def error(self, message):
         error_msg = f"There was an error parsing the curl command: {message}"
@@ -17,7 +25,7 @@ curl_parser = CurlParser()
 curl_parser.add_argument("url")
 curl_parser.add_argument("-H", "--header", dest="headers", action="append")
 curl_parser.add_argument("-X", "--request", dest="method")
-curl_parser.add_argument("-d", "--data", "--data-raw", dest="data")
+curl_parser.add_argument("-d", "--data", "--data-raw", dest="data", action=DataAction)
 curl_parser.add_argument("-u", "--user", dest="auth")
 
 
diff --git a/scrapy/utils/log.py b/scrapy/utils/log.py
index 6ae27dc29..2ce4725f4 100644
--- a/scrapy/utils/log.py
+++ b/scrapy/utils/log.py
@@ -2,6 +2,7 @@ import logging
 import sys
 import warnings
 from logging.config import dictConfig
+from typing import Tuple
 
 from twisted.python import log as twisted_log
 from twisted.python.failure import Failure
@@ -14,7 +15,7 @@ from scrapy.utils.versions import scrapy_components_versions
 logger = logging.getLogger(__name__)
 
 
-def failure_to_exc_info(failure):
+def failure_to_exc_info(failure: Failure):
     """Extract exc_info from Failure instances"""
     if isinstance(failure, Failure):
         return (failure.type, failure.value, failure.getTracebackObject())
@@ -206,7 +207,7 @@ class LogCounterHandler(logging.Handler):
         self.crawler.stats.inc_value(sname)
 
 
-def logformatter_adapter(logkws):
+def logformatter_adapter(logkws: dict) -> Tuple[int, str, dict]:
     """
     Helper that takes the dictionary output from the methods in LogFormatter
     and adapts it into a tuple of positional arguments for logger.log calls,
diff --git a/scrapy/utils/misc.py b/scrapy/utils/misc.py
index f9f9c0d5b..d861c9ab6 100644
--- a/scrapy/utils/misc.py
+++ b/scrapy/utils/misc.py
@@ -10,6 +10,7 @@ from contextlib import contextmanager
 from functools import partial
 from importlib import import_module
 from pkgutil import iter_modules
+from typing import TYPE_CHECKING, Any, Callable, Union
 
 from w3lib.html import replace_entities
 
@@ -18,6 +19,10 @@ from scrapy.utils.datatypes import LocalWeakReferencedCache
 from scrapy.utils.deprecate import ScrapyDeprecationWarning
 from scrapy.utils.python import flatten, to_unicode
 
+if TYPE_CHECKING:
+    from scrapy import Spider
+
+
 _ITERABLE_SINGLE_VALUES = dict, Item, str, bytes
 
 
@@ -34,7 +39,7 @@ def arg_to_iter(arg):
     return [arg]
 
 
-def load_object(path):
+def load_object(path: Union[str, Callable]) -> Any:
     """Load an object given its absolute object path, and return it.
 
     The object can be the import path of a class, function, variable or an
@@ -249,7 +254,7 @@ def is_generator_with_return_value(callable):
     return _generator_callbacks_cache[callable]
 
 
-def warn_on_generator_with_return_value(spider, callable):
+def warn_on_generator_with_return_value(spider: "Spider", callable: Callable) -> None:
     """
     Logs a warning if a callable is a generator function and includes
     a 'return' statement with a value different than None
diff --git a/scrapy/utils/python.py b/scrapy/utils/python.py
index fc50e0f12..27816c0df 100644
--- a/scrapy/utils/python.py
+++ b/scrapy/utils/python.py
@@ -8,7 +8,16 @@ import sys
 import weakref
 from functools import partial, wraps
 from itertools import chain
-from typing import Any, AsyncGenerator, AsyncIterable, Iterable, Union
+from typing import (
+    Any,
+    AsyncGenerator,
+    AsyncIterable,
+    Iterable,
+    Mapping,
+    Optional,
+    Union,
+    overload,
+)
 
 from scrapy.utils.asyncgen import as_async_generator
 
@@ -82,7 +91,9 @@ def unique(list_, key=lambda x: x):
     return result
 
 
-def to_unicode(text, encoding=None, errors="strict"):
+def to_unicode(
+    text: Union[str, bytes], encoding: Optional[str] = None, errors: str = "strict"
+) -> str:
     """Return the unicode representation of a bytes object ``text``. If
     ``text`` is already an unicode object, return it as-is."""
     if isinstance(text, str):
@@ -97,7 +108,9 @@ def to_unicode(text, encoding=None, errors="strict"):
     return text.decode(encoding, errors)
 
 
-def to_bytes(text, encoding=None, errors="strict"):
+def to_bytes(
+    text: Union[str, bytes], encoding: Optional[str] = None, errors: str = "strict"
+) -> bytes:
     """Return the binary representation of ``text``. If ``text``
     is already a bytes object, return it as-is."""
     if isinstance(text, bytes):
@@ -160,11 +173,12 @@ def memoizemethod_noargs(method):
     return new_method
 
 
-_BINARYCHARS = {to_bytes(chr(i)) for i in range(32)} - {b"\0", b"\t", b"\n", b"\r"}
-_BINARYCHARS |= {ord(ch) for ch in _BINARYCHARS}
+_BINARYCHARS = {
+    i for i in range(32) if to_bytes(chr(i)) not in {b"\0", b"\t", b"\n", b"\r"}
+}
 
 
-def binary_is_text(data):
+def binary_is_text(data: bytes) -> bool:
     """Returns ``True`` if the given ``data`` argument (a ``bytes`` object)
     does not contain unprintable control characters.
     """
@@ -174,33 +188,33 @@ def binary_is_text(data):
 
 
 def get_func_args(func, stripself=False):
-    """Return the argument name list of a callable"""
-    if inspect.isfunction(func):
-        spec = inspect.getfullargspec(func)
-        func_args = spec.args + spec.kwonlyargs
-    elif inspect.isclass(func):
-        return get_func_args(func.__init__, True)
-    elif inspect.ismethod(func):
-        return get_func_args(func.__func__, True)
-    elif inspect.ismethoddescriptor(func):
-        return []
-    elif isinstance(func, partial):
-        return [
-            x
-            for x in get_func_args(func.func)[len(func.args) :]
-            if not (func.keywords and x in func.keywords)
-        ]
-    elif hasattr(func, "__call__"):
-        if inspect.isroutine(func):
-            return []
-        if getattr(func, "__name__", None) == "__call__":
-            return []
-        return get_func_args(func.__call__, True)
+    """Return the argument name list of a callable object"""
+    if not callable(func):
+        raise TypeError(f"func must be callable, got '{type(func).__name__}'")
+
+    args = []
+    try:
+        sig = inspect.signature(func)
+    except ValueError:
+        return args
+
+    if isinstance(func, partial):
+        partial_args = func.args
+        partial_kw = func.keywords
+
+        for name, param in sig.parameters.items():
+            if param.name in partial_args:
+                continue
+            if partial_kw and param.name in partial_kw:
+                continue
+            args.append(name)
     else:
-        raise TypeError(f"{type(func)} is not callable")
-    if stripself:
-        func_args.pop(0)
-    return func_args
+        for name in sig.parameters.keys():
+            args.append(name)
+
+    if stripself and args and args[0] == "self":
+        args = args[1:]
+    return args
 
 
 def get_spec(func):
@@ -258,6 +272,16 @@ def equal_attributes(obj1, obj2, attributes):
     return True
 
 
+@overload
+def without_none_values(iterable: Mapping) -> dict:
+    ...
+
+
+@overload
+def without_none_values(iterable: Iterable) -> Iterable:
+    ...
+
+
 def without_none_values(iterable):
     """Return a copy of ``iterable`` with all ``None`` entries removed.
 
diff --git a/scrapy/utils/reactor.py b/scrapy/utils/reactor.py
index 7f67d036a..f1b9239e6 100644
--- a/scrapy/utils/reactor.py
+++ b/scrapy/utils/reactor.py
@@ -1,9 +1,11 @@
 import asyncio
 import sys
 from contextlib import suppress
+from typing import Any, Callable, Dict, Optional, Sequence
 from warnings import catch_warnings, filterwarnings, warn
 
 from twisted.internet import asyncioreactor, error
+from twisted.internet.base import DelayedCall
 
 from scrapy.exceptions import ScrapyDeprecationWarning
 from scrapy.utils.misc import load_object
@@ -34,23 +36,23 @@ class CallLaterOnce:
     it hasn't been already scheduled since the last time it ran.
     """
 
-    def __init__(self, func, *a, **kw):
-        self._func = func
-        self._a = a
-        self._kw = kw
-        self._call = None
+    def __init__(self, func: Callable, *a: Any, **kw: Any):
+        self._func: Callable = func
+        self._a: Sequence[Any] = a
+        self._kw: Dict[str, Any] = kw
+        self._call: Optional[DelayedCall] = None
 
-    def schedule(self, delay=0):
+    def schedule(self, delay: float = 0) -> None:
         from twisted.internet import reactor
 
         if self._call is None:
             self._call = reactor.callLater(delay, self)
 
-    def cancel(self):
+    def cancel(self) -> None:
         if self._call:
             self._call.cancel()
 
-    def __call__(self):
+    def __call__(self) -> Any:
         self._call = None
         return self._func(*self._a, **self._kw)
 
diff --git a/scrapy/utils/request.py b/scrapy/utils/request.py
index 409ca2e52..6d8be991d 100644
--- a/scrapy/utils/request.py
+++ b/scrapy/utils/request.py
@@ -327,3 +327,34 @@ def _get_method(obj, name):
         return getattr(obj, name)
     except AttributeError:
         raise ValueError(f"Method {name!r} not found in: {obj}")
+
+
+def request_to_curl(request: Request) -> str:
+    """
+    Converts a :class:`~scrapy.Request` object to a curl command.
+
+    :param :class:`~scrapy.Request`: Request object to be converted
+    :return: string containing the curl command
+    """
+    method = request.method
+
+    data = f"--data-raw '{request.body.decode('utf-8')}'" if request.body else ""
+
+    headers = " ".join(
+        f"-H '{k.decode()}: {v[0].decode()}'" for k, v in request.headers.items()
+    )
+
+    url = request.url
+    cookies = ""
+    if request.cookies:
+        if isinstance(request.cookies, dict):
+            cookie = "; ".join(f"{k}={v}" for k, v in request.cookies.items())
+            cookies = f"--cookie '{cookie}'"
+        elif isinstance(request.cookies, list):
+            cookie = "; ".join(
+                f"{list(c.keys())[0]}={list(c.values())[0]}" for c in request.cookies
+            )
+            cookies = f"--cookie '{cookie}'"
+
+    curl_cmd = f"curl -X {method} {url} {data} {headers} {cookies}".strip()
+    return " ".join(curl_cmd.split())
diff --git a/scrapy/utils/signal.py b/scrapy/utils/signal.py
index b95786d35..9e7ddd827 100644
--- a/scrapy/utils/signal.py
+++ b/scrapy/utils/signal.py
@@ -1,6 +1,8 @@
 """Helper functions for working with signals"""
 import collections.abc
 import logging
+from typing import Any as TypingAny
+from typing import List, Tuple
 
 from pydispatch.dispatcher import (
     Anonymous,
@@ -20,7 +22,9 @@ from scrapy.utils.log import failure_to_exc_info
 logger = logging.getLogger(__name__)
 
 
-def send_catch_log(signal=Any, sender=Anonymous, *arguments, **named):
+def send_catch_log(
+    signal=Any, sender=Anonymous, *arguments, **named
+) -> List[Tuple[TypingAny, TypingAny]]:
     """Like pydispatcher.robust.sendRobust but it also logs errors and returns
     Failures instead of exceptions.
     """
@@ -32,8 +36,9 @@ def send_catch_log(signal=Any, sender=Anonymous, *arguments, **named):
     )
     dont_log += (StopDownload,)
     spider = named.get("spider", None)
-    responses = []
+    responses: List[Tuple[TypingAny, TypingAny]] = []
     for receiver in liveReceivers(getAllReceivers(sender, signal)):
+        result: TypingAny
         try:
             response = robustApply(
                 receiver, signal=signal, sender=sender, *arguments, **named
diff --git a/scrapy/utils/ssl.py b/scrapy/utils/ssl.py
index 3ddceea35..03ae4ba9e 100644
--- a/scrapy/utils/ssl.py
+++ b/scrapy/utils/ssl.py
@@ -1,24 +1,28 @@
+from typing import Any, Optional, cast
+
 import OpenSSL._util as pyOpenSSLutil
 import OpenSSL.SSL
+import OpenSSL.version
+from OpenSSL.crypto import X509Name
 
 from scrapy.utils.python import to_unicode
 
 
-def ffi_buf_to_string(buf):
+def ffi_buf_to_string(buf: Any) -> str:
     return to_unicode(pyOpenSSLutil.ffi.string(buf))
 
 
-def x509name_to_string(x509name):
+def x509name_to_string(x509name: X509Name) -> str:
     # from OpenSSL.crypto.X509Name.__repr__
-    result_buffer = pyOpenSSLutil.ffi.new("char[]", 512)
+    result_buffer: Any = pyOpenSSLutil.ffi.new("char[]", 512)
     pyOpenSSLutil.lib.X509_NAME_oneline(
-        x509name._name, result_buffer, len(result_buffer)
+        x509name._name, result_buffer, len(result_buffer)  # type: ignore[attr-defined]
     )
 
     return ffi_buf_to_string(result_buffer)
 
 
-def get_temp_key_info(ssl_object):
+def get_temp_key_info(ssl_object: Any) -> Optional[str]:
     # adapted from OpenSSL apps/s_cb.c::ssl_print_tmp_key()
     if not hasattr(pyOpenSSLutil.lib, "SSL_get_server_tmp_key"):
         # removed in cryptography 40.0.0
@@ -53,8 +57,10 @@ def get_temp_key_info(ssl_object):
     return ", ".join(key_info)
 
 
-def get_openssl_version():
-    system_openssl = OpenSSL.SSL.SSLeay_version(OpenSSL.SSL.SSLEAY_VERSION).decode(
-        "ascii", errors="replace"
+def get_openssl_version() -> str:
+    # https://github.com/python/typeshed/issues/10024
+    system_openssl_bytes = cast(
+        bytes, OpenSSL.SSL.SSLeay_version(OpenSSL.SSL.SSLEAY_VERSION)
     )
+    system_openssl = system_openssl_bytes.decode("ascii", errors="replace")
     return f"{OpenSSL.version.__version__} ({system_openssl})"
diff --git a/tox.ini b/tox.ini
index 5a9d9cf29..af8f1f57a 100644
--- a/tox.ini
+++ b/tox.ini
@@ -30,20 +30,20 @@ passenv =
 #allow tox virtualenv to upgrade pip/wheel/setuptools
 download = true
 commands =
-    pytest --cov=scrapy --cov-report=xml --cov-report= {posargs:--durations=10 docs scrapy tests}
+    pytest --cov=scrapy --cov-report=xml --cov-report= {posargs:--durations=10 docs scrapy tests} --doctest-modules
 install_command =
     python -I -m pip install -ctests/upper-constraints.txt {opts} {packages}
 
 [testenv:typing]
 basepython = python3
 deps =
-    mypy==1.0.1
+    mypy==1.2.0
     types-attrs==19.1.0
-    types-lxml==2023.2.11
-    types-Pillow==9.4.0.16
-    types-Pygments==2.14.0.5
-    types-pyOpenSSL==23.0.0.4
-    types-setuptools==67.4.0.1
+    types-lxml==2023.3.28
+    types-Pillow==9.4.0.19
+    types-Pygments==2.14.0.7
+    types-pyOpenSSL==23.1.0.1
+    types-setuptools==67.6.0.7
 commands =
     mypy --show-error-codes {posargs: scrapy tests}
 
@@ -58,7 +58,7 @@ commands =
 basepython = python3
 deps =
     {[testenv:extra-deps]deps}
-    pylint==2.16.0
+    pylint==2.17.2
 commands =
     pylint conftest.py docs extras scrapy setup.py tests
 
@@ -99,14 +99,18 @@ setenv =
     _SCRAPY_PINNED=true
 install_command =
     python -I -m pip install {opts} {packages}
+commands =
+    pytest --cov=scrapy --cov-report=xml --cov-report= {posargs:--durations=10 scrapy tests}
 
 [testenv:pinned]
+basepython = python3.7
 deps =
     {[pinned]deps}
     PyDispatcher==2.0.5
 install_command = {[pinned]install_command}
 setenv =
     {[pinned]setenv}
+commands = {[pinned]commands}
 
 [testenv:windows-pinned]
 basepython = python3
@@ -116,6 +120,7 @@ deps =
 install_command = {[pinned]install_command}
 setenv =
     {[pinned]setenv}
+commands = {[pinned]commands}
 
 [testenv:extra-deps]
 basepython = python3
@@ -136,7 +141,7 @@ commands =
 
 [testenv:asyncio-pinned]
 deps = {[testenv:pinned]deps}
-commands = {[testenv:asyncio]commands}
+commands = {[pinned]commands} --reactor=asyncio
 install_command = {[pinned]install_command}
 setenv =
     {[pinned]setenv}
@@ -151,7 +156,8 @@ basepython = {[testenv:pypy3]basepython}
 deps =
     {[pinned]deps}
     PyPyDispatcher==2.1.0
-commands = {[testenv:pypy3]commands}
+commands =
+    pytest --durations=10 scrapy tests
 install_command = {[pinned]install_command}
 setenv =
     {[pinned]setenv}


Test patch:
diff --git a/pytest.ini b/pytest.ini
index f5fbf2529..866f0c950 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -5,7 +5,6 @@ python_files=test_*.py __init__.py
 python_classes=
 addopts =
     --assert=plain
-    --doctest-modules
     --ignore=docs/_ext
     --ignore=docs/conf.py
     --ignore=docs/news.rst
diff --git a/tests/test_feedexport.py b/tests/test_feedexport.py
index eafe1b334..b1059099a 100644
--- a/tests/test_feedexport.py
+++ b/tests/test_feedexport.py
@@ -32,18 +32,19 @@ from zope.interface import implementer
 from zope.interface.verify import verifyObject
 
 import scrapy
+from scrapy import signals
 from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
 from scrapy.exporters import CsvItemExporter, JsonItemExporter
 from scrapy.extensions.feedexport import (
     BlockingFeedStorage,
     FeedExporter,
+    FeedSlot,
     FileFeedStorage,
     FTPFeedStorage,
     GCSFeedStorage,
     IFeedStorage,
     S3FeedStorage,
     StdoutFeedStorage,
-    _FeedSlot,
 )
 from scrapy.settings import Settings
 from scrapy.utils.python import to_unicode
@@ -660,8 +661,8 @@ class FeedExportTestBase(ABC, unittest.TestCase):
         return result
 
 
-class InstrumentedFeedSlot(_FeedSlot):
-    """Instrumented _FeedSlot subclass for keeping track of calls to
+class InstrumentedFeedSlot(FeedSlot):
+    """Instrumented FeedSlot subclass for keeping track of calls to
     start_exporting and finish_exporting."""
 
     def start_exporting(self):
@@ -964,7 +965,7 @@ class FeedExportTest(FeedExportTestBase):
         listener = IsExportingListener()
         InstrumentedFeedSlot.subscribe__listener(listener)
 
-        with mock.patch("scrapy.extensions.feedexport._FeedSlot", InstrumentedFeedSlot):
+        with mock.patch("scrapy.extensions.feedexport.FeedSlot", InstrumentedFeedSlot):
             _ = yield self.exported_data(items, settings)
             self.assertFalse(listener.start_without_finish)
             self.assertFalse(listener.finish_without_start)
@@ -982,7 +983,7 @@ class FeedExportTest(FeedExportTestBase):
         listener = IsExportingListener()
         InstrumentedFeedSlot.subscribe__listener(listener)
 
-        with mock.patch("scrapy.extensions.feedexport._FeedSlot", InstrumentedFeedSlot):
+        with mock.patch("scrapy.extensions.feedexport.FeedSlot", InstrumentedFeedSlot):
             _ = yield self.exported_data(items, settings)
             self.assertFalse(listener.start_without_finish)
             self.assertFalse(listener.finish_without_start)
@@ -1003,7 +1004,7 @@ class FeedExportTest(FeedExportTestBase):
         listener = IsExportingListener()
         InstrumentedFeedSlot.subscribe__listener(listener)
 
-        with mock.patch("scrapy.extensions.feedexport._FeedSlot", InstrumentedFeedSlot):
+        with mock.patch("scrapy.extensions.feedexport.FeedSlot", InstrumentedFeedSlot):
             _ = yield self.exported_data(items, settings)
             self.assertFalse(listener.start_without_finish)
             self.assertFalse(listener.finish_without_start)
@@ -1022,7 +1023,7 @@ class FeedExportTest(FeedExportTestBase):
         listener = IsExportingListener()
         InstrumentedFeedSlot.subscribe__listener(listener)
 
-        with mock.patch("scrapy.extensions.feedexport._FeedSlot", InstrumentedFeedSlot):
+        with mock.patch("scrapy.extensions.feedexport.FeedSlot", InstrumentedFeedSlot):
             _ = yield self.exported_data(items, settings)
             self.assertFalse(listener.start_without_finish)
             self.assertFalse(listener.finish_without_start)
@@ -1638,6 +1639,57 @@ class FeedExportTest(FeedExportTestBase):
             data = yield self.exported_data(items, settings)
             self.assertEqual(row["expected"], data[feed_options["format"]])
 
+    @defer.inlineCallbacks
+    def test_storage_file_no_postprocessing(self):
+        @implementer(IFeedStorage)
+        class Storage:
+            def __init__(self, uri, *, feed_options=None):
+                pass
+
+            def open(self, spider):
+                Storage.open_file = tempfile.NamedTemporaryFile(prefix="feed-")
+                return Storage.open_file
+
+            def store(self, file):
+                Storage.store_file = file
+                file.close()
+
+        settings = {
+            "FEEDS": {self._random_temp_filename(): {"format": "jsonlines"}},
+            "FEED_STORAGES": {"file": Storage},
+        }
+        yield self.exported_no_data(settings)
+        self.assertIs(Storage.open_file, Storage.store_file)
+
+    @defer.inlineCallbacks
+    def test_storage_file_postprocessing(self):
+        @implementer(IFeedStorage)
+        class Storage:
+            def __init__(self, uri, *, feed_options=None):
+                pass
+
+            def open(self, spider):
+                Storage.open_file = tempfile.NamedTemporaryFile(prefix="feed-")
+                return Storage.open_file
+
+            def store(self, file):
+                Storage.store_file = file
+                file.close()
+
+        settings = {
+            "FEEDS": {
+                self._random_temp_filename(): {
+                    "format": "jsonlines",
+                    "postprocessing": [
+                        "scrapy.extensions.postprocessing.GzipPlugin",
+                    ],
+                },
+            },
+            "FEED_STORAGES": {"file": Storage},
+        }
+        yield self.exported_no_data(settings)
+        self.assertIs(Storage.open_file, Storage.store_file)
+
 
 class FeedPostProcessedExportsTest(FeedExportTestBase):
     __test__ = True
@@ -2491,7 +2543,7 @@ class BatchDeliveriesTest(FeedExportTestBase):
     def test_batch_path_differ(self):
         """
         Test that the name of all batch files differ from each other.
-        So %(batch_time)s replaced with the current date.
+        So %(batch_id)d replaced with the current id.
         """
         items = [
             self.MyItem({"foo": "bar1", "egg": "spam1"}),
@@ -2501,7 +2553,7 @@ class BatchDeliveriesTest(FeedExportTestBase):
         settings = {
             "FEEDS": {
                 self._random_temp_filename()
-                / "%(batch_time)s": {
+                / "%(batch_id)d": {
                     "format": "json",
                 },
             },
@@ -2564,7 +2616,7 @@ class BatchDeliveriesTest(FeedExportTestBase):
                 return super().open(*args, **kwargs)
 
         key = "export.csv"
-        uri = f"s3://{bucket}/{key}/%(batch_time)s.json"
+        uri = f"s3://{bucket}/{key}/%(batch_id)d.json"
         batch_item_count = 1
         settings = {
             "AWS_ACCESS_KEY_ID": "access_key",
@@ -2600,6 +2652,83 @@ class BatchDeliveriesTest(FeedExportTestBase):
             stub.assert_no_pending_responses()
 
 
+# Test that the FeedExporer sends the feed_exporter_closed and feed_slot_closed signals
+class FeedExporterSignalsTest(unittest.TestCase):
+    items = [
+        {"foo": "bar1", "egg": "spam1"},
+        {"foo": "bar2", "egg": "spam2", "baz": "quux2"},
+        {"foo": "bar3", "baz": "quux3"},
+    ]
+
+    with tempfile.NamedTemporaryFile(suffix="json") as tmp:
+        settings = {
+            "FEEDS": {
+                f"file:///{tmp.name}": {
+                    "format": "json",
+                },
+            },
+        }
+
+    def feed_exporter_closed_signal_handler(self):
+        self.feed_exporter_closed_received = True
+
+    def feed_slot_closed_signal_handler(self, slot):
+        self.feed_slot_closed_received = True
+
+    def feed_exporter_closed_signal_handler_deferred(self):
+        d = defer.Deferred()
+        d.addCallback(lambda _: setattr(self, "feed_exporter_closed_received", True))
+        d.callback(None)
+        return d
+
+    def feed_slot_closed_signal_handler_deferred(self, slot):
+        d = defer.Deferred()
+        d.addCallback(lambda _: setattr(self, "feed_slot_closed_received", True))
+        d.callback(None)
+        return d
+
+    def run_signaled_feed_exporter(
+        self, feed_exporter_signal_handler, feed_slot_signal_handler
+    ):
+        crawler = get_crawler(settings_dict=self.settings)
+        feed_exporter = FeedExporter.from_crawler(crawler)
+        spider = scrapy.Spider("default")
+        spider.crawler = crawler
+        crawler.signals.connect(
+            feed_exporter_signal_handler,
+            signal=signals.feed_exporter_closed,
+        )
+        crawler.signals.connect(
+            feed_slot_signal_handler, signal=signals.feed_slot_closed
+        )
+        feed_exporter.open_spider(spider)
+        for item in self.items:
+            feed_exporter.item_scraped(item, spider)
+        defer.ensureDeferred(feed_exporter.close_spider(spider))
+
+    def test_feed_exporter_signals_sent(self):
+        self.feed_exporter_closed_received = False
+        self.feed_slot_closed_received = False
+
+        self.run_signaled_feed_exporter(
+            self.feed_exporter_closed_signal_handler,
+            self.feed_slot_closed_signal_handler,
+        )
+        self.assertTrue(self.feed_slot_closed_received)
+        self.assertTrue(self.feed_exporter_closed_received)
+
+    def test_feed_exporter_signals_sent_deferred(self):
+        self.feed_exporter_closed_received = False
+        self.feed_slot_closed_received = False
+
+        self.run_signaled_feed_exporter(
+            self.feed_exporter_closed_signal_handler_deferred,
+            self.feed_slot_closed_signal_handler_deferred,
+        )
+        self.assertTrue(self.feed_slot_closed_received)
+        self.assertTrue(self.feed_exporter_closed_received)
+
+
 class FeedExportInitTest(unittest.TestCase):
     def test_unsupported_storage(self):
         settings = {
diff --git a/tests/test_linkextractors.py b/tests/test_linkextractors.py
index f663013ba..e1ec19601 100644
--- a/tests/test_linkextractors.py
+++ b/tests/test_linkextractors.py
@@ -2,6 +2,10 @@ import pickle
 import re
 import unittest
 
+from packaging.version import Version
+from pytest import mark
+from w3lib import __version__ as w3lib_version
+
 from scrapy.http import HtmlResponse, XmlResponse
 from scrapy.link import Link
 from scrapy.linkextractors.lxmlhtml import LxmlLinkExtractor
@@ -815,3 +819,34 @@ class LxmlLinkExtractorTestCase(Base.LinkExtractorTestCase):
 
     def test_restrict_xpaths_with_html_entities(self):
         super().test_restrict_xpaths_with_html_entities()
+
+    @mark.skipif(
+        Version(w3lib_version) < Version("2.0.0"),
+        reason=(
+            "Before w3lib 2.0.0, w3lib.url.safe_url_string would not complain "
+            "about an invalid port value."
+        ),
+    )
+    def test_skip_bad_links(self):
+        html = b"""
+        <a href="http://example.org:non-port">Why would you do this?</a>
+        <a href="http://example.org/item2.html">Good Link</a>
+        <a href="http://example.org/item3.html">Good Link 2</a>
+        """
+        response = HtmlResponse("http://example.org/index.html", body=html)
+        lx = self.extractor_cls()
+        self.assertEqual(
+            [link for link in lx.extract_links(response)],
+            [
+                Link(
+                    url="http://example.org/item2.html",
+                    text="Good Link",
+                    nofollow=False,
+                ),
+                Link(
+                    url="http://example.org/item3.html",
+                    text="Good Link 2",
+                    nofollow=False,
+                ),
+            ],
+        )
diff --git a/tests/test_pipeline_files.py b/tests/test_pipeline_files.py
index 9701e5d4e..c80666586 100644
--- a/tests/test_pipeline_files.py
+++ b/tests/test_pipeline_files.py
@@ -601,7 +601,7 @@ class TestGCSFilesStore(unittest.TestCase):
         s = yield store.stat_file(path, info=None)
         self.assertIn("last_modified", s)
         self.assertIn("checksum", s)
-        self.assertEqual(s["checksum"], "zc2oVgXkbQr2EQdSdw3OPA==")
+        self.assertEqual(s["checksum"], "cdcda85605e46d0af6110752770dce3c")
         u = urlparse(uri)
         content, acl, blob = get_gcs_content_and_delete(u.hostname, u.path[1:] + path)
         self.assertEqual(content, data)
diff --git a/tests/test_utils_curl.py b/tests/test_utils_curl.py
index fd4612eba..1816db29b 100644
--- a/tests/test_utils_curl.py
+++ b/tests/test_utils_curl.py
@@ -154,6 +154,15 @@ class CurlToRequestKwargsTest(unittest.TestCase):
         }
         self._test_command(curl_command, expected_result)
 
+    def test_post_data_raw_with_string_prefix(self):
+        curl_command = "curl 'https://www.example.org/' --data-raw $'{\"$filters\":\"Filter\u0021\"}'"
+        expected_result = {
+            "method": "POST",
+            "url": "https://www.example.org/",
+            "body": '{"$filters":"Filter!"}',
+        }
+        self._test_command(curl_command, expected_result)
+
     def test_explicit_get_with_data(self):
         curl_command = "curl httpbin.org/anything -X GET --data asdf"
         expected_result = {
diff --git a/tests/test_utils_python.py b/tests/test_utils_python.py
index 57f40c2e5..80d2e8da1 100644
--- a/tests/test_utils_python.py
+++ b/tests/test_utils_python.py
@@ -235,20 +235,16 @@ class UtilsPythonTestCase(unittest.TestCase):
         self.assertEqual(get_func_args(partial_f3), ["c"])
         self.assertEqual(get_func_args(cal), ["a", "b", "c"])
         self.assertEqual(get_func_args(object), [])
+        self.assertEqual(get_func_args(str.split, stripself=True), ["sep", "maxsplit"])
+        self.assertEqual(get_func_args(" ".join, stripself=True), ["iterable"])
 
         if platform.python_implementation() == "CPython":
-            # TODO: how do we fix this to return the actual argument names?
-            self.assertEqual(get_func_args(str.split), [])
-            self.assertEqual(get_func_args(" ".join), [])
+            # doesn't work on CPython: https://bugs.python.org/issue42785
             self.assertEqual(get_func_args(operator.itemgetter(2)), [])
         elif platform.python_implementation() == "PyPy":
-            self.assertEqual(
-                get_func_args(str.split, stripself=True), ["sep", "maxsplit"]
-            )
             self.assertEqual(
                 get_func_args(operator.itemgetter(2), stripself=True), ["obj"]
             )
-            self.assertEqual(get_func_args(" ".join, stripself=True), ["iterable"])
 
     def test_without_none_values(self):
         self.assertEqual(without_none_values([1, None, 3, 4]), [1, 3, 4])
diff --git a/tests/test_utils_request.py b/tests/test_utils_request.py
index 6ca272de1..e6d1abe3f 100644
--- a/tests/test_utils_request.py
+++ b/tests/test_utils_request.py
@@ -1,3 +1,4 @@
+import json
 import unittest
 import warnings
 from hashlib import sha1
@@ -18,6 +19,7 @@ from scrapy.utils.request import (
     request_authenticate,
     request_fingerprint,
     request_httprepr,
+    request_to_curl,
 )
 from scrapy.utils.test import get_crawler
 
@@ -666,5 +668,67 @@ class CustomRequestFingerprinterTestCase(unittest.TestCase):
         self.assertEqual(fingerprint, settings["FINGERPRINT"])
 
 
+class RequestToCurlTest(unittest.TestCase):
+    def _test_request(self, request_object, expected_curl_command):
+        curl_command = request_to_curl(request_object)
+        self.assertEqual(curl_command, expected_curl_command)
+
+    def test_get(self):
+        request_object = Request("https://www.example.com")
+        expected_curl_command = "curl -X GET https://www.example.com"
+        self._test_request(request_object, expected_curl_command)
+
+    def test_post(self):
+        request_object = Request(
+            "https://www.httpbin.org/post",
+            method="POST",
+            body=json.dumps({"foo": "bar"}),
+        )
+        expected_curl_command = (
+            'curl -X POST https://www.httpbin.org/post --data-raw \'{"foo": "bar"}\''
+        )
+        self._test_request(request_object, expected_curl_command)
+
+    def test_headers(self):
+        request_object = Request(
+            "https://www.httpbin.org/post",
+            method="POST",
+            headers={"Content-Type": "application/json", "Accept": "application/json"},
+            body=json.dumps({"foo": "bar"}),
+        )
+        expected_curl_command = (
+            "curl -X POST https://www.httpbin.org/post"
+            ' --data-raw \'{"foo": "bar"}\''
+            " -H 'Content-Type: application/json' -H 'Accept: application/json'"
+        )
+        self._test_request(request_object, expected_curl_command)
+
+    def test_cookies_dict(self):
+        request_object = Request(
+            "https://www.httpbin.org/post",
+            method="POST",
+            cookies={"foo": "bar"},
+            body=json.dumps({"foo": "bar"}),
+        )
+        expected_curl_command = (
+            "curl -X POST https://www.httpbin.org/post"
+            " --data-raw '{\"foo\": \"bar\"}' --cookie 'foo=bar'"
+        )
+        self._test_request(request_object, expected_curl_command)
+
+    def test_cookies_list(self):
+        request_object = Request(
+            "https://www.httpbin.org/post",
+            method="POST",
+            cookies=[{"foo": "bar"}],
+            body=json.dumps({"foo": "bar"}),
+        )
+        expected_curl_command = (
+            "curl -X POST https://www.httpbin.org/post"
+            " --data-raw '{\"foo\": \"bar\"}' --cookie 'foo=bar'"
+        )
+        self._test_request(request_object, expected_curl_command)
+
+
 if __name__ == "__main__":
     unittest.main()