RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Full typing for scrapy/http/response/text.py.

This commit is contained in:
Andrey Rakhmatullin 2023-11-01 01:53:25 +04:00
parent 01d9d28324
commit 24f21e96b9
2 changed files with 68 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
scrapy/http/response/__init__.py
View File

@ -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,

113
scrapy/http/response/text.py
View File

@ -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)