Address remaining typing issues in scrapy.core.

scrapy/core/downloader/__init__.py

@@ -95,7 +95,7 @@ class Downloader:
         dfd = self.middleware.download(self._enqueue_request, request, spider)
         return dfd.addBoth(_deactivate)
 
-    def needs_backout(self):
+    def needs_backout(self) -> bool:
         return len(self.active) >= self.total_concurrency
 
     def _get_slot(self, request, spider):

scrapy/core/downloader/handlers/http2.py

@@ -28,7 +28,7 @@ class H2DownloadHandler:
 
         from twisted.internet import reactor
 
-        self._pool = H2ConnectionPool(reactor, settings)
+        self._pool = H2ConnectionPool(reactor, settings)  # type: ignore[arg-type]
         self._context_factory = load_context_factory_from_settings(settings, crawler)
 
     @classmethod
@@ -82,7 +82,7 @@ class ScrapyH2Agent:
                     "Tunneling via CONNECT method using HTTP/2.0 is not yet supported"
                 )
             return self._ProxyAgent(
-                reactor=reactor,
+                reactor=reactor,  # type: ignore[arg-type]
                 context_factory=self._context_factory,
                 proxy_uri=URI.fromBytes(to_bytes(proxy, encoding="ascii")),
                 connect_timeout=timeout,
@@ -91,7 +91,7 @@ class ScrapyH2Agent:
             )
 
         return self._Agent(
-            reactor=reactor,
+            reactor=reactor,  # type: ignore[arg-type]
             context_factory=self._context_factory,
             connect_timeout=timeout,
             bind_address=bind_address,
@@ -108,7 +108,7 @@ class ScrapyH2Agent:
         d = agent.request(request, spider)
         d.addCallback(self._cb_latency, request, start_time)
 
-        timeout_cl = reactor.callLater(timeout, d.cancel)
+        timeout_cl = reactor.callLater(timeout, d.cancel)  # type: ignore[attr-defined]
         d.addBoth(self._cb_timeout, request, timeout, timeout_cl)
         return d
 

scrapy/core/engine.py

@@ -7,13 +7,24 @@ For more information see docs/topics/architecture.rst
 import logging
 import warnings
 from time import time
-from typing import Any, Callable, Generator, Iterable, Iterator, Optional, Set, Union
+from typing import (
+    Any,
+    Callable,
+    cast,
+    Generator,
+    Iterable,
+    Iterator,
+    Optional,
+    Set,
+    Union,
+)
 
 from twisted.internet.defer import Deferred, inlineCallbacks, succeed
 from twisted.internet.task import LoopingCall
 from twisted.python.failure import Failure
 
 from scrapy import signals
+from scrapy.core.downloader import Downloader
 from scrapy.core.scraper import Scraper
 from scrapy.exceptions import (
     CloseSpider,
@@ -80,7 +91,7 @@ class ExecutionEngine:
         self.paused = False
         self.scheduler_cls = self._get_scheduler_class(crawler.settings)
         downloader_cls = load_object(self.settings["DOWNLOADER"])
-        self.downloader = downloader_cls(crawler)
+        self.downloader: Downloader = downloader_cls(crawler)
         self.scraper = Scraper(crawler)
         self._spider_closed_callback = spider_closed_callback
 
@@ -102,7 +113,7 @@ class ExecutionEngine:
         self.start_time = time()
         yield self.signals.send_catch_log_deferred(signal=signals.engine_started)
         self.running = True
-        self._closewait = Deferred()
+        self._closewait: Deferred = Deferred()
         yield self._closewait
 
     def stop(self) -> Deferred:
@@ -177,11 +188,13 @@ class ExecutionEngine:
             self._spider_idle()
 
     def _needs_backout(self) -> bool:
+        assert self.slot is not None  # typing
+        assert self.scraper.slot is not None  # typing
         return (
             not self.running
-            or self.slot.closing  # type: ignore[union-attr]
+            or bool(self.slot.closing)
             or self.downloader.needs_backout()
-            or self.scraper.slot.needs_backout()  # type: ignore[union-attr]
+            or self.scraper.slot.needs_backout()
         )
 
     def _next_request_from_scheduler(self) -> Optional[Deferred]:
@@ -201,7 +214,7 @@ class ExecutionEngine:
                 extra={"spider": self.spider},
             )
         )
-        d.addBoth(lambda _: self.slot.remove_request(request))
+        d.addBoth(lambda _: cast(Slot, self.slot).remove_request(request))
         d.addErrback(
             lambda f: logger.info(
                 "Error while removing request from slot",
@@ -429,7 +442,7 @@ class ExecutionEngine:
         dfd.addErrback(log_failure("Scraper close failure"))
 
         if hasattr(self.slot.scheduler, "close"):
-            dfd.addBoth(lambda _: self.slot.scheduler.close(reason))
+            dfd.addBoth(lambda _: cast(Slot, self.slot).scheduler.close(reason))
             dfd.addErrback(log_failure("Scheduler close failure"))
 
         dfd.addBoth(

scrapy/core/http2/agent.py

@@ -35,7 +35,7 @@ class H2ConnectionPool:
             # Received a request while connecting to remote
             # Create a deferred which will fire with the H2ClientProtocol
             # instance
-            d = Deferred()
+            d: Deferred = Deferred()
             self._pending_requests[key].append(d)
             return d
 
@@ -53,14 +53,14 @@ class H2ConnectionPool:
     ) -> Deferred:
         self._pending_requests[key] = deque()
 
-        conn_lost_deferred = Deferred()
+        conn_lost_deferred: Deferred = Deferred()
         conn_lost_deferred.addCallback(self._remove_connection, key)
 
         factory = H2ClientFactory(uri, self.settings, conn_lost_deferred)
         conn_d = endpoint.connect(factory)
         conn_d.addCallback(self.put_connection, key)
 
-        d = Deferred()
+        d: Deferred = Deferred()
         self._pending_requests[key].append(d)
         return d
 
@@ -83,7 +83,8 @@ class H2ConnectionPool:
         pending_requests = self._pending_requests.pop(key, None)
         while pending_requests:
             d = pending_requests.popleft()
-            d.errback(errors)
+            # TODO: this is incorrect, errback takes a single exception
+            d.errback(errors)  # type: ignore[arg-type]
 
     def close_connections(self) -> None:
         """Close all the HTTP/2 connections and remove them from pool
@@ -92,6 +93,7 @@ class H2ConnectionPool:
             Deferred that fires when all connections have been closed
         """
         for conn in self._connections.values():
+            assert conn.transport is not None  # typing
             conn.transport.abortConnection()
 
 

scrapy/core/http2/protocol.py

@@ -141,6 +141,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin):
         This is used while initiating pending streams to make sure
         that we initiate stream only during active HTTP/2 Connection
         """
+        assert self.transport is not None  # typing
         return bool(self.transport.connected) and self.metadata["settings_acknowledged"]
 
     @property
@@ -197,6 +198,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin):
         """Write data to the underlying transport connection
         from the HTTP2 connection instance if any
         """
+        assert self.transport is not None  # typing
         # Reset the idle timeout as connection is still actively sending data
         self.resetTimeout()
 
@@ -227,6 +229,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin):
         # Initialize the timeout
         self.setTimeout(self.IDLE_TIMEOUT)
 
+        assert self.transport is not None  # typing
         destination = self.transport.getPeer()
         self.metadata["ip_address"] = ipaddress.ip_address(destination.host)
 
@@ -238,12 +241,14 @@ class H2ClientProtocol(Protocol, TimeoutMixin):
         """Helper function to lose the connection with the error sent as a
         reason"""
         self._conn_lost_errors += errors
+        assert self.transport is not None  # typing
         self.transport.loseConnection()
 
     def handshakeCompleted(self) -> None:
         """
         Close the connection if it's not made via the expected protocol
         """
+        assert self.transport is not None  # typing
         if (
             self.transport.negotiatedProtocol is not None
             and self.transport.negotiatedProtocol != PROTOCOL_NAME
@@ -276,6 +281,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin):
                 # hyper-h2 does not drop the connection in this scenario, we
                 # need to abort the connection manually.
                 self._conn_lost_errors += [e]
+                assert self.transport is not None  # typing
                 self.transport.abortConnection()
                 return
 
@@ -389,6 +395,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin):
         self._send_pending_requests()
 
         # Update certificate when our HTTP/2 connection is established
+        assert self.transport is not None  # typing
         self.metadata["certificate"] = Certificate(self.transport.getPeerCertificate())
 
     def stream_ended(self, event: StreamEnded) -> None:

scrapy/core/http2/stream.py

@@ -151,7 +151,7 @@ class Stream:
             else:
                 self.close(StreamCloseReason.CANCELLED)
 
-        self._deferred_response = Deferred(_cancel)
+        self._deferred_response: Deferred = Deferred(_cancel)
 
     def __repr__(self) -> str:
         return f"Stream(id={self.stream_id!r})"

scrapy/core/scraper.py

@@ -66,7 +66,7 @@ class Slot:
     def add_response_request(
         self, result: Union[Response, Failure], request: Request
     ) -> Deferred:
-        deferred = Deferred()
+        deferred: Deferred = Deferred()
         self.queue.append((result, request, deferred))
         if isinstance(result, Response):
             self.active_size += max(len(result.body), self.MIN_RESPONSE_SIZE)
@@ -205,10 +205,12 @@ class Scraper:
                 callback=callback, callbackKeywords=result.request.cb_kwargs
             )
         else:  # result is a Failure
-            result.request = request
+            # TODO: properly type adding this attribute to a Failure
+            result.request = request  # type: ignore[attr-defined]
             warn_on_generator_with_return_value(spider, request.errback)
             dfd = defer_fail(result)
-            dfd.addErrback(request.errback)
+            if request.errback:
+                dfd.addErrback(request.errback)
         return dfd.addCallback(iterate_spider_output)
 
     def handle_spider_error(

scrapy/core/spidermw.py

@@ -157,7 +157,7 @@ class SpiderMiddlewareManager(MiddlewareManager):
                 # simplified when downgrading is removed.
                 if dfd.called:
                     # the result is available immediately if _process_spider_output didn't do downgrading
-                    return dfd.result
+                    return cast(MutableChain, dfd.result)
                 # we forbid waiting here because otherwise we would need to return a deferred from
                 # _process_spider_exception too, which complicates the architecture
                 msg = f"Async iterable returned from {method.__qualname__} cannot be downgraded"