diff --git a/docs/index.rst b/docs/index.rst
index 71047f9ef..fe61f3513 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -229,6 +229,7 @@ Extending Scrapy
    topics/signals
    topics/scheduler
    topics/exporters
+   topics/download-handlers
    topics/components
    topics/api
 
@@ -257,6 +258,9 @@ Extending Scrapy
 :doc:`topics/exporters`
     Quickly export your scraped items to a file (XML, CSV, etc).
 
+:doc:`topics/download-handlers`
+    Customize how requests are downloaded or add support for new URL schemes.
+
 :doc:`topics/components`
     Learn the common API and some good practices when building custom Scrapy
     components.
diff --git a/docs/news.rst b/docs/news.rst
index 19d990a50..7e15c2167 100644
--- a/docs/news.rst
+++ b/docs/news.rst
@@ -74,6 +74,12 @@ Backward-incompatible changes
     :class:`~scrapy.downloadermiddlewares.robotstxt.RobotsTxtMiddleware` or
     calls its methods directly.
 
+-   The built-in :ref:`download handlers <download-handlers-ref>` were
+    refactored, changing signatures of their methods. This change should only
+    affect user code that subclasses any of these handlers or calls their
+    methods directly.
+    (:issue:`6778`, :issue:`7164`)
+
 -   :meth:`scrapy.pipelines.media.MediaPipeline.process_item` now returns a
     coroutine, previously it returned a
     :class:`~twisted.internet.defer.Deferred` object. This
@@ -2616,7 +2622,7 @@ Modified requirements
 ~~~~~~~~~~~~~~~~~~~~~
 
 -   The h2_ dependency is now optional, only needed to
-    :ref:`enable HTTP/2 support <http2>`. (:issue:`5113`)
+    :ref:`enable HTTP/2 support <twisted-http2-handler>`. (:issue:`5113`)
 
     .. _h2: https://pypi.org/project/h2/
 
@@ -2998,7 +3004,7 @@ Highlights:
 
 -   Official Python 3.9 support
 
--   Experimental :ref:`HTTP/2 support <http2>`
+-   Experimental :ref:`HTTP/2 support <twisted-http2-handler>`
 
 -   New :func:`~scrapy.downloadermiddlewares.retry.get_retry_request` function
     to retry requests from spider callbacks
@@ -3029,7 +3035,7 @@ Deprecations
 New features
 ~~~~~~~~~~~~
 
--   Experimental :ref:`HTTP/2 support <http2>` through a new download handler
+-   Experimental :ref:`HTTP/2 support <twisted-http2-handler>` through a new download handler
     that can be assigned to the ``https`` protocol in the
     :setting:`DOWNLOAD_HANDLERS` setting.
     (:issue:`1854`, :issue:`4769`, :issue:`5058`, :issue:`5059`, :issue:`5066`)
diff --git a/docs/requirements.txt b/docs/requirements.txt
index 81a3119e4..500db7a7e 100644
--- a/docs/requirements.txt
+++ b/docs/requirements.txt
@@ -1,3 +1,4 @@
+h2==4.3.0
 pydantic==2.12.3
 scrapy-spider-metadata==0.2.0
 sphinx==8.1.3
diff --git a/docs/topics/addons.rst b/docs/topics/addons.rst
index dba14e749..64f00f622 100644
--- a/docs/topics/addons.rst
+++ b/docs/topics/addons.rst
@@ -175,16 +175,19 @@ Use a fallback component:
     class MyHandler:
         lazy = False
 
-        def __init__(self, settings, crawler):
-            dhcls = load_object(settings.get(FALLBACK_SETTING))
+        def __init__(self, crawler):
+            dhcls = load_object(crawler.settings.get(FALLBACK_SETTING))
             self._fallback_handler = build_from_crawler(dhcls, crawler)
 
-        def download_request(self, request, spider):
+        async def download_request(self, request):
             if request.meta.get("my_params"):
                 # handle the request
                 ...
             else:
-                return self._fallback_handler.download_request(request, spider)
+                return await self._fallback_handler.download_request(request)
+
+        async def close(self):
+            pass
 
 
     class MyAddon:
diff --git a/docs/topics/coroutines.rst b/docs/topics/coroutines.rst
index c82e6c2f3..a78e98828 100644
--- a/docs/topics/coroutines.rst
+++ b/docs/topics/coroutines.rst
@@ -55,6 +55,10 @@ hence use coroutine syntax (e.g. ``await``, ``async for``, ``async with``):
 
 -   :ref:`Signal handlers that support deferreds <signal-deferred>`.
 
+-   Methods of :ref:`download handlers <topics-download-handlers>`.
+
+    .. versionadded:: VERSION
+
 
 .. _coroutine-deferred-apis:
 
@@ -95,12 +99,6 @@ The following user-supplied methods can return
 :class:`~twisted.internet.defer.Deferred` objects (the methods that can also
 return coroutines are listed in :ref:`coroutine-support`):
 
--   Custom download handlers (see :setting:`DOWNLOAD_HANDLERS`):
-
-    - ``download_request()``
-
-    - ``close()``
-
 -   Custom downloader implementations (see :setting:`DOWNLOADER`):
 
     - ``fetch()``
@@ -146,11 +144,11 @@ For example:
     email is sent. You can use this object directly in Deferred-based code or
     convert it into a :class:`~asyncio.Future` object with
     :func:`~scrapy.utils.defer.maybe_deferred_to_future`.
--   A custom download handler needs to define a ``download_request()`` method
-    that returns a :class:`~twisted.internet.defer.Deferred` object. You can
-    write a method that works with Deferreds and returns one directly, or you
-    can write a coroutine and convert it into a function that returns a
-    Deferred with :func:`~scrapy.utils.defer.deferred_f_from_coro_f`.
+-   A custom scheduler needs to define an ``open()`` method that can return a
+    :class:`~twisted.internet.defer.Deferred` object. You can write a method
+    that works with Deferreds and returns one directly, or you can write a
+    coroutine and convert it into a function that returns a Deferred with
+    :func:`~scrapy.utils.defer.deferred_f_from_coro_f`.
 
 
 General usage
diff --git a/docs/topics/download-handlers.rst b/docs/topics/download-handlers.rst
new file mode 100644
index 000000000..5f00d7131
--- /dev/null
+++ b/docs/topics/download-handlers.rst
@@ -0,0 +1,219 @@
+.. _topics-download-handlers:
+
+=================
+Download handlers
+=================
+
+Download handlers are Scrapy :ref:`components <topics-components>` used to
+download :ref:`requests <topics-request-response>` and produce responses from
+them.
+
+Using download handlers
+=======================
+
+The :setting:`DOWNLOAD_HANDLERS_BASE` and :setting:`DOWNLOAD_HANDLERS` settings
+tell Scrapy which handler is responsible for a given URL scheme. Their values
+are merged into a mapping from scheme names to handler classes. When Scrapy
+initializes it creates instances of all configured download handlers (except
+for :ref:`lazy ones <lazy-download-handlers>`) and stores them in a similar
+mapping. When Scrapy needs to download a request it extracts the scheme from
+its URL, finds the handler for this scheme, passes the request to it and gets a
+response from it.  If there is no handler for the scheme, the request is not
+downloaded and a :exc:`~scrapy.exceptions.NotSupported` exception is raised.
+
+The :setting:`DOWNLOAD_HANDLERS_BASE` setting contains the default mapping of
+handlers. You can use the :setting:`DOWNLOAD_HANDLERS` setting to add handlers
+for additional schemes and to replace or disable default ones:
+
+.. code-block:: python
+
+    DOWNLOAD_HANDLERS = {
+        # disable support for ftp:// requests
+        "ftp": None,
+        # replace the default one for http://
+        "http": "my.download_handlers.HttpHandler",
+        # http:// and https:// are different schemes,
+        # even though they may use the same handler
+        "https": "my.download_handlers.HttpHandler",
+        # support for any custom scheme can be added
+        "sftp": "my.download_handlers.SftpHandler",
+    }
+
+Replacing HTTP(S) download handlers
+-----------------------------------
+
+While Scrapy provides a default handler for ``http`` and ``https`` schemes,
+users may want to use a different handler, provided by Scrapy or by some
+3rd-party package. There are several considerations to keep in mind related to
+this.
+
+First of all, as ``http`` and ``https`` are separate schemes, they need
+separate entries in the :setting:`DOWNLOAD_HANDLERS` setting, even though it's
+likely that the same handler class will be used for both schemes.
+
+Additionally, some of the Scrapy settings, like :setting:`DOWNLOAD_MAXSIZE`,
+are honored by the default HTTP(S) handler but not necessarily by alternative
+ones. The same may apply to other Scrapy features, e.g. the
+:signal:`bytes_received` and :signal:`headers_received` signals.
+
+.. _lazy-download-handlers:
+
+Lazy instantiation of download handlers
+---------------------------------------
+
+A download handler can be marked as "lazy" by setting its ``lazy`` class
+attribute to ``True``. Such handlers are only instantiated when they need to
+download their first request. This may be useful when the instantiation is slow
+or requires dependencies that are not always available, and the handler is not
+needed on every spider run. For example, :class:`the built-in S3 handler
+<.S3DownloadHandler>` is lazy.
+
+Writing your own download handler
+=================================
+
+A download handler is a :ref:`component <topics-components>` that defines
+the following methods:
+
+.. class:: SampleDownloadHandler
+
+    .. attribute:: lazy
+        :type: bool
+
+        If ``False``, the handler will be instantiated when Scrapy is
+        initialized.
+
+        If ``True``, the handler will only be instantiated when the first
+        request handled by it needs to be downloaded.
+
+    .. method:: download_request(request: Request) -> Response:
+        :async:
+
+        Download the given request and return a response.
+
+    .. method:: close() -> None
+        :async:
+
+        Clean up any resources used by the handler.
+
+An optional base class for custom handlers is provided:
+
+.. autoclass:: scrapy.core.downloader.handlers.base.BaseDownloadHandler
+    :members:
+    :undoc-members:
+    :member-order: bysource
+
+.. _download-handlers-ref:
+
+Built-in download handlers reference
+====================================
+
+DataURIDownloadHandler
+----------------------
+
+.. autoclass:: scrapy.core.downloader.handlers.datauri.DataURIDownloadHandler
+
+| Supported scheme: ``data``.
+| Lazy: no.
+
+This handler supports RFC 2397 ``data:content/type;base64,`` data URIs.
+
+FileDownloadHandler
+-------------------
+
+.. autoclass:: scrapy.core.downloader.handlers.file.FileDownloadHandler
+
+| Supported scheme: ``file``.
+| Lazy: no.
+
+This handler supports ``file:///path`` local file URIs. It doesn't
+support remote files.
+
+FTPDownloadHandler
+------------------
+
+.. autoclass:: scrapy.core.downloader.handlers.ftp.FTPDownloadHandler
+
+| Supported scheme: ``ftp``.
+| Lazy: no.
+
+This handler supports ``ftp://host/path`` FTP URIs.
+
+It's implemented using :mod:`twisted.protocols.ftp`.
+
+.. _twisted-http2-handler:
+
+H2DownloadHandler
+-----------------
+
+.. autoclass:: scrapy.core.downloader.handlers.http2.H2DownloadHandler
+
+| Supported scheme: ``https``.
+| Lazy: yes.
+
+This handler supports ``https://host/path`` URLs and uses the HTTP/2 protocol
+for them.
+
+It's implemented using :mod:`twisted.web.client` and the ``h2`` library.
+
+For this handler to work you need to install the ``Twisted[http2]`` extra
+dependency.
+
+If you want to use this handler you need to replace the default one for the
+``https`` scheme:
+
+.. code-block:: python
+
+    DOWNLOAD_HANDLERS = {
+        "https": "scrapy.core.downloader.handlers.http2.H2DownloadHandler",
+    }
+
+.. warning::
+
+    This handler is experimental, and not yet recommended for production
+    environments. Future Scrapy versions may introduce related changes without
+    a deprecation period or warning.
+
+.. note::
+
+    Known limitations of the HTTP/2 implementation in this handler include:
+
+    -   No support for HTTP/2 Cleartext (h2c), since no major browser supports
+        HTTP/2 unencrypted (refer `http2 faq`_).
+
+    -   No setting to specify a maximum `frame size`_ larger than the default
+        value, 16384. Connections to servers that send a larger frame will
+        fail.
+
+    -   No support for `server pushes`_, which are ignored.
+
+    -   No support for the :signal:`bytes_received` and
+        :signal:`headers_received` signals.
+
+.. _frame size: https://datatracker.ietf.org/doc/html/rfc7540#section-4.2
+.. _http2 faq: https://http2.github.io/faq/#does-http2-require-encryption
+.. _server pushes: https://datatracker.ietf.org/doc/html/rfc7540#section-8.2
+
+HTTP11DownloadHandler
+---------------------
+
+.. autoclass:: scrapy.core.downloader.handlers.http11.HTTP11DownloadHandler
+
+| Supported schemes: ``http``, ``https``.
+| Lazy: no.
+
+This handler supports ``http://host/path`` and ``https://host/path`` URLs and
+uses the HTTP/1.1 protocol for them.
+
+It's implemented using :mod:`twisted.web.client`.
+
+S3DownloadHandler
+-----------------
+
+.. autoclass:: scrapy.core.downloader.handlers.s3.S3DownloadHandler
+
+| Supported scheme: ``s3``.
+| Lazy: yes.
+
+This handler supports ``s3://bucket/path`` S3 URIs.
+
+It's implemented using the ``botocore`` library and needs it to be installed.
diff --git a/docs/topics/downloader-middleware.rst b/docs/topics/downloader-middleware.rst
index c78045c4b..9be984f5d 100644
--- a/docs/topics/downloader-middleware.rst
+++ b/docs/topics/downloader-middleware.rst
@@ -129,9 +129,10 @@ defines one or more of these methods:
 
    .. method:: process_exception(request, exception)
 
-      Scrapy calls :meth:`process_exception` when a download handler
-      or a :meth:`process_request` (from a downloader middleware) raises an
-      exception (including an :exc:`~scrapy.exceptions.IgnoreRequest` exception)
+      Scrapy calls :meth:`process_exception` when a :ref:`download handler
+      <topics-download-handlers>` or a :meth:`process_request` (from a
+      downloader middleware) raises an exception (including an
+      :exc:`~scrapy.exceptions.IgnoreRequest` exception).
 
       :meth:`process_exception` should return: either ``None``,
       a :class:`~scrapy.http.Response` object, or a :class:`~scrapy.Request` object.
diff --git a/docs/topics/settings.rst b/docs/topics/settings.rst
index 27a8764f3..b446f197f 100644
--- a/docs/topics/settings.rst
+++ b/docs/topics/settings.rst
@@ -691,24 +691,6 @@ Default: ``'scrapy.core.downloader.Downloader'``
 
 The downloader to use for crawling.
 
-.. setting:: DOWNLOADER_HTTPCLIENTFACTORY
-
-DOWNLOADER_HTTPCLIENTFACTORY
-----------------------------
-
-Default: ``'scrapy.core.downloader.webclient.ScrapyHTTPClientFactory'``
-
-Defines a Twisted ``protocol.ClientFactory``  class to use for HTTP/1.0
-connections (for ``HTTP10DownloadHandler``).
-
-.. note::
-
-    HTTP/1.0 is rarely used nowadays and its Scrapy support is deprecated,
-    so you can safely ignore this setting,
-    unless you really want to use HTTP/1.0 and override
-    :setting:`DOWNLOAD_HANDLERS` for ``http(s)`` scheme accordingly,
-    i.e. to ``'scrapy.core.downloader.handlers.http10.HTTP10DownloadHandler'``.
-
 .. setting:: DOWNLOADER_CLIENTCONTEXTFACTORY
 
 DOWNLOADER_CLIENTCONTEXTFACTORY
@@ -738,6 +720,12 @@ accepts a ``method`` parameter (this is the ``OpenSSL.SSL`` method mapping
 parameter (``bool``) and a ``tls_ciphers`` parameter (see
 :setting:`DOWNLOADER_CLIENT_TLS_CIPHERS`).
 
+.. note::
+
+    This setting is specific to the built-in Twisted-based download handlers:
+    :class:`scrapy.core.downloader.handlers.http11.HTTP11DownloadHandler` and
+    :class:`scrapy.core.downloader.handlers.http2.H2DownloadHandler`.
+
 .. setting:: DOWNLOADER_CLIENT_TLS_CIPHERS
 
 DOWNLOADER_CLIENT_TLS_CIPHERS
@@ -745,8 +733,8 @@ DOWNLOADER_CLIENT_TLS_CIPHERS
 
 Default: ``'DEFAULT'``
 
-Use  this setting to customize the TLS/SSL ciphers used by the default
-HTTP/1.1 downloader.
+Use this setting to customize the TLS/SSL ciphers used by the HTTPS download
+handler.
 
 The setting should contain a string in the `OpenSSL cipher list format`_,
 these ciphers will be used as client ciphers. Changing this setting may be
@@ -756,6 +744,16 @@ specific cipher that is not included in ``DEFAULT`` if a website requires it.
 
 .. _OpenSSL cipher list format: https://docs.openssl.org/master/man1/openssl-ciphers/#cipher-list-format
 
+.. note::
+
+    Handling of this setting needs to be implemented inside the :ref:`download
+    handler <topics-download-handlers>`, so it's not guaranteed to be supported
+    by all 3rd-party handlers. Moreover, for the built-in Twisted-based
+    download handlers
+    (:class:`scrapy.core.downloader.handlers.http11.HTTP11DownloadHandler` and
+    :class:`scrapy.core.downloader.handlers.http2.H2DownloadHandler`) it needs
+    to be implemented in the :setting:`DOWNLOADER_CLIENTCONTEXTFACTORY` class.
+
 .. setting:: DOWNLOADER_CLIENT_TLS_METHOD
 
 DOWNLOADER_CLIENT_TLS_METHOD
@@ -763,8 +761,8 @@ DOWNLOADER_CLIENT_TLS_METHOD
 
 Default: ``'TLS'``
 
-Use this setting to customize the TLS/SSL method used by the default
-HTTP/1.1 downloader.
+Use this setting to customize the TLS/SSL method used by the HTTPS download
+handler.
 
 This setting must be one of these string values:
 
@@ -776,6 +774,15 @@ This setting must be one of these string values:
 - ``'TLSv1.1'``: forces TLS version 1.1
 - ``'TLSv1.2'``: forces TLS version 1.2
 
+.. note::
+
+    Handling of this setting needs to be implemented inside the :ref:`download
+    handler <topics-download-handlers>`, so it's not guaranteed to be supported
+    by all 3rd-party handlers. Moreover, for the built-in Twisted-based
+    download handlers
+    (:class:`scrapy.core.downloader.handlers.http11.HTTP11DownloadHandler` and
+    :class:`scrapy.core.downloader.handlers.http2.H2DownloadHandler`) it needs
+    to be implemented in the :setting:`DOWNLOADER_CLIENTCONTEXTFACTORY` class.
 
 .. setting:: DOWNLOADER_CLIENT_TLS_VERBOSE_LOGGING
 
@@ -786,10 +793,18 @@ Default: ``False``
 
 Setting this to ``True`` will enable DEBUG level messages about TLS connection
 parameters after establishing HTTPS connections. The kind of information logged
-depends on the versions of OpenSSL and pyOpenSSL.
+depends on the implementation of the download handler and the versions of
+the TLS-related libraries.
 
-This setting is only used for the default
-:setting:`DOWNLOADER_CLIENTCONTEXTFACTORY`.
+.. note::
+
+    Handling of this setting needs to be implemented inside the :ref:`download
+    handler <topics-download-handlers>`, so it's not guaranteed to be supported
+    by all 3rd-party handlers. Moreover, for the built-in Twisted-based
+    download handlers
+    (:class:`scrapy.core.downloader.handlers.http11.HTTP11DownloadHandler` and
+    :class:`scrapy.core.downloader.handlers.http2.H2DownloadHandler`) it needs
+    to be implemented in the :setting:`DOWNLOADER_CLIENTCONTEXTFACTORY` class.
 
 .. setting:: DOWNLOADER_MIDDLEWARES
 
@@ -882,7 +897,6 @@ It is also possible to change this setting per domain, although it requires
 non-trivial code. See the implementation of the :ref:`AutoThrottle
 <topics-autothrottle>` extension for an example.
 
-
 .. setting:: DOWNLOAD_HANDLERS
 
 DOWNLOAD_HANDLERS
@@ -890,7 +904,9 @@ DOWNLOAD_HANDLERS
 
 Default: ``{}``
 
-A dict containing the request downloader handlers enabled in your project.
+A dict containing the :ref:`download handlers <topics-download-handlers>`
+enabled in your project.
+
 See :setting:`DOWNLOAD_HANDLERS_BASE` for example format.
 
 .. setting:: DOWNLOAD_HANDLERS_BASE
@@ -912,9 +928,9 @@ Default:
     }
 
 
-A dict containing the request download handlers enabled by default in Scrapy.
-You should never modify this setting in your project, modify
-:setting:`DOWNLOAD_HANDLERS` instead.
+A dict containing the :ref:`download handlers <topics-download-handlers>`
+enabled by default in Scrapy. You should never modify this setting in your
+project, modify :setting:`DOWNLOAD_HANDLERS` instead.
 
 You can disable any of these download handlers by assigning ``None`` to their
 URI scheme in :setting:`DOWNLOAD_HANDLERS`. E.g., to disable the built-in FTP
@@ -926,46 +942,6 @@ handler (without replacement), place this in your ``settings.py``:
         "ftp": None,
     }
 
-.. _http2:
-
-The default HTTPS handler uses HTTP/1.1. To use HTTP/2:
-
-#.  Install ``Twisted[http2]>=17.9.0`` to install the packages required to
-    enable HTTP/2 support in Twisted.
-
-#.  Update :setting:`DOWNLOAD_HANDLERS` as follows:
-
-    .. code-block:: python
-
-        DOWNLOAD_HANDLERS = {
-            "https": "scrapy.core.downloader.handlers.http2.H2DownloadHandler",
-        }
-
-.. warning::
-
-    HTTP/2 support in Scrapy is experimental, and not yet recommended for
-    production environments. Future Scrapy versions may introduce related
-    changes without a deprecation period or warning.
-
-.. note::
-
-    Known limitations of the current HTTP/2 implementation of Scrapy include:
-
-    -   No support for HTTP/2 Cleartext (h2c), since no major browser supports
-        HTTP/2 unencrypted (refer `http2 faq`_).
-
-    -   No setting to specify a maximum `frame size`_ larger than the default
-        value, 16384. Connections to servers that send a larger frame will
-        fail.
-
-    -   No support for `server pushes`_, which are ignored.
-
-    -   No support for the :signal:`bytes_received` and
-        :signal:`headers_received` signals.
-
-.. _frame size: https://datatracker.ietf.org/doc/html/rfc7540#section-4.2
-.. _http2 faq: https://http2.github.io/faq/#does-http2-require-encryption
-.. _server pushes: https://datatracker.ietf.org/doc/html/rfc7540#section-8.2
 
 .. setting:: DOWNLOAD_SLOTS
 
@@ -1006,6 +982,12 @@ The amount of time (in secs) that the downloader will wait before timing out.
     This timeout can be per-request using the :reqmeta:`download_timeout`
     :attr:`.Request.meta` key.
 
+.. note::
+
+    Handling of this setting needs to be implemented inside the :ref:`download
+    handler <topics-download-handlers>`, so it's not guaranteed to be supported
+    by all 3rd-party handlers.
+
 .. setting:: DOWNLOAD_MAXSIZE
 .. reqmeta:: download_maxsize
 
@@ -1028,6 +1010,12 @@ Use ``0`` to disable this limit.
     This limit can be set per-request using the :reqmeta:`download_maxsize`
     :attr:`.Request.meta` key.
 
+.. note::
+
+    Checking responses before decompressing them needs to be implemented inside
+    the :ref:`download handler <topics-download-handlers>`, so it's not
+    guaranteed to be supported by all 3rd-party handlers.
+
 .. setting:: DOWNLOAD_WARNSIZE
 .. reqmeta:: download_warnsize
 
@@ -1046,6 +1034,12 @@ Use ``0`` to disable this limit.
     This limit can be set per-request using the :reqmeta:`download_warnsize`
     :attr:`.Request.meta` key.
 
+.. note::
+
+    Checking responses before decompressing them needs to be implemented inside
+    the :ref:`download handler <topics-download-handlers>`, so it's not
+    guaranteed to be supported by all 3rd-party handlers.
+
 .. setting:: DOWNLOAD_FAIL_ON_DATALOSS
 
 DOWNLOAD_FAIL_ON_DATALOSS
@@ -1072,11 +1066,17 @@ Optionally, this can be set per-request basis by using the
   If :setting:`RETRY_ENABLED` is ``True`` and this setting is set to ``True``,
   the ``ResponseFailed([_DataLoss])`` failure will be retried as usual.
 
+.. note::
+
+    Handling of this setting needs to be implemented inside the :ref:`download
+    handler <topics-download-handlers>`, so it's not guaranteed to be supported
+    by all 3rd-party handlers.
+
 .. warning::
 
     This setting is ignored by the
     :class:`~scrapy.core.downloader.handlers.http2.H2DownloadHandler`
-    download handler (see :setting:`DOWNLOAD_HANDLERS`). In case of a data loss
+    :ref:`download handler <topics-download-handlers>`. In case of a data loss
     error, the corresponding HTTP/2 connection may be corrupted, affecting other
     requests that use the same connection; hence, a ``ResponseFailed([InvalidBodyLengthError])``
     failure is always raised for every request that was using that connection.
@@ -1269,6 +1269,12 @@ Default: ``True``
 
 Whether or not to use passive mode when initiating FTP transfers.
 
+.. note::
+
+    Handling of this setting needs to be implemented inside the :ref:`download
+    handler <topics-download-handlers>`, so it's not guaranteed to be supported
+    by all 3rd-party handlers.
+
 .. reqmeta:: ftp_password
 .. setting:: FTP_PASSWORD
 
@@ -1288,6 +1294,12 @@ in ``Request`` meta.
 
 .. _RFC 1635: https://datatracker.ietf.org/doc/html/rfc1635
 
+.. note::
+
+    Handling of this setting needs to be implemented inside the :ref:`download
+    handler <topics-download-handlers>`, so it's not guaranteed to be supported
+    by all 3rd-party handlers.
+
 .. reqmeta:: ftp_user
 .. setting:: FTP_USER
 
@@ -1299,6 +1311,12 @@ Default: ``"anonymous"``
 The username to use for FTP connections when there is no ``"ftp_user"``
 in ``Request`` meta.
 
+.. note::
+
+    Handling of this setting needs to be implemented inside the :ref:`download
+    handler <topics-download-handlers>`, so it's not guaranteed to be supported
+    by all 3rd-party handlers.
+
 .. setting:: GCS_PROJECT_ID
 
 GCS_PROJECT_ID
diff --git a/scrapy/core/downloader/__init__.py b/scrapy/core/downloader/__init__.py
index db4e41c6a..85fa8f3c2 100644
--- a/scrapy/core/downloader/__init__.py
+++ b/scrapy/core/downloader/__init__.py
@@ -236,9 +236,7 @@ class Downloader:
         slot.transferring.add(request)
         try:
             # 1. Download the response
-            response: Response = await maybe_deferred_to_future(
-                self.handlers.download_request(request)
-            )
+            response: Response = await self.handlers.download_request_async(request)
             # 2. Notify response_downloaded listeners about the recent download
             # before querying queue for next request
             self.signals.send_catch_log(
diff --git a/scrapy/core/downloader/handlers/__init__.py b/scrapy/core/downloader/handlers/__init__.py
index 2d3d40dae..fb27cdb8b 100644
--- a/scrapy/core/downloader/handlers/__init__.py
+++ b/scrapy/core/downloader/handlers/__init__.py
@@ -2,16 +2,21 @@
 
 from __future__ import annotations
 
+import inspect
 import logging
+import warnings
 from typing import TYPE_CHECKING, Any, Protocol, cast
 
 from scrapy import Request, Spider, signals
-from scrapy.exceptions import NotConfigured, NotSupported
-from scrapy.utils.decorators import _warn_spider_arg
-from scrapy.utils.defer import ensure_awaitable
+from scrapy.exceptions import NotConfigured, NotSupported, ScrapyDeprecationWarning
+from scrapy.utils.defer import (
+    deferred_from_coro,
+    ensure_awaitable,
+    maybe_deferred_to_future,
+)
 from scrapy.utils.httpobj import urlparse_cached
 from scrapy.utils.misc import build_from_crawler, load_object
-from scrapy.utils.python import without_none_values
+from scrapy.utils.python import global_object_name, without_none_values
 
 if TYPE_CHECKING:
     from collections.abc import Callable
@@ -25,10 +30,20 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+# This is the official API but we temporarily support the old deprecated one:
+# * lazy is not mandatory (defaults to True).
+# * download_request() can return a Deferred[Response] instead of a coroutine,
+# and takes a spider argument in this case.
+# * close() can return None or Deferred[None] instead of a coroutine.
+# * close() is not mandatory.
+
+
 class DownloadHandlerProtocol(Protocol):
-    def download_request(
-        self, request: Request, spider: Spider
-    ) -> Deferred[Response]: ...
+    lazy: bool
+
+    async def download_request(self, request: Request) -> Response: ...
+
+    async def close(self) -> None: ...
 
 
 class DownloadHandlers:
@@ -40,6 +55,8 @@ class DownloadHandlers:
         self._handlers: dict[str, DownloadHandlerProtocol] = {}
         # remembers failed handlers
         self._notconfigured: dict[str, str] = {}
+        # remembers handlers with Deferred-based download_request()
+        self._old_style_handlers: set[str] = set()
         handlers: dict[str, str | Callable[..., Any]] = without_none_values(
             cast(
                 "dict[str, str | Callable[..., Any]]",
@@ -72,8 +89,17 @@ class DownloadHandlers:
         path = self._schemes[scheme]
         try:
             dhcls: type[DownloadHandlerProtocol] = load_object(path)
-            if skip_lazy and getattr(dhcls, "lazy", True):
-                return None
+            if skip_lazy:
+                if not hasattr(dhcls, "lazy"):
+                    warnings.warn(
+                        f"{global_object_name(dhcls)} doesn't define a 'lazy' attribute."
+                        f" This is deprecated, please add 'lazy = True' (which is the current"
+                        f" default value) to the class definition.",
+                        category=ScrapyDeprecationWarning,
+                        stacklevel=1,
+                    )
+                if getattr(dhcls, "lazy", True):
+                    return None
             dh = build_from_crawler(
                 dhcls,
                 self._crawler,
@@ -91,12 +117,28 @@ class DownloadHandlers:
             self._notconfigured[scheme] = str(ex)
             return None
         self._handlers[scheme] = dh
+        if not inspect.iscoroutinefunction(dh.download_request):  # pragma: no cover
+            warnings.warn(
+                f"{global_object_name(dh.download_request)} is not a coroutine function."
+                f" This is deprecated, please rewrite it to return a coroutine and remove"
+                f" the 'spider' argument.",
+                category=ScrapyDeprecationWarning,
+                stacklevel=1,
+            )
+            self._old_style_handlers.add(scheme)
         return dh
 
-    @_warn_spider_arg
     def download_request(
         self, request: Request, spider: Spider | None = None
-    ) -> Deferred[Response]:
+    ) -> Deferred[Response]:  # pragma: no cover
+        warnings.warn(
+            "DownloadHandlers.download_request() is deprecated, use download_request_async() instead",
+            category=ScrapyDeprecationWarning,
+            stacklevel=2,
+        )
+        return deferred_from_coro(self.download_request_async(request))
+
+    async def download_request_async(self, request: Request) -> Response:
         scheme = urlparse_cached(request).scheme
         handler = self._get_handler(scheme)
         if not handler:
@@ -104,11 +146,33 @@ class DownloadHandlers:
                 f"Unsupported URL scheme '{scheme}': {self._notconfigured[scheme]}"
             )
         assert self._crawler.spider
-        return handler.download_request(request, self._crawler.spider)
+        if scheme in self._old_style_handlers:  # pragma: no cover
+            return await maybe_deferred_to_future(
+                cast(
+                    "Deferred[Response]",
+                    handler.download_request(request, self._crawler.spider),  # type: ignore[call-arg]
+                )
+            )
+        return await handler.download_request(request)
 
     async def _close(self) -> None:
         for dh in self._handlers.values():
-            if not hasattr(dh, "close"):
+            if not hasattr(dh, "close"):  # pragma: no cover
+                warnings.warn(
+                    f"{global_object_name(dh)} doesn't define a close() method."
+                    f" This is deprecated, please add an empty 'async def close()' method.",
+                    category=ScrapyDeprecationWarning,
+                    stacklevel=1,
+                )
                 continue
 
-            await ensure_awaitable(dh.close())
+            if inspect.iscoroutinefunction(dh.close):
+                await dh.close()
+            else:  # pragma: no cover
+                warnings.warn(
+                    f"{global_object_name(dh.close)} is not a coroutine function."
+                    f" This is deprecated, please rewrite it to return a coroutine.",
+                    category=ScrapyDeprecationWarning,
+                    stacklevel=1,
+                )
+                await ensure_awaitable(dh.close())
diff --git a/scrapy/core/downloader/handlers/base.py b/scrapy/core/downloader/handlers/base.py
new file mode 100644
index 000000000..999c003b4
--- /dev/null
+++ b/scrapy/core/downloader/handlers/base.py
@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    # typing.Self requires Python 3.11
+    from typing_extensions import Self
+
+    from scrapy import Request
+    from scrapy.crawler import Crawler
+    from scrapy.http import Response
+
+
+class BaseDownloadHandler(ABC):
+    """Optional base class for download handlers."""
+
+    lazy: bool = False
+
+    def __init__(self, crawler: Crawler):
+        self.crawler = crawler
+
+    @classmethod
+    def from_crawler(cls, crawler: Crawler) -> Self:
+        return cls(crawler)
+
+    @abstractmethod
+    async def download_request(self, request: Request) -> Response:
+        raise NotImplementedError
+
+    async def close(self) -> None:  # noqa: B027
+        pass
diff --git a/scrapy/core/downloader/handlers/datauri.py b/scrapy/core/downloader/handlers/datauri.py
index b3f286d87..84f241d88 100644
--- a/scrapy/core/downloader/handlers/datauri.py
+++ b/scrapy/core/downloader/handlers/datauri.py
@@ -4,19 +4,16 @@ from typing import TYPE_CHECKING, Any
 
 from w3lib.url import parse_data_uri
 
+from scrapy.core.downloader.handlers.base import BaseDownloadHandler
 from scrapy.http import Response, TextResponse
 from scrapy.responsetypes import responsetypes
-from scrapy.utils.decorators import defers
 
 if TYPE_CHECKING:
-    from scrapy import Request, Spider
+    from scrapy import Request
 
 
-class DataURIDownloadHandler:
-    lazy = False
-
-    @defers
-    def download_request(self, request: Request, spider: Spider) -> Response:
+class DataURIDownloadHandler(BaseDownloadHandler):
+    async def download_request(self, request: Request) -> Response:
         uri = parse_data_uri(request.url)
         respcls = responsetypes.from_mimetype(uri.media_type)
 
diff --git a/scrapy/core/downloader/handlers/file.py b/scrapy/core/downloader/handlers/file.py
index d55c516f0..21fd2c353 100644
--- a/scrapy/core/downloader/handlers/file.py
+++ b/scrapy/core/downloader/handlers/file.py
@@ -5,19 +5,16 @@ from typing import TYPE_CHECKING
 
 from w3lib.url import file_uri_to_path
 
+from scrapy.core.downloader.handlers.base import BaseDownloadHandler
 from scrapy.responsetypes import responsetypes
-from scrapy.utils.decorators import defers
 
 if TYPE_CHECKING:
-    from scrapy import Request, Spider
+    from scrapy import Request
     from scrapy.http import Response
 
 
-class FileDownloadHandler:
-    lazy = False
-
-    @defers
-    def download_request(self, request: Request, spider: Spider) -> Response:
+class FileDownloadHandler(BaseDownloadHandler):
+    async def download_request(self, request: Request) -> Response:
         filepath = file_uri_to_path(request.url)
         body = Path(filepath).read_bytes()
         respcls = responsetypes.from_args(filename=filepath, body=body)
diff --git a/scrapy/core/downloader/handlers/ftp.py b/scrapy/core/downloader/handlers/ftp.py
index 1d947b1e3..f261e8e7f 100644
--- a/scrapy/core/downloader/handlers/ftp.py
+++ b/scrapy/core/downloader/handlers/ftp.py
@@ -33,27 +33,22 @@ from __future__ import annotations
 import re
 from io import BytesIO
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, BinaryIO
+from typing import TYPE_CHECKING, BinaryIO
 from urllib.parse import unquote
 
 from twisted.internet.protocol import ClientCreator, Protocol
 
+from scrapy.core.downloader.handlers.base import BaseDownloadHandler
 from scrapy.http import Response
 from scrapy.responsetypes import responsetypes
+from scrapy.utils.defer import maybe_deferred_to_future
 from scrapy.utils.httpobj import urlparse_cached
-from scrapy.utils.python import to_bytes
 
 if TYPE_CHECKING:
-    from twisted.internet.defer import Deferred
     from twisted.protocols.ftp import FTPClient
-    from twisted.python.failure import Failure
 
-    # typing.Self requires Python 3.11
-    from typing_extensions import Self
-
-    from scrapy import Request, Spider
+    from scrapy import Request
     from scrapy.crawler import Crawler
-    from scrapy.settings import BaseSettings
 
 
 class ReceivedDataProtocol(Protocol):
@@ -82,26 +77,21 @@ class ReceivedDataProtocol(Protocol):
 _CODE_RE = re.compile(r"\d+")
 
 
-class FTPDownloadHandler:
-    lazy = False
-
+class FTPDownloadHandler(BaseDownloadHandler):
     CODE_MAPPING: dict[str, int] = {
         "550": 404,
         "default": 503,
     }
 
-    def __init__(self, settings: BaseSettings):
-        self.default_user = settings["FTP_USER"]
-        self.default_password = settings["FTP_PASSWORD"]
-        self.passive_mode = settings["FTP_PASSIVE_MODE"]
+    def __init__(self, crawler: Crawler):
+        super().__init__(crawler)
+        self.default_user = crawler.settings["FTP_USER"]
+        self.default_password = crawler.settings["FTP_PASSWORD"]
+        self.passive_mode = crawler.settings["FTP_PASSIVE_MODE"]
 
-    @classmethod
-    def from_crawler(cls, crawler: Crawler) -> Self:
-        return cls(crawler.settings)
-
-    def download_request(self, request: Request, spider: Spider) -> Deferred[Response]:
+    async def download_request(self, request: Request) -> Response:
         from twisted.internet import reactor
-        from twisted.protocols.ftp import FTPClient
+        from twisted.protocols.ftp import CommandFailed, FTPClient
 
         parsed_url = urlparse_cached(request)
         user = request.meta.get("ftp_user", self.default_user)
@@ -112,43 +102,23 @@ class FTPDownloadHandler:
         creator = ClientCreator(
             reactor, FTPClient, user, password, passive=passive_mode
         )
-        dfd: Deferred[FTPClient] = creator.connectTCP(
-            parsed_url.hostname, parsed_url.port or 21
+        client: FTPClient = await maybe_deferred_to_future(
+            creator.connectTCP(parsed_url.hostname, parsed_url.port or 21)
         )
-        return dfd.addCallback(self.gotClient, request, unquote(parsed_url.path))
-
-    def gotClient(
-        self, client: FTPClient, request: Request, filepath: str
-    ) -> Deferred[Response]:
-        self.client = client
+        filepath = unquote(parsed_url.path)
         protocol = ReceivedDataProtocol(request.meta.get("ftp_local_filename"))
-        d = client.retrieveFile(filepath, protocol)
-        d.addCallback(self._build_response, request, protocol)
-        d.addErrback(self._failed, request)
-        return d
-
-    def _build_response(
-        self, result: Any, request: Request, protocol: ReceivedDataProtocol
-    ) -> Response:
-        self.result = result
+        try:
+            await maybe_deferred_to_future(client.retrieveFile(filepath, protocol))
+        except CommandFailed as e:
+            message = str(e)
+            if m := _CODE_RE.search(message):
+                ftpcode = m.group()
+                httpcode = self.CODE_MAPPING.get(ftpcode, self.CODE_MAPPING["default"])
+                return Response(url=request.url, status=httpcode, body=message.encode())
+            raise
         protocol.close()
         headers = {"local filename": protocol.filename or b"", "size": protocol.size}
         body = protocol.filename or protocol.body.read()
         respcls = responsetypes.from_args(url=request.url, body=body)
         # hints for Headers-related types may need to be fixed to not use AnyStr
         return respcls(url=request.url, status=200, body=body, headers=headers)  # type: ignore[arg-type]
-
-    def _failed(self, result: Failure, request: Request) -> Response:
-        from twisted.protocols.ftp import CommandFailed
-
-        message = result.getErrorMessage()
-        if result.type == CommandFailed:
-            m = _CODE_RE.search(message)
-            if m:
-                ftpcode = m.group()
-                httpcode = self.CODE_MAPPING.get(ftpcode, self.CODE_MAPPING["default"])
-                return Response(
-                    url=request.url, status=httpcode, body=to_bytes(message)
-                )
-        assert result.type
-        raise result.type(result.value)
diff --git a/scrapy/core/downloader/handlers/http10.py b/scrapy/core/downloader/handlers/http10.py
index 0fbe5fc23..af3588019 100644
--- a/scrapy/core/downloader/handlers/http10.py
+++ b/scrapy/core/downloader/handlers/http10.py
@@ -6,17 +6,17 @@ import warnings
 from typing import TYPE_CHECKING
 
 from scrapy.exceptions import ScrapyDeprecationWarning
+from scrapy.utils.defer import maybe_deferred_to_future
 from scrapy.utils.misc import build_from_crawler, load_object
 from scrapy.utils.python import to_unicode
 
 if TYPE_CHECKING:
-    from twisted.internet.defer import Deferred
     from twisted.internet.interfaces import IConnector
 
     # typing.Self requires Python 3.11
     from typing_extensions import Self
 
-    from scrapy import Request, Spider
+    from scrapy import Request
     from scrapy.core.downloader.contextfactory import ScrapyClientContextFactory
     from scrapy.core.downloader.webclient import ScrapyHTTPClientFactory
     from scrapy.crawler import Crawler
@@ -46,11 +46,10 @@ class HTTP10DownloadHandler:
     def from_crawler(cls, crawler: Crawler) -> Self:
         return cls(crawler.settings, crawler)
 
-    def download_request(self, request: Request, spider: Spider) -> Deferred[Response]:
-        """Return a deferred for the HTTP download"""
+    async def download_request(self, request: Request) -> Response:
         factory = self.HTTPClientFactory(request)
         self._connect(factory)
-        return factory.deferred
+        return await maybe_deferred_to_future(factory.deferred)
 
     def _connect(self, factory: ScrapyHTTPClientFactory) -> IConnector:
         from twisted.internet import reactor
@@ -63,3 +62,6 @@ class HTTP10DownloadHandler:
             )
             return reactor.connectSSL(host, port, factory, client_context_factory)
         return reactor.connectTCP(host, port, factory)
+
+    async def close(self) -> None:
+        pass
diff --git a/scrapy/core/downloader/handlers/http11.py b/scrapy/core/downloader/handlers/http11.py
index 02aaf7c5c..cec73566a 100644
--- a/scrapy/core/downloader/handlers/http11.py
+++ b/scrapy/core/downloader/handlers/http11.py
@@ -30,11 +30,13 @@ from twisted.web.http_headers import Headers as TxHeaders
 from twisted.web.iweb import UNKNOWN_LENGTH, IBodyProducer, IPolicyForHTTPS, IResponse
 from zope.interface import implementer
 
-from scrapy import Request, Spider, signals
+from scrapy import Request, signals
 from scrapy.core.downloader.contextfactory import load_context_factory_from_settings
+from scrapy.core.downloader.handlers.base import BaseDownloadHandler
 from scrapy.exceptions import StopDownload
 from scrapy.http import Headers, Response
 from scrapy.responsetypes import responsetypes
+from scrapy.utils.defer import maybe_deferred_to_future
 from scrapy.utils.deprecate import warn_on_deprecated_spider_attribute
 from scrapy.utils.httpobj import urlparse_cached
 from scrapy.utils.python import to_bytes, to_unicode
@@ -44,11 +46,10 @@ if TYPE_CHECKING:
     from twisted.internet.base import ReactorBase
     from twisted.internet.interfaces import IConsumer
 
-    # typing.NotRequired and typing.Self require Python 3.11
-    from typing_extensions import NotRequired, Self
+    # typing.NotRequired requires Python 3.11
+    from typing_extensions import NotRequired
 
     from scrapy.crawler import Crawler
-    from scrapy.settings import BaseSettings
 
 
 logger = logging.getLogger(__name__)
@@ -65,37 +66,34 @@ class _ResultT(TypedDict):
     failure: NotRequired[Failure | None]
 
 
-class HTTP11DownloadHandler:
-    lazy = False
-
-    def __init__(self, settings: BaseSettings, crawler: Crawler):
+class HTTP11DownloadHandler(BaseDownloadHandler):
+    def __init__(self, crawler: Crawler):
+        super().__init__(crawler)
         self._crawler = crawler
 
         from twisted.internet import reactor
 
         self._pool: HTTPConnectionPool = HTTPConnectionPool(reactor, persistent=True)
-        self._pool.maxPersistentPerHost = settings.getint(
+        self._pool.maxPersistentPerHost = crawler.settings.getint(
             "CONCURRENT_REQUESTS_PER_DOMAIN"
         )
         self._pool._factory.noisy = False
 
         self._contextFactory: IPolicyForHTTPS = load_context_factory_from_settings(
-            settings, crawler
+            crawler.settings, crawler
+        )
+        self._default_maxsize: int = crawler.settings.getint("DOWNLOAD_MAXSIZE")
+        self._default_warnsize: int = crawler.settings.getint("DOWNLOAD_WARNSIZE")
+        self._fail_on_dataloss: bool = crawler.settings.getbool(
+            "DOWNLOAD_FAIL_ON_DATALOSS"
         )
-        self._default_maxsize: int = settings.getint("DOWNLOAD_MAXSIZE")
-        self._default_warnsize: int = settings.getint("DOWNLOAD_WARNSIZE")
-        self._fail_on_dataloss: bool = settings.getbool("DOWNLOAD_FAIL_ON_DATALOSS")
         self._disconnect_timeout: int = 1
 
-    @classmethod
-    def from_crawler(cls, crawler: Crawler) -> Self:
-        return cls(crawler.settings, crawler)
-
-    def download_request(self, request: Request, spider: Spider) -> Deferred[Response]:
+    async def download_request(self, request: Request) -> Response:
         """Return a deferred for the HTTP download"""
-        if hasattr(spider, "download_maxsize"):  # pragma: no cover
+        if hasattr(self._crawler.spider, "download_maxsize"):  # pragma: no cover
             warn_on_deprecated_spider_attribute("download_maxsize", "DOWNLOAD_MAXSIZE")
-        if hasattr(spider, "download_warnsize"):  # pragma: no cover
+        if hasattr(self._crawler.spider, "download_warnsize"):  # pragma: no cover
             warn_on_deprecated_spider_attribute(
                 "download_warnsize", "DOWNLOAD_WARNSIZE"
             )
@@ -103,14 +101,18 @@ class HTTP11DownloadHandler:
         agent = ScrapyAgent(
             contextFactory=self._contextFactory,
             pool=self._pool,
-            maxsize=getattr(spider, "download_maxsize", self._default_maxsize),
-            warnsize=getattr(spider, "download_warnsize", self._default_warnsize),
+            maxsize=getattr(
+                self._crawler.spider, "download_maxsize", self._default_maxsize
+            ),
+            warnsize=getattr(
+                self._crawler.spider, "download_warnsize", self._default_warnsize
+            ),
             fail_on_dataloss=self._fail_on_dataloss,
             crawler=self._crawler,
         )
-        return agent.download_request(request)
+        return await maybe_deferred_to_future(agent.download_request(request))
 
-    def close(self) -> Deferred[None]:
+    async def close(self) -> None:
         from twisted.internet import reactor
 
         d: Deferred[None] = self._pool.closeCachedConnections()
@@ -118,19 +120,19 @@ class HTTP11DownloadHandler:
         # we'll manually timeout the deferred.
         #
         # Twisted issue addressing this problem can be found here:
-        # https://twistedmatrix.com/trac/ticket/7738.
+        # https://github.com/twisted/twisted/issues/7738
         #
         # closeCachedConnections doesn't handle external errbacks, so we'll
         # issue a callback after `_disconnect_timeout` seconds.
+        #
+        # See also https://github.com/scrapy/scrapy/issues/2653
         delayed_call = reactor.callLater(self._disconnect_timeout, d.callback, [])
 
-        def cancel_delayed_call(result: _T) -> _T:
+        try:
+            await maybe_deferred_to_future(d)
+        finally:
             if delayed_call.active():
                 delayed_call.cancel()
-            return result
-
-        d.addBoth(cancel_delayed_call)
-        return d
 
 
 class TunnelError(Exception):
diff --git a/scrapy/core/downloader/handlers/http2.py b/scrapy/core/downloader/handlers/http2.py
index 8e6236094..ff6e8ab98 100644
--- a/scrapy/core/downloader/handlers/http2.py
+++ b/scrapy/core/downloader/handlers/http2.py
@@ -8,7 +8,9 @@ from twisted.internet.error import TimeoutError as TxTimeoutError
 from twisted.web.client import URI
 
 from scrapy.core.downloader.contextfactory import load_context_factory_from_settings
+from scrapy.core.downloader.handlers.base import BaseDownloadHandler
 from scrapy.core.http2.agent import H2Agent, H2ConnectionPool, ScrapyProxyH2Agent
+from scrapy.utils.defer import maybe_deferred_to_future
 from scrapy.utils.httpobj import urlparse_cached
 from scrapy.utils.python import to_bytes
 
@@ -17,37 +19,37 @@ if TYPE_CHECKING:
     from twisted.internet.defer import Deferred
     from twisted.web.iweb import IPolicyForHTTPS
 
-    # typing.Self requires Python 3.11
-    from typing_extensions import Self
-
     from scrapy.crawler import Crawler
     from scrapy.http import Request, Response
-    from scrapy.settings import Settings
     from scrapy.spiders import Spider
 
 
-class H2DownloadHandler:
-    def __init__(self, settings: Settings, crawler: Crawler):
+class H2DownloadHandler(BaseDownloadHandler):
+    lazy = True
+
+    def __init__(self, crawler: Crawler):
+        super().__init__(crawler)
         self._crawler = crawler
 
         from twisted.internet import reactor
 
-        self._pool = H2ConnectionPool(reactor, settings)
-        self._context_factory = load_context_factory_from_settings(settings, crawler)
+        self._pool = H2ConnectionPool(reactor, crawler.settings)
+        self._context_factory = load_context_factory_from_settings(
+            crawler.settings, crawler
+        )
 
-    @classmethod
-    def from_crawler(cls, crawler: Crawler) -> Self:
-        return cls(crawler.settings, crawler)
-
-    def download_request(self, request: Request, spider: Spider) -> Deferred[Response]:
+    async def download_request(self, request: Request) -> Response:
         agent = ScrapyH2Agent(
             context_factory=self._context_factory,
             pool=self._pool,
             crawler=self._crawler,
         )
-        return agent.download_request(request, spider)
+        assert self._crawler.spider
+        return await maybe_deferred_to_future(
+            agent.download_request(request, self._crawler.spider)
+        )
 
-    def close(self) -> None:
+    async def close(self) -> None:
         self._pool.close_connections()
 
 
diff --git a/scrapy/core/downloader/handlers/s3.py b/scrapy/core/downloader/handlers/s3.py
index 05a71b742..a601bc2aa 100644
--- a/scrapy/core/downloader/handlers/s3.py
+++ b/scrapy/core/downloader/handlers/s3.py
@@ -1,7 +1,8 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
+from scrapy.core.downloader.handlers.base import BaseDownloadHandler
 from scrapy.core.downloader.handlers.http11 import HTTP11DownloadHandler
 from scrapy.exceptions import NotConfigured
 from scrapy.utils.boto import is_botocore_available
@@ -9,57 +10,28 @@ from scrapy.utils.httpobj import urlparse_cached
 from scrapy.utils.misc import build_from_crawler
 
 if TYPE_CHECKING:
-    from twisted.internet.defer import Deferred
-
-    # typing.Self requires Python 3.11
-    from typing_extensions import Self
-
-    from scrapy import Request, Spider
+    from scrapy import Request
     from scrapy.crawler import Crawler
     from scrapy.http import Response
-    from scrapy.settings import BaseSettings
 
 
-class S3DownloadHandler:
-    def __init__(
-        self,
-        settings: BaseSettings,
-        *,
-        crawler: Crawler,
-        aws_access_key_id: str | None = None,
-        aws_secret_access_key: str | None = None,
-        aws_session_token: str | None = None,
-        httpdownloadhandler: type[HTTP11DownloadHandler] = HTTP11DownloadHandler,
-        **kw: Any,
-    ):
+class S3DownloadHandler(BaseDownloadHandler):
+    lazy = True
+
+    def __init__(self, crawler: Crawler):
         if not is_botocore_available():
             raise NotConfigured("missing botocore library")
 
-        if not aws_access_key_id:
-            aws_access_key_id = settings["AWS_ACCESS_KEY_ID"]
-        if not aws_secret_access_key:
-            aws_secret_access_key = settings["AWS_SECRET_ACCESS_KEY"]
-        if not aws_session_token:
-            aws_session_token = settings["AWS_SESSION_TOKEN"]
-
-        # If no credentials could be found anywhere,
-        # consider this an anonymous connection request by default;
-        # unless 'anon' was set explicitly (True/False).
-        anon = kw.get("anon")
-        if anon is None and not aws_access_key_id and not aws_secret_access_key:
-            kw["anon"] = True
-        self.anon = kw.get("anon")
-
+        super().__init__(crawler)
+        aws_access_key_id = crawler.settings["AWS_ACCESS_KEY_ID"]
+        aws_secret_access_key = crawler.settings["AWS_SECRET_ACCESS_KEY"]
+        aws_session_token = crawler.settings["AWS_SESSION_TOKEN"]
+        self.anon = not aws_access_key_id and not aws_secret_access_key
         self._signer = None
-        import botocore.auth  # noqa: PLC0415
-        import botocore.credentials  # noqa: PLC0415
-
-        kw.pop("anon", None)
-        if kw:
-            raise TypeError(f"Unexpected keyword arguments: {kw}")
         if not self.anon:
-            assert aws_access_key_id is not None
-            assert aws_secret_access_key is not None
+            import botocore.auth  # noqa: PLC0415
+            import botocore.credentials  # noqa: PLC0415
+
             SignerCls = botocore.auth.AUTH_TYPE_MAPS["s3"]
             # botocore.auth.BaseSigner doesn't have an __init__() with args, only subclasses do
             self._signer = SignerCls(  # type: ignore[call-arg]
@@ -68,17 +40,10 @@ class S3DownloadHandler:
                 )
             )
 
-        _http_handler = build_from_crawler(
-            httpdownloadhandler,
-            crawler,
-        )
+        _http_handler = build_from_crawler(HTTP11DownloadHandler, crawler)
         self._download_http = _http_handler.download_request
 
-    @classmethod
-    def from_crawler(cls, crawler: Crawler, **kwargs: Any) -> Self:
-        return cls(crawler.settings, crawler=crawler, **kwargs)
-
-    def download_request(self, request: Request, spider: Spider) -> Deferred[Response]:
+    async def download_request(self, request: Request) -> Response:
         p = urlparse_cached(request)
         scheme = "https" if request.meta.get("is_secure") else "http"
         bucket = p.hostname
@@ -98,4 +63,4 @@ class S3DownloadHandler:
             assert self._signer
             self._signer.add_auth(awsrequest)
             request = request.replace(url=url, headers=awsrequest.headers.items())
-        return self._download_http(request, spider)
+        return await self._download_http(request)
diff --git a/scrapy/extensions/telnet.py b/scrapy/extensions/telnet.py
index 094a0195e..d24c1b6c4 100644
--- a/scrapy/extensions/telnet.py
+++ b/scrapy/extensions/telnet.py
@@ -15,10 +15,10 @@ from typing import TYPE_CHECKING, Any
 from twisted.conch import telnet
 from twisted.conch.insults import insults
 from twisted.internet import protocol
+from twisted.internet.defer import fail, succeed
 
 from scrapy import signals
 from scrapy.exceptions import NotConfigured
-from scrapy.utils.decorators import defers
 from scrapy.utils.engine import print_engine_status
 from scrapy.utils.reactor import listen_tcp
 from scrapy.utils.trackref import print_live_refs
@@ -80,20 +80,19 @@ class TelnetConsole(protocol.ServerFactory):
         class Portal:
             """An implementation of IPortal"""
 
-            @defers
             def login(self_, credentials, mind, *interfaces):  # pylint: disable=no-self-argument
                 if not (
                     credentials.username == self.username.encode("utf8")
                     and credentials.checkPassword(self.password.encode("utf8"))
                 ):
-                    raise ValueError("Invalid credentials")
+                    return fail(ValueError("Invalid credentials"))
 
                 from twisted.conch import manhole
 
                 protocol = telnet.TelnetBootstrapProtocol(
                     insults.ServerProtocol, manhole.Manhole, self._get_telnet_vars()
                 )
-                return (interfaces[0], protocol, lambda: None)
+                return succeed((interfaces[0], protocol, lambda: None))
 
         return telnet.TelnetTransport(telnet.AuthenticatingTelnetProtocol, Portal())
 
diff --git a/scrapy/utils/decorators.py b/scrapy/utils/decorators.py
index 2dde0aa55..044d1d4c7 100644
--- a/scrapy/utils/decorators.py
+++ b/scrapy/utils/decorators.py
@@ -42,8 +42,13 @@ def deprecated(
     return deco
 
 
-def defers(func: Callable[_P, _T]) -> Callable[_P, Deferred[_T]]:
+def defers(func: Callable[_P, _T]) -> Callable[_P, Deferred[_T]]:  # pragma: no cover
     """Decorator to make sure a function always returns a deferred"""
+    warnings.warn(
+        "@defers is deprecated, you can use maybeDeferred() directly if needed.",
+        category=ScrapyDeprecationWarning,
+        stacklevel=2,
+    )
 
     @wraps(func)
     def wrapped(*a: _P.args, **kw: _P.kwargs) -> Deferred[_T]:
diff --git a/tests/test_crawl.py b/tests/test_crawl.py
index 699ac3911..9fd5a40df 100644
--- a/tests/test_crawl.py
+++ b/tests/test_crawl.py
@@ -420,6 +420,12 @@ with multiples lines
         self._assert_retried(caplog.text)
         assert "Got response 200" in caplog.text
 
+    @deferred_f_from_coro_f
+    async def test_unknown_url_scheme(self, caplog: pytest.LogCaptureFixture) -> None:
+        crawler = get_crawler(SimpleSpider)
+        await maybe_deferred_to_future(crawler.crawl("foo://bar"))
+        assert "NotSupported: Unsupported URL scheme 'foo'" in caplog.text
+
 
 class TestCrawlSpider:
     mockserver: MockServer
diff --git a/tests/test_downloader_handler_twisted_ftp.py b/tests/test_downloader_handler_twisted_ftp.py
index a43fe318d..6cda607cc 100644
--- a/tests/test_downloader_handler_twisted_ftp.py
+++ b/tests/test_downloader_handler_twisted_ftp.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import os
 import sys
+from abc import ABC, abstractmethod
 from pathlib import Path
 from tempfile import mkstemp
 from typing import TYPE_CHECKING, Any
@@ -13,17 +14,18 @@ from twisted.cred import checkers, credentials, portal
 from scrapy.core.downloader.handlers.ftp import FTPDownloadHandler
 from scrapy.http import HtmlResponse, Request, Response
 from scrapy.http.response.text import TextResponse
-from scrapy.utils.defer import deferred_f_from_coro_f, maybe_deferred_to_future
+from scrapy.utils.defer import deferred_f_from_coro_f
 from scrapy.utils.misc import build_from_crawler
 from scrapy.utils.python import to_bytes
-from scrapy.utils.spider import DefaultSpider
 from scrapy.utils.test import get_crawler
 
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator, Generator
 
+    from twisted.protocols.ftp import FTPFactory
 
-class TestFTPBase:
+
+class TestFTPBase(ABC):
     username = "scrapy"
     password = "passwd"
     req_meta: dict[str, Any] = {"ftp_user": username, "ftp_password": password}
@@ -34,21 +36,13 @@ class TestFTPBase:
         ("html-file-without-extension", b"<!DOCTYPE html>\n<title>.</title>"),
     )
 
+    @abstractmethod
     def _create_files(self, root: Path) -> None:
-        userdir = root / self.username
-        userdir.mkdir()
-        for filename, content in self.test_files:
-            (userdir / filename).write_bytes(content)
+        raise NotImplementedError
 
-    def _get_factory(self, root):
-        from twisted.protocols.ftp import FTPFactory, FTPRealm
-
-        realm = FTPRealm(anonymousRoot=str(root), userHome=str(root))
-        p = portal.Portal(realm)
-        users_checker = checkers.InMemoryUsernamePasswordDatabaseDontUse()
-        users_checker.addUser(self.username, self.password)
-        p.registerChecker(users_checker, credentials.IUsernamePassword)
-        return FTPFactory(portal=p)
+    @abstractmethod
+    def _get_factory(self, tmp_path: Path) -> FTPFactory:
+        raise NotImplementedError
 
     @async_yield_fixture
     async def server_url(self, tmp_path: Path) -> AsyncGenerator[str]:
@@ -76,18 +70,12 @@ class TestFTPBase:
             assert dh.client.transport
             dh.client.transport.loseConnection()
 
-    @staticmethod
-    async def download_request(dh: FTPDownloadHandler, request: Request) -> Response:
-        return await maybe_deferred_to_future(
-            dh.download_request(request, DefaultSpider())
-        )
-
     @deferred_f_from_coro_f
     async def test_ftp_download_success(
         self, server_url: str, dh: FTPDownloadHandler
     ) -> None:
         request = Request(url=server_url + "file.txt", meta=self.req_meta)
-        r = await self.download_request(dh, request)
+        r = await dh.download_request(request)
         assert r.status == 200
         assert r.body == b"I have the power!"
         assert r.headers == {b"Local Filename": [b""], b"Size": [b"17"]}
@@ -101,7 +89,7 @@ class TestFTPBase:
             url=server_url + "file with spaces.txt",
             meta=self.req_meta,
         )
-        r = await self.download_request(dh, request)
+        r = await dh.download_request(request)
         assert r.status == 200
         assert r.body == b"Moooooooooo power!"
         assert r.headers == {b"Local Filename": [b""], b"Size": [b"18"]}
@@ -111,8 +99,9 @@ class TestFTPBase:
         self, server_url: str, dh: FTPDownloadHandler
     ) -> None:
         request = Request(url=server_url + "nonexistent.txt", meta=self.req_meta)
-        r = await self.download_request(dh, request)
+        r = await dh.download_request(request)
         assert r.status == 404
+        assert r.body == b"['550 nonexistent.txt: No such file or directory.']"
 
     @deferred_f_from_coro_f
     async def test_ftp_local_filename(
@@ -125,7 +114,7 @@ class TestFTPBase:
         meta = {"ftp_local_filename": fname_bytes}
         meta.update(self.req_meta)
         request = Request(url=server_url + "file.txt", meta=meta)
-        r = await self.download_request(dh, request)
+        r = await dh.download_request(request)
         assert r.body == fname_bytes
         assert r.headers == {b"Local Filename": [fname_bytes], b"Size": [b"17"]}
         assert local_path.exists()
@@ -153,12 +142,28 @@ class TestFTPBase:
         meta = {}
         meta.update(self.req_meta)
         request = Request(url=server_url + filename, meta=meta)
-        r = await self.download_request(dh, request)
+        r = await dh.download_request(request)
         assert type(r) is response_class  # pylint: disable=unidiomatic-typecheck
         local_fname_path.unlink()
 
 
 class TestFTP(TestFTPBase):
+    def _create_files(self, root: Path) -> None:
+        userdir = root / self.username
+        userdir.mkdir()
+        for filename, content in self.test_files:
+            (userdir / filename).write_bytes(content)
+
+    def _get_factory(self, root):
+        from twisted.protocols.ftp import FTPFactory, FTPRealm
+
+        realm = FTPRealm(anonymousRoot=str(root), userHome=str(root))
+        p = portal.Portal(realm)
+        users_checker = checkers.InMemoryUsernamePasswordDatabaseDontUse()
+        users_checker.addUser(self.username, self.password)
+        p.registerChecker(users_checker, credentials.IUsernamePassword)
+        return FTPFactory(portal=p)
+
     @deferred_f_from_coro_f
     async def test_invalid_credentials(
         self, server_url: str, dh: FTPDownloadHandler, reactor_pytest: str
@@ -174,7 +179,7 @@ class TestFTP(TestFTPBase):
         meta.update({"ftp_password": "invalid"})
         request = Request(url=server_url + "file.txt", meta=meta)
         with pytest.raises(ConnectionLost):
-            await self.download_request(dh, request)
+            await dh.download_request(request)
 
 
 class TestAnonymousFTP(TestFTPBase):
diff --git a/tests/test_downloader_handler_twisted_http10.py b/tests/test_downloader_handler_twisted_http10.py
index c3e0aa7c4..4745869e4 100644
--- a/tests/test_downloader_handler_twisted_http10.py
+++ b/tests/test_downloader_handler_twisted_http10.py
@@ -9,11 +9,7 @@ import pytest
 from scrapy.core.downloader.handlers.http10 import HTTP10DownloadHandler
 from scrapy.http import Request
 from scrapy.utils.defer import deferred_f_from_coro_f
-from tests.test_downloader_handlers_http_base import (
-    TestHttpBase,
-    TestHttpProxyBase,
-    download_request,
-)
+from tests.test_downloader_handlers_http_base import TestHttpBase, TestHttpProxyBase
 
 if TYPE_CHECKING:
     from scrapy.core.downloader.handlers import DownloadHandlerProtocol
@@ -31,13 +27,12 @@ class TestHttp10(HTTP10DownloadHandlerMixin, TestHttpBase):
     """HTTP 1.0 test case"""
 
     @deferred_f_from_coro_f
-    async def test_protocol(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_protocol(self, mockserver: MockServer) -> None:
         request = Request(
             mockserver.url("/host", is_secure=self.is_secure), method="GET"
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.protocol == "HTTP/1.0"
 
 
diff --git a/tests/test_downloader_handler_twisted_http2.py b/tests/test_downloader_handler_twisted_http2.py
index 1e638392f..f60d5e7b5 100644
--- a/tests/test_downloader_handler_twisted_http2.py
+++ b/tests/test_downloader_handler_twisted_http2.py
@@ -13,10 +13,7 @@ from twisted.web.error import SchemeNotSupported
 from twisted.web.http import H2_ENABLED
 
 from scrapy.http import Request
-from scrapy.spiders import Spider
 from scrapy.utils.defer import deferred_f_from_coro_f, maybe_deferred_to_future
-from scrapy.utils.misc import build_from_crawler
-from scrapy.utils.test import get_crawler
 from tests.test_downloader_handlers_http_base import (
     TestHttpProxyBase,
     TestHttps11Base,
@@ -25,7 +22,6 @@ from tests.test_downloader_handlers_http_base import (
     TestHttpsInvalidDNSPatternBase,
     TestHttpsWrongHostnameBase,
     TestHttpWithCrawlerBase,
-    download_request,
 )
 
 if TYPE_CHECKING:
@@ -53,13 +49,12 @@ class TestHttps2(H2DownloadHandlerMixin, TestHttps11Base):
     HTTP2_DATALOSS_SKIP_REASON = "Content-Length mismatch raises InvalidBodyLengthError"
 
     @deferred_f_from_coro_f
-    async def test_protocol(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_protocol(self, mockserver: MockServer) -> None:
         request = Request(
             mockserver.url("/host", is_secure=self.is_secure), method="GET"
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.protocol == "h2"
 
     @deferred_f_from_coro_f
@@ -68,9 +63,6 @@ class TestHttps2(H2DownloadHandlerMixin, TestHttps11Base):
     ) -> None:
         from twisted.internet import reactor
 
-        crawler = get_crawler(settings_dict={"DOWNLOAD_MAXSIZE": 1_500})
-        download_handler = build_from_crawler(self.download_handler_cls, crawler)
-
         with mock.patch("scrapy.core.http2.stream.logger") as logger:
             request = Request(
                 mockserver.url("/largechunkedfile", is_secure=self.is_secure)
@@ -79,8 +71,9 @@ class TestHttps2(H2DownloadHandlerMixin, TestHttps11Base):
             def check(logger: mock.Mock) -> None:
                 logger.error.assert_called_once_with(mock.ANY)
 
-            with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
-                await download_request(download_handler, request, Spider("foo"))
+            async with self.get_dh({"DOWNLOAD_MAXSIZE": 1_500}) as download_handler:
+                with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
+                    await download_handler.download_request(request)
 
             # As the error message is logged in the dataReceived callback, we
             # have to give a bit of time to the reactor to process the queue
@@ -91,12 +84,11 @@ class TestHttps2(H2DownloadHandlerMixin, TestHttps11Base):
             await maybe_deferred_to_future(d)
 
     @deferred_f_from_coro_f
-    async def test_unsupported_scheme(
-        self, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_unsupported_scheme(self) -> None:
         request = Request("ftp://unsupported.scheme")
-        with pytest.raises(SchemeNotSupported):
-            await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            with pytest.raises(SchemeNotSupported):
+                await download_handler.download_request(request)
 
     def test_download_cause_data_loss(self) -> None:  # type: ignore[override]
         pytest.skip(self.HTTP2_DATALOSS_SKIP_REASON)
@@ -109,49 +101,46 @@ class TestHttps2(H2DownloadHandlerMixin, TestHttps11Base):
 
     @deferred_f_from_coro_f
     async def test_concurrent_requests_same_domain(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
+        self, mockserver: MockServer
     ) -> None:
         request1 = Request(mockserver.url("/text", is_secure=self.is_secure))
-        response1 = await download_request(download_handler, request1)
-        assert response1.body == b"Works"
-
         request2 = Request(
             mockserver.url("/echo", is_secure=self.is_secure), method="POST"
         )
-        response2 = await download_request(download_handler, request2)
-        assert response2.headers["Content-Length"] == b"79"
+        async with self.get_dh() as download_handler:
+            response1 = await download_handler.download_request(request1)
+            assert response1.body == b"Works"
+            response2 = await download_handler.download_request(request2)
+            assert response2.headers["Content-Length"] == b"79"
 
     @pytest.mark.xfail(reason="https://github.com/python-hyper/h2/issues/1247")
     @deferred_f_from_coro_f
-    async def test_connect_request(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_connect_request(self, mockserver: MockServer) -> None:
         request = Request(
             mockserver.url("/file", is_secure=self.is_secure), method="CONNECT"
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.body == b""
 
     @deferred_f_from_coro_f
-    async def test_custom_content_length_good(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_custom_content_length_good(self, mockserver: MockServer) -> None:
         request = Request(mockserver.url("/contentlength", is_secure=self.is_secure))
         custom_content_length = str(len(request.body))
         request.headers["Content-Length"] = custom_content_length
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.text == custom_content_length
 
     @deferred_f_from_coro_f
-    async def test_custom_content_length_bad(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_custom_content_length_bad(self, mockserver: MockServer) -> None:
         request = Request(mockserver.url("/contentlength", is_secure=self.is_secure))
         actual_content_length = str(len(request.body))
         bad_content_length = str(len(request.body) + 1)
         request.headers["Content-Length"] = bad_content_length
-        with LogCapture() as log:
-            response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            with LogCapture() as log:
+                response = await download_handler.download_request(request)
         assert response.text == actual_content_length
         log.check_present(
             (
@@ -164,14 +153,13 @@ class TestHttps2(H2DownloadHandlerMixin, TestHttps11Base):
         )
 
     @deferred_f_from_coro_f
-    async def test_duplicate_header(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_duplicate_header(self, mockserver: MockServer) -> None:
         request = Request(mockserver.url("/echo", is_secure=self.is_secure))
         header, value1, value2 = "Custom-Header", "foo", "bar"
         request.headers.appendlist(header, value1)
         request.headers.appendlist(header, value2)
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert json.loads(response.text)["headers"][header] == [value1, value2]
 
 
@@ -213,26 +201,18 @@ class TestHttps2Proxy(H2DownloadHandlerMixin, TestHttpProxyBase):
 
     @deferred_f_from_coro_f
     async def test_download_with_proxy_https_timeout(
-        self,
-        proxy_mockserver: ProxyEchoMockServer,
-        download_handler: DownloadHandlerProtocol,
+        self, proxy_mockserver: ProxyEchoMockServer
     ) -> None:
         with pytest.raises(NotImplementedError):
             await maybe_deferred_to_future(
-                super().test_download_with_proxy_https_timeout(
-                    proxy_mockserver, download_handler
-                )
+                super().test_download_with_proxy_https_timeout(proxy_mockserver)
             )
 
     @deferred_f_from_coro_f
     async def test_download_with_proxy_without_http_scheme(
-        self,
-        proxy_mockserver: ProxyEchoMockServer,
-        download_handler: DownloadHandlerProtocol,
+        self, proxy_mockserver: ProxyEchoMockServer
     ) -> None:
         with pytest.raises(SchemeNotSupported):
             await maybe_deferred_to_future(
-                super().test_download_with_proxy_without_http_scheme(
-                    proxy_mockserver, download_handler
-                )
+                super().test_download_with_proxy_without_http_scheme(proxy_mockserver)
             )
diff --git a/tests/test_downloader_handlers.py b/tests/test_downloader_handlers.py
index e37932fbb..602b2186f 100644
--- a/tests/test_downloader_handlers.py
+++ b/tests/test_downloader_handlers.py
@@ -15,22 +15,26 @@ from scrapy.core.downloader.handlers import DownloadHandlers
 from scrapy.core.downloader.handlers.datauri import DataURIDownloadHandler
 from scrapy.core.downloader.handlers.file import FileDownloadHandler
 from scrapy.core.downloader.handlers.s3 import S3DownloadHandler
-from scrapy.exceptions import NotConfigured
-from scrapy.http import Request, Response
+from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
+from scrapy.http import Request
 from scrapy.responsetypes import responsetypes
-from scrapy.utils.defer import deferred_f_from_coro_f, maybe_deferred_to_future
+from scrapy.utils.boto import is_botocore_available
+from scrapy.utils.defer import deferred_f_from_coro_f
 from scrapy.utils.misc import build_from_crawler
-from scrapy.utils.spider import DefaultSpider
 from scrapy.utils.test import get_crawler
 
 
 class DummyDH:
     lazy = False
 
+    async def download_request(self, request):
+        pass
+
 
 class DummyLazyDH:
-    # Default is lazy for backward compatibility
-    pass
+    # Default (but deprecated) is lazy for backward compatibility
+    async def download_request(self, request):
+        pass
 
 
 class OffDH:
@@ -44,6 +48,17 @@ class OffDH:
         return cls(crawler)
 
 
+class BuggyDH:
+    lazy = False
+
+    def __init__(self, crawler):
+        raise ValueError
+
+    @classmethod
+    def from_crawler(cls, crawler):
+        return cls(crawler)
+
+
 class TestLoad:
     def test_enabled_handler(self):
         handlers = {"scheme": DummyDH}
@@ -61,6 +76,18 @@ class TestLoad:
         assert "scheme" not in dh._handlers
         assert "scheme" in dh._notconfigured
 
+    def test_buggy_handler(self, caplog: pytest.LogCaptureFixture) -> None:
+        handlers = {"scheme": BuggyDH}
+        crawler = get_crawler(settings_dict={"DOWNLOAD_HANDLERS": handlers})
+        dh = DownloadHandlers(crawler)
+        assert "scheme" in dh._schemes
+        assert "scheme" not in dh._handlers
+        assert "scheme" in dh._notconfigured
+        assert (
+            'Loading "<class \'tests.test_downloader_handlers.BuggyDH\'>" for scheme "scheme"'
+            in caplog.text
+        )
+
     def test_disabled_handler(self):
         handlers = {"scheme": None}
         crawler = get_crawler(settings_dict={"DOWNLOAD_HANDLERS": handlers})
@@ -74,7 +101,11 @@ class TestLoad:
     def test_lazy_handlers(self):
         handlers = {"scheme": DummyLazyDH}
         crawler = get_crawler(settings_dict={"DOWNLOAD_HANDLERS": handlers})
-        dh = DownloadHandlers(crawler)
+        with pytest.warns(
+            ScrapyDeprecationWarning,
+            match="DummyLazyDH doesn't define a 'lazy' attribute",
+        ):
+            dh = DownloadHandlers(crawler)
         assert "scheme" in dh._schemes
         assert "scheme" not in dh._handlers
         for scheme in handlers:  # force load lazy handler
@@ -88,17 +119,13 @@ class TestFile:
         # add a special char to check that they are handled correctly
         self.fd, self.tmpname = mkstemp(suffix="^")
         Path(self.tmpname).write_text("0123456789", encoding="utf-8")
-        self.download_handler = build_from_crawler(FileDownloadHandler, get_crawler())
+        download_handler = build_from_crawler(FileDownloadHandler, get_crawler())
+        self.download_request = download_handler.download_request
 
     def teardown_method(self):
         os.close(self.fd)
         Path(self.tmpname).unlink()
 
-    async def download_request(self, request: Request) -> Response:
-        return await maybe_deferred_to_future(
-            self.download_handler.download_request(request, DefaultSpider())
-        )
-
     @deferred_f_from_coro_f
     async def test_download(self):
         request = Request(path_to_file_uri(self.tmpname))
@@ -121,7 +148,7 @@ class HttpDownloadHandlerMock:
     def __init__(self, *args, **kwargs):
         pass
 
-    def download_request(self, request, spider):
+    async def download_request(self, request):
         return request
 
 
@@ -129,18 +156,17 @@ class HttpDownloadHandlerMock:
 class TestS3Anon:
     def setup_method(self):
         crawler = get_crawler()
-        self.s3reqh = build_from_crawler(
-            S3DownloadHandler,
-            crawler,
-            httpdownloadhandler=HttpDownloadHandlerMock,
-            # anon=True, # implicit
-        )
+        with mock.patch(
+            "scrapy.core.downloader.handlers.s3.HTTP11DownloadHandler",
+            HttpDownloadHandlerMock,
+        ):
+            self.s3reqh = build_from_crawler(S3DownloadHandler, crawler)
         self.download_request = self.s3reqh.download_request
-        self.spider = DefaultSpider()
 
-    def test_anon_request(self):
+    @deferred_f_from_coro_f
+    async def test_anon_request(self):
         req = Request("s3://aws-publicdatasets/")
-        httpreq = self.download_request(req, self.spider)
+        httpreq = await self.download_request(req)
         assert hasattr(self.s3reqh, "anon")
         assert self.s3reqh.anon
         assert httpreq.url == "http://aws-publicdatasets.s3.amazonaws.com/"
@@ -148,26 +174,22 @@ class TestS3Anon:
 
 @pytest.mark.requires_botocore
 class TestS3:
-    download_handler_cls: type = S3DownloadHandler
-
-    # test use same example keys than amazon developer guide
-    # http://s3.amazonaws.com/awsdocs/S3/20060301/s3-dg-20060301.pdf
-    # and the tests described here are the examples from that manual
-
-    AWS_ACCESS_KEY_ID = "0PN5J17HBGZHT7JJ3X82"
-    AWS_SECRET_ACCESS_KEY = "uV3F3YluFJax1cknvbcGwgjvx4QpvB+leU8dUj2o"
-
     def setup_method(self):
-        crawler = get_crawler()
-        s3reqh = build_from_crawler(
-            S3DownloadHandler,
-            crawler,
-            aws_access_key_id=self.AWS_ACCESS_KEY_ID,
-            aws_secret_access_key=self.AWS_SECRET_ACCESS_KEY,
-            httpdownloadhandler=HttpDownloadHandlerMock,
+        # test use same example keys than amazon developer guide
+        # http://s3.amazonaws.com/awsdocs/S3/20060301/s3-dg-20060301.pdf
+        # and the tests described here are the examples from that manual
+        crawler = get_crawler(
+            settings_dict={
+                "AWS_ACCESS_KEY_ID": "0PN5J17HBGZHT7JJ3X82",
+                "AWS_SECRET_ACCESS_KEY": "uV3F3YluFJax1cknvbcGwgjvx4QpvB+leU8dUj2o",
+            }
         )
+        with mock.patch(
+            "scrapy.core.downloader.handlers.s3.HTTP11DownloadHandler",
+            HttpDownloadHandlerMock,
+        ):
+            s3reqh = build_from_crawler(S3DownloadHandler, crawler)
         self.download_request = s3reqh.download_request
-        self.spider = DefaultSpider()
 
     @contextlib.contextmanager
     def _mocked_date(self, date):
@@ -183,27 +205,20 @@ class TestS3:
                 mock_formatdate.return_value = date
                 yield
 
-    def test_extra_kw(self):
-        crawler = get_crawler()
-        with pytest.raises((TypeError, NotConfigured)):
-            build_from_crawler(
-                S3DownloadHandler,
-                crawler,
-                extra_kw=True,
-            )
-
-    def test_request_signing1(self):
+    @deferred_f_from_coro_f
+    async def test_request_signing1(self):
         # gets an object from the johnsmith bucket.
         date = "Tue, 27 Mar 2007 19:36:42 +0000"
         req = Request("s3://johnsmith/photos/puppy.jpg", headers={"Date": date})
         with self._mocked_date(date):
-            httpreq = self.download_request(req, self.spider)
+            httpreq = await self.download_request(req)
         assert (
             httpreq.headers["Authorization"]
             == b"AWS 0PN5J17HBGZHT7JJ3X82:xXjDGYUmKxnwqr5KXNPGldn5LbA="
         )
 
-    def test_request_signing2(self):
+    @deferred_f_from_coro_f
+    async def test_request_signing2(self):
         # puts an object into the johnsmith bucket.
         date = "Tue, 27 Mar 2007 21:15:45 +0000"
         req = Request(
@@ -216,13 +231,14 @@ class TestS3:
             },
         )
         with self._mocked_date(date):
-            httpreq = self.download_request(req, self.spider)
+            httpreq = await self.download_request(req)
         assert (
             httpreq.headers["Authorization"]
             == b"AWS 0PN5J17HBGZHT7JJ3X82:hcicpDDvL9SsO6AkvxqmIWkmOuQ="
         )
 
-    def test_request_signing3(self):
+    @deferred_f_from_coro_f
+    async def test_request_signing3(self):
         # lists the content of the johnsmith bucket.
         date = "Tue, 27 Mar 2007 19:42:41 +0000"
         req = Request(
@@ -234,24 +250,26 @@ class TestS3:
             },
         )
         with self._mocked_date(date):
-            httpreq = self.download_request(req, self.spider)
+            httpreq = await self.download_request(req)
         assert (
             httpreq.headers["Authorization"]
             == b"AWS 0PN5J17HBGZHT7JJ3X82:jsRt/rhG+Vtp88HrYL706QhE4w4="
         )
 
-    def test_request_signing4(self):
+    @deferred_f_from_coro_f
+    async def test_request_signing4(self):
         # fetches the access control policy sub-resource for the 'johnsmith' bucket.
         date = "Tue, 27 Mar 2007 19:44:46 +0000"
         req = Request("s3://johnsmith/?acl", method="GET", headers={"Date": date})
         with self._mocked_date(date):
-            httpreq = self.download_request(req, self.spider)
+            httpreq = await self.download_request(req)
         assert (
             httpreq.headers["Authorization"]
             == b"AWS 0PN5J17HBGZHT7JJ3X82:thdUi9VAkzhkniLj96JIrOPGi0g="
         )
 
-    def test_request_signing6(self):
+    @deferred_f_from_coro_f
+    async def test_request_signing6(self):
         # uploads an object to a CNAME style virtual hosted bucket with metadata.
         date = "Tue, 27 Mar 2007 21:06:08 +0000"
         req = Request(
@@ -273,13 +291,14 @@ class TestS3:
             },
         )
         with self._mocked_date(date):
-            httpreq = self.download_request(req, self.spider)
+            httpreq = await self.download_request(req)
         assert (
             httpreq.headers["Authorization"]
             == b"AWS 0PN5J17HBGZHT7JJ3X82:C0FlOtU8Ylb9KDTpZqYkZPX91iI="
         )
 
-    def test_request_signing7(self):
+    @deferred_f_from_coro_f
+    async def test_request_signing7(self):
         # ensure that spaces are quoted properly before signing
         date = "Tue, 27 Mar 2007 19:42:41 +0000"
         req = Request(
@@ -288,22 +307,25 @@ class TestS3:
             headers={"Date": date},
         )
         with self._mocked_date(date):
-            httpreq = self.download_request(req, self.spider)
+            httpreq = await self.download_request(req)
         assert (
             httpreq.headers["Authorization"]
             == b"AWS 0PN5J17HBGZHT7JJ3X82:+CfvG8EZ3YccOrRVMXNaK2eKZmM="
         )
 
 
+@pytest.mark.skipif(is_botocore_available(), reason="Requires not having botocore")
+def test_s3_no_botocore() -> None:
+    crawler = get_crawler()
+    with pytest.raises(NotConfigured, match="missing botocore library"):
+        build_from_crawler(S3DownloadHandler, crawler)
+
+
 class TestDataURI:
     def setup_method(self):
         crawler = get_crawler()
-        self.download_handler = build_from_crawler(DataURIDownloadHandler, crawler)
-
-    async def download_request(self, request: Request) -> Response:
-        return await maybe_deferred_to_future(
-            self.download_handler.download_request(request, DefaultSpider())
-        )
+        download_handler = build_from_crawler(DataURIDownloadHandler, crawler)
+        self.download_request = download_handler.download_request
 
     @deferred_f_from_coro_f
     async def test_response_attrs(self):
diff --git a/tests/test_downloader_handlers_http_base.py b/tests/test_downloader_handlers_http_base.py
index a2459911f..e325c0463 100644
--- a/tests/test_downloader_handlers_http_base.py
+++ b/tests/test_downloader_handlers_http_base.py
@@ -6,19 +6,18 @@ import gzip
 import json
 import sys
 from abc import ABC, abstractmethod
+from contextlib import asynccontextmanager
 from http import HTTPStatus
 from typing import TYPE_CHECKING, Any
 from unittest import mock
 
 import pytest
-from pytest_twisted import async_yield_fixture
 from testfixtures import LogCapture
 from twisted.internet import defer, error
 from twisted.web._newclient import ResponseFailed
 from twisted.web.http import _DataLoss
 
 from scrapy.http import Headers, HtmlResponse, Request, Response, TextResponse
-from scrapy.spiders import Spider
 from scrapy.utils.asyncio import call_later
 from scrapy.utils.defer import (
     deferred_f_from_coro_f,
@@ -40,27 +39,6 @@ if TYPE_CHECKING:
     from tests.mockserver.http import MockServer
 
 
-async def download_request(
-    download_handler: DownloadHandlerProtocol,
-    request: Request,
-    spider: Spider = DefaultSpider(),
-) -> Response:
-    return await maybe_deferred_to_future(
-        download_handler.download_request(request, spider)
-    )
-
-
-async def close_dh(dh: DownloadHandlerProtocol) -> None:
-    # needed because the interface of close() is not clearly defined
-    if not hasattr(dh, "close"):
-        return
-    c = dh.close()
-    if c is None:
-        return
-    # covers coroutines and Deferreds; won't work if close() uses Futures inside
-    await c
-
-
 class TestHttpBase(ABC):
     is_secure = False
 
@@ -69,30 +47,32 @@ class TestHttpBase(ABC):
     def download_handler_cls(self) -> type[DownloadHandlerProtocol]:
         raise NotImplementedError
 
-    @async_yield_fixture
-    async def download_handler(self) -> AsyncGenerator[DownloadHandlerProtocol]:
-        dh = build_from_crawler(self.download_handler_cls, get_crawler())
-
-        yield dh
-
-        await close_dh(dh)
+    @asynccontextmanager
+    async def get_dh(
+        self, settings_dict: dict[str, Any] | None = None
+    ) -> AsyncGenerator[DownloadHandlerProtocol]:
+        crawler = get_crawler(DefaultSpider, settings_dict)
+        crawler.spider = crawler._create_spider()
+        dh = build_from_crawler(self.download_handler_cls, crawler)
+        try:
+            yield dh
+        finally:
+            await dh.close()
 
     @deferred_f_from_coro_f
-    async def test_download(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_download(self, mockserver: MockServer) -> None:
         request = Request(mockserver.url("/text", is_secure=self.is_secure))
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.body == b"Works"
 
     @deferred_f_from_coro_f
-    async def test_download_head(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_download_head(self, mockserver: MockServer) -> None:
         request = Request(
             mockserver.url("/text", is_secure=self.is_secure), method="HEAD"
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.body == b""
 
     @pytest.mark.parametrize(
@@ -105,22 +85,18 @@ class TestHttpBase(ABC):
     )
     @deferred_f_from_coro_f
     async def test_download_has_correct_http_status_code(
-        self,
-        mockserver: MockServer,
-        download_handler: DownloadHandlerProtocol,
-        http_status: HTTPStatus,
+        self, mockserver: MockServer, http_status: HTTPStatus
     ) -> None:
         request = Request(
             mockserver.url(f"/status?n={http_status.value}", is_secure=self.is_secure)
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.status == http_status.value
 
     @deferred_f_from_coro_f
     async def test_server_receives_correct_request_headers(
-        self,
-        mockserver: MockServer,
-        download_handler: DownloadHandlerProtocol,
+        self, mockserver: MockServer
     ) -> None:
         request_headers = {
             # common request headers
@@ -136,7 +112,8 @@ class TestHttpBase(ABC):
             mockserver.url("/echo", is_secure=self.is_secure),
             headers=request_headers,
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.status == HTTPStatus.OK
         body = json.loads(response.body.decode("utf-8"))
         assert "headers" in body
@@ -146,9 +123,7 @@ class TestHttpBase(ABC):
 
     @deferred_f_from_coro_f
     async def test_server_receives_correct_request_body(
-        self,
-        mockserver: MockServer,
-        download_handler: DownloadHandlerProtocol,
+        self, mockserver: MockServer
     ) -> None:
         request_body = {
             "message": "It works!",
@@ -157,16 +132,15 @@ class TestHttpBase(ABC):
             mockserver.url("/echo", is_secure=self.is_secure),
             body=json.dumps(request_body),
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.status == HTTPStatus.OK
         body = json.loads(response.body.decode("utf-8"))
         assert json.loads(body["body"]) == request_body
 
     @deferred_f_from_coro_f
     async def test_download_has_correct_response_headers(
-        self,
-        mockserver: MockServer,
-        download_handler: DownloadHandlerProtocol,
+        self, mockserver: MockServer
     ) -> None:
         # these headers will be set on the response in the resource and returned
         response_headers = {
@@ -192,7 +166,8 @@ class TestHttpBase(ABC):
             headers={"content-type": "application/json"},
             body=json.dumps(response_headers),
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.status == 200
         for header_name, header_value in response_headers.items():
             assert header_name in response.headers, (
@@ -203,29 +178,24 @@ class TestHttpBase(ABC):
             )
 
     @deferred_f_from_coro_f
-    async def test_redirect_status(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_redirect_status(self, mockserver: MockServer) -> None:
         request = Request(mockserver.url("/redirect", is_secure=self.is_secure))
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.status == 302
 
     @deferred_f_from_coro_f
-    async def test_redirect_status_head(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_redirect_status_head(self, mockserver: MockServer) -> None:
         request = Request(
             mockserver.url("/redirect", is_secure=self.is_secure), method="HEAD"
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.status == 302
 
     @deferred_f_from_coro_f
     async def test_timeout_download_from_spider_nodata_rcvd(
-        self,
-        mockserver: MockServer,
-        download_handler: DownloadHandlerProtocol,
-        reactor_pytest: str,
+        self, mockserver: MockServer, reactor_pytest: str
     ) -> None:
         if reactor_pytest == "asyncio" and sys.platform == "win32":
             # https://twistedmatrix.com/trac/ticket/10279
@@ -236,15 +206,15 @@ class TestHttpBase(ABC):
         # client connects but no data is received
         meta = {"download_timeout": 0.5}
         request = Request(mockserver.url("/wait", is_secure=self.is_secure), meta=meta)
-        d = deferred_from_coro(download_request(download_handler, request))
-        with pytest.raises((defer.TimeoutError, error.TimeoutError)):
-            await maybe_deferred_to_future(d)
+        async with self.get_dh() as download_handler:
+            d = deferred_from_coro(download_handler.download_request(request))
+            with pytest.raises((defer.TimeoutError, error.TimeoutError)):
+                await maybe_deferred_to_future(d)
 
     @deferred_f_from_coro_f
     async def test_timeout_download_from_spider_server_hangs(
         self,
         mockserver: MockServer,
-        download_handler: DownloadHandlerProtocol,
         reactor_pytest: str,
     ) -> None:
         if reactor_pytest == "asyncio" and sys.platform == "win32":
@@ -257,24 +227,21 @@ class TestHttpBase(ABC):
         request = Request(
             mockserver.url("/hang-after-headers", is_secure=self.is_secure), meta=meta
         )
-        d = deferred_from_coro(download_request(download_handler, request))
-        with pytest.raises((defer.TimeoutError, error.TimeoutError)):
-            await maybe_deferred_to_future(d)
+        async with self.get_dh() as download_handler:
+            d = deferred_from_coro(download_handler.download_request(request))
+            with pytest.raises((defer.TimeoutError, error.TimeoutError)):
+                await maybe_deferred_to_future(d)
 
     @pytest.mark.parametrize("send_header", [True, False])
     @deferred_f_from_coro_f
-    async def test_host_header(
-        self,
-        send_header: bool,
-        mockserver: MockServer,
-        download_handler: DownloadHandlerProtocol,
-    ) -> None:
+    async def test_host_header(self, send_header: bool, mockserver: MockServer) -> None:
         host_port = f"{mockserver.host}:{mockserver.port(is_secure=self.is_secure)}"
         request = Request(
             mockserver.url("/host", is_secure=self.is_secure),
             headers={"Host": host_port} if send_header else {},
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.body == host_port.encode()
         if send_header:
             assert request.headers.get("Host") == host_port.encode()
@@ -283,7 +250,7 @@ class TestHttpBase(ABC):
 
     @deferred_f_from_coro_f
     async def test_content_length_zero_bodyless_post_request_headers(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
+        self, mockserver: MockServer
     ) -> None:
         """Tests if "Content-Length: 0" is sent for bodyless POST requests.
 
@@ -298,43 +265,43 @@ class TestHttpBase(ABC):
         request = Request(
             mockserver.url("/contentlength", is_secure=self.is_secure), method="POST"
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.body == b"0"
 
     @deferred_f_from_coro_f
     async def test_content_length_zero_bodyless_post_only_one(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
+        self, mockserver: MockServer
     ) -> None:
         request = Request(
             mockserver.url("/echo", is_secure=self.is_secure), method="POST"
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         headers = Headers(json.loads(response.text)["headers"])
         contentlengths = headers.getlist("Content-Length")
         assert len(contentlengths) == 1
         assert contentlengths == [b"0"]
 
     @deferred_f_from_coro_f
-    async def test_payload(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_payload(self, mockserver: MockServer) -> None:
         body = b"1" * 100  # PayloadResource requires body length to be 100
         request = Request(
             mockserver.url("/payload", is_secure=self.is_secure),
             method="POST",
             body=body,
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.body == body
 
     @deferred_f_from_coro_f
-    async def test_response_header_content_length(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_response_header_content_length(self, mockserver: MockServer) -> None:
         request = Request(
             mockserver.url("/text", is_secure=self.is_secure), method="GET"
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.headers[b"content-length"] == b"5"
 
     @pytest.mark.parametrize(
@@ -351,25 +318,24 @@ class TestHttpBase(ABC):
         body: bytes,
         response_class: type[Response],
         mockserver: MockServer,
-        download_handler: DownloadHandlerProtocol,
     ) -> None:
         request = Request(
             mockserver.url(f"/{filename}", is_secure=self.is_secure), body=body
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert type(response) is response_class  # pylint: disable=unidiomatic-typecheck
 
     @deferred_f_from_coro_f
-    async def test_get_duplicate_header(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_get_duplicate_header(self, mockserver: MockServer) -> None:
         request = Request(mockserver.url("/duplicate-header", is_secure=self.is_secure))
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.headers.getlist(b"Set-Cookie") == [b"a=b", b"c=d"]
 
     @deferred_f_from_coro_f
     async def test_download_is_not_automatically_gzip_decoded(
-        self, download_handler: DownloadHandlerProtocol, mockserver: MockServer
+        self, mockserver: MockServer
     ) -> None:
         """Test download handler does not automatically decode content using the scheme provided in Content-Encoding header"""
 
@@ -382,7 +348,8 @@ class TestHttpBase(ABC):
                 "accept-encoding": "gzip",
             },
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
 
         assert response.status == 200
 
@@ -402,7 +369,7 @@ class TestHttpBase(ABC):
 
     @deferred_f_from_coro_f
     async def test_no_cookie_processing_or_persistence(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
+        self, mockserver: MockServer
     ) -> None:
         cookie_name = "foo"
         cookie_value = "bar"
@@ -413,34 +380,34 @@ class TestHttpBase(ABC):
                 f"/set-cookie?{cookie_name}={cookie_value}", is_secure=self.is_secure
             )
         )
-        response = await download_request(download_handler, request)
-        assert response.status == 200
-        set_cookie = response.headers.get(b"Set-Cookie")
-        assert set_cookie == f"{cookie_name}={cookie_value}".encode()
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
+            assert response.status == 200
+            set_cookie = response.headers.get(b"Set-Cookie")
+            assert set_cookie == f"{cookie_name}={cookie_value}".encode()
 
-        # check that cookies are not sent in the next request
-        request = Request(mockserver.url("/echo", is_secure=self.is_secure))
-        response = await download_request(download_handler, request)
-        assert response.status == 200
-        headers = Headers(json.loads(response.text)["headers"])
-        assert "Cookie" not in headers
-        assert "cookie" not in headers
+            # check that cookies are not sent in the next request
+            request = Request(mockserver.url("/echo", is_secure=self.is_secure))
+            response = await download_handler.download_request(request)
+            assert response.status == 200
+            headers = Headers(json.loads(response.text)["headers"])
+            assert "Cookie" not in headers
+            assert "cookie" not in headers
 
 
 class TestHttp11Base(TestHttpBase):
     """HTTP 1.1 test case"""
 
     @deferred_f_from_coro_f
-    async def test_download_without_maxsize_limit(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_download_without_maxsize_limit(self, mockserver: MockServer) -> None:
         request = Request(mockserver.url("/text", is_secure=self.is_secure))
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.body == b"Works"
 
     @deferred_f_from_coro_f
     async def test_response_class_choosing_request(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
+        self, mockserver: MockServer
     ) -> None:
         """Tests choosing of correct response type
         in case of Content-Type is empty but body contains text.
@@ -449,7 +416,8 @@ class TestHttp11Base(TestHttpBase):
         request = Request(
             mockserver.url("/nocontenttype", is_secure=self.is_secure), body=body
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert type(response) is TextResponse  # pylint: disable=unidiomatic-typecheck
 
     @deferred_f_from_coro_f
@@ -458,24 +426,19 @@ class TestHttp11Base(TestHttpBase):
 
         # 10 is minimal size for this request and the limit is only counted on
         # response body. (regardless of headers)
-        crawler = get_crawler(settings_dict={"DOWNLOAD_MAXSIZE": 5})
-        download_handler = build_from_crawler(self.download_handler_cls, crawler)
-        response = await download_request(download_handler, request, Spider("foo"))
+        async with self.get_dh({"DOWNLOAD_MAXSIZE": 5}) as download_handler:
+            response = await download_handler.download_request(request)
         assert response.body == b"Works"
 
-        crawler = get_crawler(settings_dict={"DOWNLOAD_MAXSIZE": 4})
-        download_handler = build_from_crawler(self.download_handler_cls, crawler)
-
-        with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
-            await download_request(download_handler, request, Spider("foo"))
+        async with self.get_dh({"DOWNLOAD_MAXSIZE": 4}) as download_handler:
+            with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
+                await download_handler.download_request(request)
 
     @deferred_f_from_coro_f
     async def test_download_with_maxsize_very_large_file(
         self, mockserver: MockServer
     ) -> None:
         # TODO: the logger check is specific to scrapy.core.downloader.handlers.http11
-        crawler = get_crawler(settings_dict={"DOWNLOAD_MAXSIZE": 1_500})
-        download_handler = build_from_crawler(self.download_handler_cls, crawler)
         with mock.patch("scrapy.core.downloader.handlers.http11.logger") as logger:
             request = Request(
                 mockserver.url("/largechunkedfile", is_secure=self.is_secure)
@@ -484,8 +447,9 @@ class TestHttp11Base(TestHttpBase):
             def check(logger: mock.Mock) -> None:
                 logger.warning.assert_called_once_with(mock.ANY, mock.ANY)
 
-            with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
-                await download_request(download_handler, request, Spider("foo"))
+            async with self.get_dh({"DOWNLOAD_MAXSIZE": 1_500}) as download_handler:
+                with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
+                    await download_handler.download_request(request)
 
             # As the error message is logged in the dataReceived callback, we
             # have to give a bit of time to the reactor to process the queue
@@ -496,69 +460,61 @@ class TestHttp11Base(TestHttpBase):
             await maybe_deferred_to_future(d)
 
     @deferred_f_from_coro_f
-    async def test_download_with_maxsize_per_req(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_download_with_maxsize_per_req(self, mockserver: MockServer) -> None:
         meta = {"download_maxsize": 2}
         request = Request(mockserver.url("/text", is_secure=self.is_secure), meta=meta)
-        with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
-            await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
+                await download_handler.download_request(request)
 
     @deferred_f_from_coro_f
     async def test_download_with_small_maxsize_via_setting(
         self, mockserver: MockServer
     ) -> None:
-        crawler = get_crawler(settings_dict={"DOWNLOAD_MAXSIZE": 2})
-        download_handler = build_from_crawler(self.download_handler_cls, crawler)
         request = Request(mockserver.url("/text", is_secure=self.is_secure))
-        with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
-            await download_request(download_handler, request, Spider("foo"))
+        async with self.get_dh({"DOWNLOAD_MAXSIZE": 2}) as download_handler:
+            with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
+                await download_handler.download_request(request)
 
     @deferred_f_from_coro_f
     async def test_download_with_large_maxsize_via_setting(
         self, mockserver: MockServer
     ) -> None:
-        crawler = get_crawler(settings_dict={"DOWNLOAD_MAXSIZE": 5})
-        download_handler = build_from_crawler(self.download_handler_cls, crawler)
         request = Request(mockserver.url("/text", is_secure=self.is_secure))
-        response = await download_request(download_handler, request, Spider("foo"))
+        async with self.get_dh({"DOWNLOAD_MAXSIZE": 100}) as download_handler:
+            response = await download_handler.download_request(request)
         assert response.body == b"Works"
 
     @deferred_f_from_coro_f
-    async def test_download_chunked_content(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_download_chunked_content(self, mockserver: MockServer) -> None:
         request = Request(mockserver.url("/chunked", is_secure=self.is_secure))
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.body == b"chunked content\n"
 
     @pytest.mark.parametrize("url", ["broken", "broken-chunked"])
     @deferred_f_from_coro_f
     async def test_download_cause_data_loss(
-        self,
-        url: str,
-        mockserver: MockServer,
-        download_handler: DownloadHandlerProtocol,
+        self, url: str, mockserver: MockServer
     ) -> None:
         # TODO: this one checks for Twisted-specific exceptions
         request = Request(mockserver.url(f"/{url}", is_secure=self.is_secure))
-        with pytest.raises(ResponseFailed) as exc_info:
-            await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            with pytest.raises(ResponseFailed) as exc_info:
+                await download_handler.download_request(request)
         assert any(r.check(_DataLoss) for r in exc_info.value.reasons)
 
     @pytest.mark.parametrize("url", ["broken", "broken-chunked"])
     @deferred_f_from_coro_f
     async def test_download_allow_data_loss(
-        self,
-        url: str,
-        mockserver: MockServer,
-        download_handler: DownloadHandlerProtocol,
+        self, url: str, mockserver: MockServer
     ) -> None:
         request = Request(
             mockserver.url(f"/{url}", is_secure=self.is_secure),
             meta={"download_fail_on_dataloss": False},
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.flags == ["dataloss"]
 
     @pytest.mark.parametrize("url", ["broken", "broken-chunked"])
@@ -566,27 +522,20 @@ class TestHttp11Base(TestHttpBase):
     async def test_download_allow_data_loss_via_setting(
         self, url: str, mockserver: MockServer
     ) -> None:
-        crawler = get_crawler(settings_dict={"DOWNLOAD_FAIL_ON_DATALOSS": False})
-        download_handler = build_from_crawler(self.download_handler_cls, crawler)
         request = Request(mockserver.url(f"/{url}", is_secure=self.is_secure))
-        try:
-            response = await maybe_deferred_to_future(
-                download_handler.download_request(request, DefaultSpider())
-            )
-        finally:
-            d = download_handler.close()  # type: ignore[attr-defined]
-            if d is not None:
-                await maybe_deferred_to_future(d)
+        async with self.get_dh(
+            {"DOWNLOAD_FAIL_ON_DATALOSS": False}
+        ) as download_handler:
+            response = await download_handler.download_request(request)
         assert response.flags == ["dataloss"]
 
     @deferred_f_from_coro_f
-    async def test_protocol(
-        self, mockserver: MockServer, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_protocol(self, mockserver: MockServer) -> None:
         request = Request(
             mockserver.url("/host", is_secure=self.is_secure), method="GET"
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.protocol == "HTTP/1.1"
 
 
@@ -600,24 +549,16 @@ class TestHttps11Base(TestHttp11Base):
 
     @deferred_f_from_coro_f
     async def test_tls_logging(self, mockserver: MockServer) -> None:
-        crawler = get_crawler(
-            settings_dict={"DOWNLOADER_CLIENT_TLS_VERBOSE_LOGGING": True}
-        )
-        download_handler = build_from_crawler(self.download_handler_cls, crawler)
-        try:
+        request = Request(mockserver.url("/text", is_secure=self.is_secure))
+        async with self.get_dh(
+            {"DOWNLOADER_CLIENT_TLS_VERBOSE_LOGGING": True}
+        ) as download_handler:
             with LogCapture() as log_capture:
-                request = Request(mockserver.url("/text", is_secure=self.is_secure))
-                response = await maybe_deferred_to_future(
-                    download_handler.download_request(request, DefaultSpider())
-                )
-                assert response.body == b"Works"
-                log_capture.check_present(
-                    ("scrapy.core.downloader.tls", "DEBUG", self.tls_log_message)
-                )
-        finally:
-            d = download_handler.close()  # type: ignore[attr-defined]
-            if d is not None:
-                await maybe_deferred_to_future(d)
+                response = await download_handler.download_request(request)
+        assert response.body == b"Works"
+        log_capture.check_present(
+            ("scrapy.core.downloader.tls", "DEBUG", self.tls_log_message)
+        )
 
 
 class TestSimpleHttpsBase(ABC):
@@ -645,25 +586,25 @@ class TestSimpleHttpsBase(ABC):
     def download_handler_cls(self) -> type[DownloadHandlerProtocol]:
         raise NotImplementedError
 
-    @async_yield_fixture
-    async def download_handler(self) -> AsyncGenerator[DownloadHandlerProtocol]:
+    @asynccontextmanager
+    async def get_dh(self) -> AsyncGenerator[DownloadHandlerProtocol]:
         if self.cipher_string is not None:
             settings_dict = {"DOWNLOADER_CLIENT_TLS_CIPHERS": self.cipher_string}
         else:
             settings_dict = None
-        crawler = get_crawler(settings_dict=settings_dict)
+        crawler = get_crawler(DefaultSpider, settings_dict=settings_dict)
+        crawler.spider = crawler._create_spider()
         dh = build_from_crawler(self.download_handler_cls, crawler)
-
-        yield dh
-
-        await close_dh(dh)
+        try:
+            yield dh
+        finally:
+            await dh.close()
 
     @deferred_f_from_coro_f
-    async def test_download(
-        self, url: str, download_handler: DownloadHandlerProtocol
-    ) -> None:
+    async def test_download(self, url: str) -> None:
         request = Request(url)
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.body == b"0123456789"
 
 
@@ -748,65 +689,63 @@ class TestHttpProxyBase(ABC):
         with ProxyEchoMockServer() as proxy:
             yield proxy
 
-    @async_yield_fixture
-    async def download_handler(self) -> AsyncGenerator[DownloadHandlerProtocol]:
-        dh = build_from_crawler(self.download_handler_cls, get_crawler())
-
-        yield dh
-
-        await close_dh(dh)
+    @asynccontextmanager
+    async def get_dh(self) -> AsyncGenerator[DownloadHandlerProtocol]:
+        crawler = get_crawler(DefaultSpider)
+        crawler.spider = crawler._create_spider()
+        dh = build_from_crawler(self.download_handler_cls, crawler)
+        try:
+            yield dh
+        finally:
+            await dh.close()
 
     @deferred_f_from_coro_f
     async def test_download_with_proxy(
-        self,
-        proxy_mockserver: ProxyEchoMockServer,
-        download_handler: DownloadHandlerProtocol,
+        self, proxy_mockserver: ProxyEchoMockServer
     ) -> None:
         http_proxy = proxy_mockserver.url("", is_secure=self.is_secure)
         request = Request("http://example.com", meta={"proxy": http_proxy})
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.status == 200
         assert response.url == request.url
         assert response.body == self.expected_http_proxy_request_body
 
     @deferred_f_from_coro_f
     async def test_download_without_proxy(
-        self,
-        proxy_mockserver: ProxyEchoMockServer,
-        download_handler: DownloadHandlerProtocol,
+        self, proxy_mockserver: ProxyEchoMockServer
     ) -> None:
         request = Request(
             proxy_mockserver.url("/path/to/resource", is_secure=self.is_secure)
         )
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.status == 200
         assert response.url == request.url
         assert response.body == b"/path/to/resource"
 
     @deferred_f_from_coro_f
     async def test_download_with_proxy_https_timeout(
-        self,
-        proxy_mockserver: ProxyEchoMockServer,
-        download_handler: DownloadHandlerProtocol,
+        self, proxy_mockserver: ProxyEchoMockServer
     ) -> None:
         if NON_EXISTING_RESOLVABLE:
             pytest.skip("Non-existing hosts are resolvable")
         http_proxy = proxy_mockserver.url("", is_secure=self.is_secure)
         domain = "https://no-such-domain.nosuch"
         request = Request(domain, meta={"proxy": http_proxy, "download_timeout": 0.2})
-        with pytest.raises(error.TimeoutError) as exc_info:
-            await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            with pytest.raises(error.TimeoutError) as exc_info:
+                await download_handler.download_request(request)
         assert domain in exc_info.value.osError
 
     @deferred_f_from_coro_f
     async def test_download_with_proxy_without_http_scheme(
-        self,
-        proxy_mockserver: ProxyEchoMockServer,
-        download_handler: DownloadHandlerProtocol,
+        self, proxy_mockserver: ProxyEchoMockServer
     ) -> None:
         http_proxy = f"{proxy_mockserver.host}:{proxy_mockserver.port()}"
         request = Request("http://example.com", meta={"proxy": http_proxy})
-        response = await download_request(download_handler, request)
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
         assert response.status == 200
         assert response.url == request.url
         assert response.body == self.expected_http_proxy_request_body