diff --git a/scrapy/core/downloader/handlers/datauri.py b/scrapy/core/downloader/handlers/datauri.py index 8b78c53c1..a7ae56a85 100644 --- a/scrapy/core/downloader/handlers/datauri.py +++ b/scrapy/core/downloader/handlers/datauri.py @@ -1,6 +1,9 @@ +from typing import Any, Dict + from w3lib.url import parse_data_uri -from scrapy.http import TextResponse +from scrapy import Request, Spider +from scrapy.http import Response, TextResponse from scrapy.responsetypes import responsetypes from scrapy.utils.decorators import defers @@ -9,11 +12,11 @@ class DataURIDownloadHandler: lazy = False @defers - def download_request(self, request, spider): + def download_request(self, request: Request, spider: Spider) -> Response: uri = parse_data_uri(request.url) respcls = responsetypes.from_mimetype(uri.media_type) - resp_kwargs = {} + resp_kwargs: Dict[str, Any] = {} if issubclass(respcls, TextResponse) and uri.media_type.split("/")[0] == "text": charset = uri.media_type_parameters.get("charset") resp_kwargs["encoding"] = charset diff --git a/scrapy/core/scraper.py b/scrapy/core/scraper.py index b2c26507c..8fb16b8a9 100644 --- a/scrapy/core/scraper.py +++ b/scrapy/core/scraper.py @@ -206,6 +206,7 @@ class Scraper: if isinstance(result, Response): if getattr(result, "request", None) is None: result.request = request + assert result.request callback = result.request.callback or spider._parse warn_on_generator_with_return_value(spider, callback) dfd = defer_succeed(result) diff --git a/scrapy/http/response/__init__.py b/scrapy/http/response/__init__.py index 82274fc3a..8e9237dad 100644 --- a/scrapy/http/response/__init__.py +++ b/scrapy/http/response/__init__.py @@ -4,9 +4,28 @@ responses in Scrapy. See documentation in docs/topics/request-response.rst """ -from typing import Any, AnyStr, Generator, Iterable, Mapping, Tuple, Union +from __future__ import annotations + +from ipaddress import IPv4Address, IPv6Address +from typing import ( + TYPE_CHECKING, + Any, + AnyStr, + Callable, + Dict, + Generator, + Iterable, + List, + Mapping, + Optional, + Tuple, + Union, + cast, +) from urllib.parse import urljoin +from twisted.internet.ssl import Certificate + from scrapy.exceptions import NotSupported from scrapy.http.common import obsolete_setter from scrapy.http.headers import Headers @@ -14,6 +33,9 @@ from scrapy.http.request import Request from scrapy.link import Link from scrapy.utils.trackref import object_ref +if TYPE_CHECKING: + from scrapy.selector import SelectorList + class Response(object_ref): """An object that represents an HTTP response, which is usually @@ -41,29 +63,29 @@ class Response(object_ref): def __init__( self, url: str, - status=200, + status: int = 200, headers: Union[Mapping[AnyStr, Any], Iterable[Tuple[AnyStr, Any]], None] = None, - body=b"", - flags=None, - request=None, - certificate=None, - ip_address=None, - protocol=None, + body: bytes = b"", + flags: Optional[List[str]] = None, + request: Optional[Request] = None, + certificate: Optional[Certificate] = None, + ip_address: Union[IPv4Address, IPv6Address, None] = None, + protocol: Optional[str] = None, ): - self.headers = Headers(headers or {}) - self.status = int(status) + self.headers: Headers = Headers(headers or {}) + self.status: int = int(status) self._set_body(body) self._set_url(url) - self.request = request - self.flags = [] if flags is None else list(flags) - self.certificate = certificate - self.ip_address = ip_address - self.protocol = protocol + self.request: Optional[Request] = request + self.flags: List[str] = [] if flags is None else list(flags) + self.certificate: Optional[Certificate] = certificate + self.ip_address: Union[IPv4Address, IPv6Address, None] = ip_address + self.protocol: Optional[str] = protocol @property - def cb_kwargs(self): + def cb_kwargs(self) -> Dict[str, Any]: try: - return self.request.cb_kwargs + return self.request.cb_kwargs # type: ignore[union-attr] except AttributeError: raise AttributeError( "Response.cb_kwargs not available, this response " @@ -71,21 +93,21 @@ class Response(object_ref): ) @property - def meta(self): + def meta(self) -> Dict[str, Any]: try: - return self.request.meta + return self.request.meta # type: ignore[union-attr] except AttributeError: raise AttributeError( "Response.meta not available, this response " "is not tied to any request" ) - def _get_url(self): + def _get_url(self) -> str: return self._url - def _set_url(self, url: str): + def _set_url(self, url: str) -> None: if isinstance(url, str): - self._url = url + self._url: str = url else: raise TypeError( f"{type(self).__name__} url must be str, " f"got {type(url).__name__}" @@ -93,10 +115,10 @@ class Response(object_ref): url = property(_get_url, obsolete_setter(_set_url, "url")) - def _get_body(self): + def _get_body(self) -> bytes: return self._body - def _set_body(self, body): + def _set_body(self, body: Optional[bytes]) -> None: if body is None: self._body = b"" elif not isinstance(body, bytes): @@ -110,45 +132,45 @@ class Response(object_ref): body = property(_get_body, obsolete_setter(_set_body, "body")) - def __repr__(self): + def __repr__(self) -> str: return f"<{self.status} {self.url}>" - def copy(self): + def copy(self) -> Response: """Return a copy of this Response""" return self.replace() - def replace(self, *args, **kwargs): + def replace(self, *args: Any, **kwargs: Any) -> Response: """Create a new Response with the same attributes except for those given new values""" for x in self.attributes: kwargs.setdefault(x, getattr(self, x)) cls = kwargs.pop("cls", self.__class__) - return cls(*args, **kwargs) + return cast(Response, cls(*args, **kwargs)) - def urljoin(self, url): + def urljoin(self, url: str) -> str: """Join this Response's url with a possible relative url to form an absolute interpretation of the latter.""" - return urljoin(self.url, url) + return urljoin(cast(str, self.url), url) @property - def text(self): + def text(self) -> str: """For subclasses of TextResponse, this will return the body as str """ raise AttributeError("Response content isn't text") - def css(self, *a, **kw): + def css(self, *a: Any, **kw: Any) -> SelectorList: """Shortcut method implemented only by responses whose content is text (subclasses of TextResponse). """ raise NotSupported("Response content isn't text") - def jmespath(self, *a, **kw): + def jmespath(self, *a: Any, **kw: Any) -> SelectorList: """Shortcut method implemented only by responses whose content is text (subclasses of TextResponse). """ raise NotSupported("Response content isn't text") - def xpath(self, *a, **kw): + def xpath(self, *a: Any, **kw: Any) -> SelectorList: """Shortcut method implemented only by responses whose content is text (subclasses of TextResponse). """ @@ -156,19 +178,19 @@ class Response(object_ref): def follow( self, - url, - callback=None, - method="GET", - headers=None, - body=None, - cookies=None, - meta=None, - encoding="utf-8", - priority=0, - dont_filter=False, - errback=None, - cb_kwargs=None, - flags=None, + url: Union[str, Link], + callback: Optional[Callable] = None, + method: str = "GET", + headers: Union[Mapping[AnyStr, Any], Iterable[Tuple[AnyStr, Any]], None] = None, + body: Optional[Union[bytes, str]] = None, + cookies: Optional[Union[dict, List[dict]]] = None, + meta: Optional[Dict[str, Any]] = None, + encoding: str = "utf-8", + priority: int = 0, + dont_filter: bool = False, + errback: Optional[Callable] = None, + cb_kwargs: Optional[Dict[str, Any]] = None, + flags: Optional[List[str]] = None, ) -> Request: """ Return a :class:`~.Request` instance to follow a link ``url``. @@ -207,19 +229,19 @@ class Response(object_ref): def follow_all( self, - urls, - callback=None, - method="GET", - headers=None, - body=None, - cookies=None, - meta=None, - encoding="utf-8", - priority=0, - dont_filter=False, - errback=None, - cb_kwargs=None, - flags=None, + urls: Iterable[Union[str, Link]], + callback: Optional[Callable] = None, + method: str = "GET", + headers: Union[Mapping[AnyStr, Any], Iterable[Tuple[AnyStr, Any]], None] = None, + body: Optional[Union[bytes, str]] = None, + cookies: Optional[Union[dict, List[dict]]] = None, + meta: Optional[Dict[str, Any]] = None, + encoding: str = "utf-8", + priority: int = 0, + dont_filter: bool = False, + errback: Optional[Callable] = None, + cb_kwargs: Optional[Dict[str, Any]] = None, + flags: Optional[List[str]] = None, ) -> Generator[Request, None, None]: """ .. versionadded:: 2.0 diff --git a/scrapy/http/response/text.py b/scrapy/http/response/text.py index 98ae1f307..21a93fbd2 100644 --- a/scrapy/http/response/text.py +++ b/scrapy/http/response/text.py @@ -27,7 +27,7 @@ from scrapy.utils.python import memoizemethod_noargs, to_unicode from scrapy.utils.response import get_base_url if TYPE_CHECKING: - from scrapy.selector import Selector + from scrapy.selector import Selector, SelectorList _NONE = object() @@ -138,26 +138,26 @@ class TextResponse(Response): return read_bom(self.body)[0] @property - def selector(self): + def selector(self) -> Selector: from scrapy.selector import Selector if self._cached_selector is None: self._cached_selector = Selector(self) return self._cached_selector - def jmespath(self, query, **kwargs): + def jmespath(self, query: str, **kwargs: Any) -> SelectorList: if not hasattr(self.selector, "jmespath"): # type: ignore[attr-defined] raise AttributeError( "Please install parsel >= 1.8.1 to get jmespath support" ) - return self.selector.jmespath(query, **kwargs) # type: ignore[attr-defined] + return cast(SelectorList, self.selector.jmespath(query, **kwargs)) # type: ignore[attr-defined] - def xpath(self, query, **kwargs): - return self.selector.xpath(query, **kwargs) + def xpath(self, query: str, **kwargs: Any) -> SelectorList: + return cast(SelectorList, self.selector.xpath(query, **kwargs)) - def css(self, query): - return self.selector.css(query) + def css(self, query: str) -> SelectorList: + return cast(SelectorList, self.selector.css(query)) def follow( self, diff --git a/tests/test_http2_client_protocol.py b/tests/test_http2_client_protocol.py index deb35a579..8fdf3d56f 100644 --- a/tests/test_http2_client_protocol.py +++ b/tests/test_http2_client_protocol.py @@ -600,6 +600,7 @@ class Https2ClientProtocolTestCase(TestCase): def assert_metadata(response: Response): self.assertEqual(response.request, request) self.assertIsInstance(response.certificate, Certificate) + assert response.certificate # typing self.assertIsNotNone(response.certificate.original) self.assertEqual( response.certificate.getIssuer(), self.client_certificate.getIssuer()