From 732557e6988dd3d1df26f973e3ae651f7bcbdc5b Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Tue, 31 Oct 2023 22:34:43 +0400 Subject: [PATCH 1/4] Full typing for scrapy/http/request. --- scrapy/http/common.py | 7 +++-- scrapy/http/request/__init__.py | 47 ++++++++++++++++------------- scrapy/http/request/form.py | 27 +++++++++++------ scrapy/http/request/json_request.py | 8 +++-- scrapy/http/request/rpc.py | 4 +-- 5 files changed, 55 insertions(+), 38 deletions(-) diff --git a/scrapy/http/common.py b/scrapy/http/common.py index bc8861574..a3d9d5b81 100644 --- a/scrapy/http/common.py +++ b/scrapy/http/common.py @@ -1,5 +1,8 @@ -def obsolete_setter(setter, attrname): - def newsetter(self, value): +from typing import Any, Callable, NoReturn + + +def obsolete_setter(setter: Callable, attrname: str) -> Callable[[Any, Any], NoReturn]: + def newsetter(self: Any, value: Any) -> NoReturn: c = self.__class__.__name__ msg = f"{c}.{attrname} is not modifiable, use {c}.replace() instead" raise AttributeError(msg) diff --git a/scrapy/http/request/__init__.py b/scrapy/http/request/__init__.py index 06a52f5fd..7c9a4ba95 100644 --- a/scrapy/http/request/__init__.py +++ b/scrapy/http/request/__init__.py @@ -9,14 +9,17 @@ from typing import ( Any, AnyStr, Callable, + Dict, Iterable, List, Mapping, + NoReturn, Optional, Tuple, Type, TypeVar, Union, + cast, ) from w3lib.url import safe_url_string @@ -32,7 +35,7 @@ from scrapy.utils.url import escape_ajax RequestTypeVar = TypeVar("RequestTypeVar", bound="Request") -def NO_CALLBACK(*args, **kwargs): +def NO_CALLBACK(*args: Any, **kwargs: Any) -> NoReturn: """When assigned to the ``callback`` parameter of :class:`~scrapy.http.Request`, it indicates that the request is not meant to have a spider callback at all. @@ -92,21 +95,21 @@ class Request(object_ref): 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] = None, + meta: Optional[Dict[str, Any]] = None, encoding: str = "utf-8", priority: int = 0, dont_filter: bool = False, errback: Optional[Callable] = None, flags: Optional[List[str]] = None, - cb_kwargs: Optional[dict] = None, + cb_kwargs: Optional[Dict[str, Any]] = None, ) -> None: - self._encoding = encoding # this one has to be set first - self.method = str(method).upper() + self._encoding: str = encoding # this one has to be set first + self.method: str = str(method).upper() self._set_url(url) self._set_body(body) if not isinstance(priority, int): raise TypeError(f"Request priority not an integer: {priority!r}") - self.priority = priority + self.priority: int = priority if not (callable(callback) or callback is None): raise TypeError( @@ -114,25 +117,27 @@ class Request(object_ref): ) if not (callable(errback) or errback is None): raise TypeError(f"errback must be a callable, got {type(errback).__name__}") - self.callback = callback - self.errback = errback + self.callback: Optional[Callable] = callback + self.errback: Optional[Callable] = errback - self.cookies = cookies or {} - self.headers = Headers(headers or {}, encoding=encoding) - self.dont_filter = dont_filter + self.cookies: Union[dict, List[dict]] = cookies or {} + self.headers: Headers = Headers(headers or {}, encoding=encoding) + self.dont_filter: bool = dont_filter - self._meta = dict(meta) if meta else None - self._cb_kwargs = dict(cb_kwargs) if cb_kwargs else None - self.flags = [] if flags is None else list(flags) + self._meta: Optional[Dict[str, Any]] = dict(meta) if meta else None + self._cb_kwargs: Optional[Dict[str, Any]] = ( + dict(cb_kwargs) if cb_kwargs else None + ) + self.flags: List[str] = [] if flags is None else list(flags) @property - def cb_kwargs(self) -> dict: + def cb_kwargs(self) -> Dict[str, Any]: if self._cb_kwargs is None: self._cb_kwargs = {} return self._cb_kwargs @property - def meta(self) -> dict: + def meta(self) -> Dict[str, Any]: if self._meta is None: self._meta = {} return self._meta @@ -174,19 +179,19 @@ class Request(object_ref): def copy(self) -> "Request": return self.replace() - def replace(self, *args, **kwargs) -> "Request": + def replace(self, *args: Any, **kwargs: Any) -> "Request": """Create a new Request 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(Request, cls(*args, **kwargs)) @classmethod def from_curl( cls: Type[RequestTypeVar], curl_command: str, ignore_unknown_options: bool = True, - **kwargs, + **kwargs: Any, ) -> RequestTypeVar: """Create a Request object from a string containing a `cURL `_ command. It populates the HTTP method, the @@ -219,7 +224,7 @@ class Request(object_ref): request_kwargs.update(kwargs) return cls(**request_kwargs) - def to_dict(self, *, spider: Optional["scrapy.Spider"] = None) -> dict: + def to_dict(self, *, spider: Optional["scrapy.Spider"] = None) -> Dict[str, Any]: """Return a dictionary containing the Request's data. Use :func:`~scrapy.utils.request.request_from_dict` to convert back into a :class:`~scrapy.Request` object. @@ -244,7 +249,7 @@ class Request(object_ref): return d -def _find_method(obj, func): +def _find_method(obj: Any, func: Callable) -> str: """Helper function for Request.to_dict""" # Only instance methods contain ``__func__`` if obj and hasattr(func, "__func__"): diff --git a/scrapy/http/request/form.py b/scrapy/http/request/form.py index 8da779b7c..0f80a0ab7 100644 --- a/scrapy/http/request/form.py +++ b/scrapy/http/request/form.py @@ -5,7 +5,9 @@ This module implements the FormRequest class which is a more convenient class See documentation in docs/topics/request-response.rst """ -from typing import Iterable, List, Optional, Tuple, Type, TypeVar, Union, cast +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Iterable, List, Optional, Tuple, Union, cast from urllib.parse import urlencode, urljoin, urlsplit, urlunsplit from lxml.html import ( @@ -24,7 +26,10 @@ from scrapy.http.response.text import TextResponse from scrapy.utils.python import is_listlike, to_bytes from scrapy.utils.response import get_base_url -FormRequestTypeVar = TypeVar("FormRequestTypeVar", bound="FormRequest") +if TYPE_CHECKING: + # typing.Self requires Python 3.11 + from typing_extensions import Self + FormdataKVType = Tuple[str, Union[str, Iterable[str]]] FormdataType = Optional[Union[dict, List[FormdataKVType]]] @@ -33,7 +38,9 @@ FormdataType = Optional[Union[dict, List[FormdataKVType]]] class FormRequest(Request): valid_form_methods = ["GET", "POST"] - def __init__(self, *args, formdata: FormdataType = None, **kwargs) -> None: + def __init__( + self, *args: Any, formdata: FormdataType = None, **kwargs: Any + ) -> None: if formdata and kwargs.get("method") is None: kwargs["method"] = "POST" @@ -54,7 +61,7 @@ class FormRequest(Request): @classmethod def from_response( - cls: Type[FormRequestTypeVar], + cls, response: TextResponse, formname: Optional[str] = None, formid: Optional[str] = None, @@ -64,8 +71,8 @@ class FormRequest(Request): dont_click: bool = False, formxpath: Optional[str] = None, formcss: Optional[str] = None, - **kwargs, - ) -> FormRequestTypeVar: + **kwargs: Any, + ) -> Self: kwargs.setdefault("encoding", response.encoding) if formcss is not None: @@ -121,12 +128,12 @@ def _get_form( if formname is not None: f = root.xpath(f'//form[@name="{formname}"]') if f: - return f[0] + return cast(FormElement, f[0]) if formid is not None: f = root.xpath(f'//form[@id="{formid}"]') if f: - return f[0] + return cast(FormElement, f[0]) # Get form element from xpath, if not found, go up if formxpath is not None: @@ -135,7 +142,7 @@ def _get_form( el = nodes[0] while True: if el.tag == "form": - return el + return cast(FormElement, el) el = el.getparent() if el is None: break @@ -147,7 +154,7 @@ def _get_form( except IndexError: raise IndexError(f"Form number {formnumber} not found in {response}") else: - return form + return cast(FormElement, form) def _get_inputs( diff --git a/scrapy/http/request/json_request.py b/scrapy/http/request/json_request.py index f52c0401d..1dd9e6c87 100644 --- a/scrapy/http/request/json_request.py +++ b/scrapy/http/request/json_request.py @@ -8,7 +8,7 @@ See documentation in docs/topics/request-response.rst import copy import json import warnings -from typing import Optional, Tuple +from typing import Any, Optional, Tuple from scrapy.http.request import Request @@ -16,7 +16,9 @@ from scrapy.http.request import Request class JsonRequest(Request): attributes: Tuple[str, ...] = Request.attributes + ("dumps_kwargs",) - def __init__(self, *args, dumps_kwargs: Optional[dict] = None, **kwargs) -> None: + def __init__( + self, *args: Any, dumps_kwargs: Optional[dict] = None, **kwargs: Any + ) -> None: dumps_kwargs = copy.deepcopy(dumps_kwargs) if dumps_kwargs is not None else {} dumps_kwargs.setdefault("sort_keys", True) self._dumps_kwargs = dumps_kwargs @@ -42,7 +44,7 @@ class JsonRequest(Request): def dumps_kwargs(self) -> dict: return self._dumps_kwargs - def replace(self, *args, **kwargs) -> Request: + def replace(self, *args: Any, **kwargs: Any) -> Request: body_passed = kwargs.get("body", None) is not None data = kwargs.pop("data", None) data_passed = data is not None diff --git a/scrapy/http/request/rpc.py b/scrapy/http/request/rpc.py index 43692923b..bde860a66 100644 --- a/scrapy/http/request/rpc.py +++ b/scrapy/http/request/rpc.py @@ -5,7 +5,7 @@ This module implements the XmlRpcRequest class which is a more convenient class See documentation in docs/topics/request-response.rst """ import xmlrpc.client as xmlrpclib -from typing import Optional +from typing import Any, Optional from scrapy.http.request import Request from scrapy.utils.python import get_func_args @@ -14,7 +14,7 @@ DUMPS_ARGS = get_func_args(xmlrpclib.dumps) class XmlRpcRequest(Request): - def __init__(self, *args, encoding: Optional[str] = None, **kwargs): + def __init__(self, *args: Any, encoding: Optional[str] = None, **kwargs: Any): if "body" not in kwargs and "params" in kwargs: kw = dict((k, kwargs.pop(k)) for k in DUMPS_ARGS if k in kwargs) kwargs["body"] = xmlrpclib.dumps(**kw) From 4cb2fc2c3b9deba49ae3a32400819ed95ea262c9 Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Tue, 31 Oct 2023 22:55:39 +0400 Subject: [PATCH 2/4] Update typing package versions. --- scrapy/utils/ssl.py | 2 +- tox.ini | 10 +++++----- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/scrapy/utils/ssl.py b/scrapy/utils/ssl.py index d520ef809..e74769c65 100644 --- a/scrapy/utils/ssl.py +++ b/scrapy/utils/ssl.py @@ -1,6 +1,6 @@ from typing import Any, Optional -import OpenSSL._util as pyOpenSSLutil +import OpenSSL._util as pyOpenSSLutil # type: ignore[import-untyped] import OpenSSL.SSL import OpenSSL.version from OpenSSL.crypto import X509Name diff --git a/tox.ini b/tox.ini index 381da9773..932c0b805 100644 --- a/tox.ini +++ b/tox.ini @@ -33,13 +33,13 @@ install_command = [testenv:typing] basepython = python3 deps = - mypy==1.5.1 - typing-extensions==4.7.1 + mypy==1.6.1 + typing-extensions==4.8.0 types-attrs==19.1.0 - types-lxml==2023.3.28 - types-Pillow==10.0.0.3 + types-lxml==2023.10.21 + types-Pillow==10.1.0.0 types-Pygments==2.16.0.0 - types-pyOpenSSL==23.2.0.2 + types-pyOpenSSL==23.3.0.0 types-setuptools==68.2.0.0 # 2.1.2 fixes a typing bug: https://github.com/scrapy/w3lib/pull/211 w3lib >= 2.1.2 From 01d9d28324cbd5c7edff0d25ba00d8491400698a Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Tue, 31 Oct 2023 23:14:00 +0400 Subject: [PATCH 3/4] Full typing for scrapy/http/response/__init__.py. --- scrapy/core/downloader/handlers/datauri.py | 9 +- scrapy/core/scraper.py | 1 + scrapy/http/response/__init__.py | 142 ++++++++++++--------- scrapy/http/response/text.py | 16 +-- tests/test_http2_client_protocol.py | 1 + 5 files changed, 98 insertions(+), 71 deletions(-) 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() From 24f21e96b99bac1ba991fe02c8823782833ca298 Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Wed, 1 Nov 2023 01:53:25 +0400 Subject: [PATCH 4/4] Full typing for scrapy/http/response/text.py. --- scrapy/http/response/__init__.py | 6 +- scrapy/http/response/text.py | 113 +++++++++++++++++-------------- 2 files changed, 68 insertions(+), 51 deletions(-) diff --git a/scrapy/http/response/__init__.py b/scrapy/http/response/__init__.py index 8e9237dad..61010f14f 100644 --- a/scrapy/http/response/__init__.py +++ b/scrapy/http/response/__init__.py @@ -185,7 +185,7 @@ class Response(object_ref): body: Optional[Union[bytes, str]] = None, cookies: Optional[Union[dict, List[dict]]] = None, meta: Optional[Dict[str, Any]] = None, - encoding: str = "utf-8", + encoding: Optional[str] = "utf-8", priority: int = 0, dont_filter: bool = False, errback: Optional[Callable] = None, @@ -205,6 +205,8 @@ class Response(object_ref): .. versionadded:: 2.0 The *flags* parameter. """ + if encoding is None: + raise ValueError("encoding can't be None") if isinstance(url, Link): url = url.url elif url is None: @@ -236,7 +238,7 @@ class Response(object_ref): body: Optional[Union[bytes, str]] = None, cookies: Optional[Union[dict, List[dict]]] = None, meta: Optional[Dict[str, Any]] = None, - encoding: str = "utf-8", + encoding: Optional[str] = "utf-8", priority: int = 0, dont_filter: bool = False, errback: Optional[Callable] = None, diff --git a/scrapy/http/response/text.py b/scrapy/http/response/text.py index 21a93fbd2..6596d8a5c 100644 --- a/scrapy/http/response/text.py +++ b/scrapy/http/response/text.py @@ -8,7 +8,21 @@ from __future__ import annotations import json from contextlib import suppress -from typing import TYPE_CHECKING, Any, Generator, Optional, Tuple, cast +from typing import ( + TYPE_CHECKING, + Any, + AnyStr, + Callable, + Dict, + Generator, + Iterable, + List, + Mapping, + Optional, + Tuple, + Union, + cast, +) from urllib.parse import urljoin import parsel @@ -23,6 +37,7 @@ from w3lib.html import strip_html5_whitespace from scrapy.http import Request from scrapy.http.response import Response +from scrapy.link import Link from scrapy.utils.python import memoizemethod_noargs, to_unicode from scrapy.utils.response import get_base_url @@ -39,20 +54,14 @@ class TextResponse(Response): attributes: Tuple[str, ...] = Response.attributes + ("encoding",) def __init__(self, *args: Any, **kwargs: Any): - self._encoding = kwargs.pop("encoding", None) + self._encoding: Optional[str] = kwargs.pop("encoding", None) self._cached_benc: Optional[str] = None self._cached_ubody: Optional[str] = None self._cached_selector: Optional[Selector] = None super().__init__(*args, **kwargs) - def _set_url(self, url): - if isinstance(url, str): - self._url = to_unicode(url, self.encoding) - else: - super()._set_url(url) - - def _set_body(self, body): - self._body = b"" # used by encoding detection + def _set_body(self, body: Union[str, bytes, None]) -> None: + self._body: bytes = b"" # used by encoding detection if isinstance(body, str): if self._encoding is None: raise TypeError( @@ -64,10 +73,10 @@ class TextResponse(Response): super()._set_body(body) @property - def encoding(self): + def encoding(self) -> str: return self._declared_encoding() or self._body_inferred_encoding() - def _declared_encoding(self): + def _declared_encoding(self) -> Optional[str]: return ( self._encoding or self._bom_encoding() @@ -75,7 +84,7 @@ class TextResponse(Response): or self._body_declared_encoding() ) - def json(self): + def json(self) -> Any: """ .. versionadded:: 2.2 @@ -96,7 +105,7 @@ class TextResponse(Response): self._cached_ubody = html_to_unicode(charset, self.body)[1] return self._cached_ubody - 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(get_base_url(self), url) @@ -106,7 +115,7 @@ class TextResponse(Response): content_type = cast(bytes, self.headers.get(b"Content-Type", b"")) return http_content_type_encoding(to_unicode(content_type, encoding="latin-1")) - def _body_inferred_encoding(self): + def _body_inferred_encoding(self) -> str: if self._cached_benc is None: content_type = to_unicode( cast(bytes, self.headers.get(b"Content-Type", b"")), encoding="latin-1" @@ -121,20 +130,21 @@ class TextResponse(Response): self._cached_ubody = ubody return self._cached_benc - def _auto_detect_fun(self, text): + def _auto_detect_fun(self, text: bytes) -> Optional[str]: for enc in (self._DEFAULT_ENCODING, "utf-8", "cp1252"): try: text.decode(enc) except UnicodeError: continue return resolve_encoding(enc) + return None @memoizemethod_noargs - def _body_declared_encoding(self): + def _body_declared_encoding(self) -> Optional[str]: return html_body_declared_encoding(self.body) @memoizemethod_noargs - def _bom_encoding(self): + def _bom_encoding(self) -> Optional[str]: return read_bom(self.body)[0] @property @@ -146,6 +156,8 @@ class TextResponse(Response): return self._cached_selector def jmespath(self, query: str, **kwargs: Any) -> SelectorList: + from scrapy.selector import SelectorList + if not hasattr(self.selector, "jmespath"): # type: ignore[attr-defined] raise AttributeError( "Please install parsel >= 1.8.1 to get jmespath support" @@ -154,26 +166,30 @@ class TextResponse(Response): return cast(SelectorList, self.selector.jmespath(query, **kwargs)) # type: ignore[attr-defined] def xpath(self, query: str, **kwargs: Any) -> SelectorList: + from scrapy.selector import SelectorList + return cast(SelectorList, self.selector.xpath(query, **kwargs)) def css(self, query: str) -> SelectorList: + from scrapy.selector import SelectorList + return cast(SelectorList, self.selector.css(query)) def follow( self, - url, - callback=None, - method="GET", - headers=None, - body=None, - cookies=None, - meta=None, - encoding=None, - priority=0, - dont_filter=False, - errback=None, - cb_kwargs=None, - flags=None, + url: Union[str, Link, parsel.Selector], + 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: Optional[str] = None, + 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``. @@ -214,21 +230,21 @@ class TextResponse(Response): def follow_all( self, - urls=None, - callback=None, - method="GET", - headers=None, - body=None, - cookies=None, - meta=None, - encoding=None, - priority=0, - dont_filter=False, - errback=None, - cb_kwargs=None, - flags=None, - css=None, - xpath=None, + urls: Union[Iterable[Union[str, Link]], parsel.SelectorList, None] = None, + 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: Optional[str] = None, + priority: int = 0, + dont_filter: bool = False, + errback: Optional[Callable] = None, + cb_kwargs: Optional[Dict[str, Any]] = None, + flags: Optional[List[str]] = None, + css: Optional[str] = None, + xpath: Optional[str] = None, ) -> Generator[Request, None, None]: """ A generator that produces :class:`~.Request` instances to follow all @@ -270,7 +286,7 @@ class TextResponse(Response): with suppress(_InvalidSelector): urls.append(_url_from_selector(sel)) return super().follow_all( - urls=urls, + urls=cast(Iterable[Union[str, Link]], urls), callback=callback, method=method, headers=headers, @@ -292,8 +308,7 @@ class _InvalidSelector(ValueError): """ -def _url_from_selector(sel): - # type: (parsel.Selector) -> str +def _url_from_selector(sel: parsel.Selector) -> str: if isinstance(sel.root, str): # e.g. ::attr(href) result return strip_html5_whitespace(sel.root)