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)