mirror of https://github.com/scrapy/scrapy.git
Merge pull request #6127 from wRAR/typing-response
Full typing for Request and Response
This commit is contained in:
Andrey Rakhmatullin
GitHub
commit
593bfd895a
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
<https://curl.haxx.se/>`_ 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__"):
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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: Optional[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``.
|
||||
|
|
@ -183,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:
|
||||
|
|
@ -207,19 +231,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: Optional[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
|
||||
|
|
|
|||
|
|
@ -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,11 +37,12 @@ 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
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from scrapy.selector import Selector
|
||||
from scrapy.selector import Selector, SelectorList
|
||||
|
||||
_NONE = object()
|
||||
|
||||
|
|
@ -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,59 +130,66 @@ 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
|
||||
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:
|
||||
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"
|
||||
)
|
||||
|
||||
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:
|
||||
from scrapy.selector import SelectorList
|
||||
|
||||
def css(self, query):
|
||||
return self.selector.css(query)
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
10
tox.ini
10
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
|
||||
|
|
|
|||
Loading…
Reference in New Issue