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/__init__.py.

This commit is contained in:
Andrey Rakhmatullin 2023-10-31 23:14:00 +04:00
parent 4cb2fc2c3b
commit 01d9d28324
5 changed files with 98 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
scrapy/core/downloader/handlers/datauri.py
View File

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

1
scrapy/core/scraper.py
View File

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

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

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

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

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

1
tests/test_http2_client_protocol.py
View File

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