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

Merge pull request #5171 from elacuesta/request-types 

Type hints for Request and subclasses
This commit is contained in:
Andrey Rahmatullin 2021-06-11 17:11:19 +05:00 committed by GitHub
parent 28858574d9 ce6447731a
commit 9f81de2a50
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 101 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
scrapy/downloadermiddlewares/retry.py
View File

@ -98,7 +98,7 @@ def get_retry_request(
{'request': request, 'retry_times': retry_times, 'reason': reason},
extra={'spider': spider}
)
new_request = request.copy()
new_request: Request = request.copy()
new_request.meta['retry_times'] = retry_times
new_request.dont_filter = True
if priority_adjust is None:

57
scrapy/http/request/__init__.py
View File

@ -5,7 +5,7 @@ requests in Scrapy.
See documentation in docs/topics/request-response.rst
"""
import inspect
from typing import Optional, Tuple
from typing import Callable, List, Optional, Tuple, Type, TypeVar, Union
from w3lib.url import safe_url_string
@ -18,6 +18,9 @@ from scrapy.utils.trackref import object_ref
from scrapy.utils.url import escape_ajax
RequestTypeVar = TypeVar("RequestTypeVar", bound="Request")
class Request(object_ref):
"""Represents an HTTP request, which is usually generated in a Spider and
executed by the Downloader, thus generating a :class:`Response`.
@ -36,10 +39,22 @@ class Request(object_ref):
:func:`~scrapy.utils.request.request_from_dict`.
"""
def __init__(self, url, callback=None, method='GET', headers=None, body=None,
cookies=None, meta=None, encoding='utf-8', priority=0,
dont_filter=False, errback=None, flags=None, cb_kwargs=None):
def __init__(
self,
url: str,
callback: Optional[Callable] = None,
method: str = "GET",
headers: Optional[dict] = None,
body: Optional[Union[bytes, str]] = None,
cookies: Optional[Union[dict, List[dict]]] = None,
meta: Optional[dict] = 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,
) -> None:
self._encoding = encoding # this one has to be set first
self.method = str(method).upper()
self._set_url(url)
@ -64,23 +79,23 @@ class Request(object_ref):
self.flags = [] if flags is None else list(flags)
@property
def cb_kwargs(self):
def cb_kwargs(self) -> dict:
if self._cb_kwargs is None:
self._cb_kwargs = {}
return self._cb_kwargs
@property
def meta(self):
def meta(self) -> dict:
if self._meta is None:
self._meta = {}
return self._meta
def _get_url(self):
def _get_url(self) -> str:
return self._url
def _set_url(self, url):
def _set_url(self, url: str) -> None:
if not isinstance(url, str):
raise TypeError(f'Request url must be str or unicode, got {type(url).__name__}')
raise TypeError(f"Request url must be str, got {type(url).__name__}")
s = safe_url_string(url, self.encoding)
self._url = escape_ajax(s)
@ -94,31 +109,27 @@ class Request(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):
if body is None:
self._body = b''
else:
self._body = to_bytes(body, self.encoding)
def _set_body(self, body: Optional[Union[str, bytes]]) -> None:
self._body = b"" if body is None else to_bytes(body, self.encoding)
body = property(_get_body, obsolete_setter(_set_body, 'body'))
@property
def encoding(self):
def encoding(self) -> str:
return self._encoding
def __str__(self):
def __str__(self) -> str:
return f"<{self.method} {self.url}>"
__repr__ = __str__
def copy(self):
"""Return a copy of this Request"""
def copy(self) -> "Request":
return self.replace()
def replace(self, *args, **kwargs):
def replace(self, *args, **kwargs) -> "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))
@ -126,7 +137,9 @@ class Request(object_ref):
return cls(*args, **kwargs)
@classmethod
def from_curl(cls, curl_command, ignore_unknown_options=True, **kwargs):
def from_curl(
cls: Type[RequestTypeVar], curl_command: str, ignore_unknown_options: bool = True, **kwargs
) -> RequestTypeVar:
"""Create a Request object from a string containing a `cURL
<https://curl.haxx.se/>`_ command. It populates the HTTP method, the
URL, the headers, the cookies and the body. It accepts the same

84
scrapy/http/request/form.py
View File

@ -5,22 +5,28 @@ 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
from urllib.parse import urljoin, urlencode
import lxml.html
from lxml.html import FormElement, HtmlElement, HTMLParser, SelectElement
from parsel.selector import create_root_node
from w3lib.html import strip_html5_whitespace
from scrapy.http.request import Request
from scrapy.http.response.text import TextResponse
from scrapy.utils.python import to_bytes, is_listlike
from scrapy.utils.response import get_base_url
FormRequestTypeVar = TypeVar("FormRequestTypeVar", bound="FormRequest")
FormdataType = Optional[Union[dict, List[Tuple[str, str]]]]
class FormRequest(Request):
valid_form_methods = ['GET', 'POST']
def __init__(self, *args, **kwargs):
formdata = kwargs.pop('formdata', None)
def __init__(self, *args, formdata: FormdataType = None, **kwargs) -> None:
if formdata and kwargs.get('method') is None:
kwargs['method'] = 'POST'
@ -36,9 +42,19 @@ class FormRequest(Request):
self._set_url(self.url + ('&' if '?' in self.url else '?') + querystr)
@classmethod
def from_response(cls, response, formname=None, formid=None, formnumber=0, formdata=None,
clickdata=None, dont_click=False, formxpath=None, formcss=None, **kwargs):
def from_response(
cls: Type[FormRequestTypeVar],
response: TextResponse,
formname: Optional[str] = None,
formid: Optional[str] = None,
formnumber: Optional[int] = 0,
formdata: FormdataType = None,
clickdata: Optional[dict] = None,
dont_click: bool = False,
formxpath: Optional[str] = None,
formcss: Optional[str] = None,
**kwargs,
) -> FormRequestTypeVar:
kwargs.setdefault('encoding', response.encoding)
if formcss is not None:
@ -46,7 +62,7 @@ class FormRequest(Request):
formxpath = HTMLTranslator().css_to_xpath(formcss)
form = _get_form(response, formname, formid, formnumber, formxpath)
formdata = _get_inputs(form, formdata, dont_click, clickdata, response)
formdata = _get_inputs(form, formdata, dont_click, clickdata)
url = _get_form_url(form, kwargs.pop('url', None))
method = kwargs.pop('method', form.method)
@ -58,7 +74,7 @@ class FormRequest(Request):
return cls(url=url, method=method, formdata=formdata, **kwargs)
def _get_form_url(form, url):
def _get_form_url(form: FormElement, url: Optional[str]) -> str:
if url is None:
action = form.get('action')
if action is None:
@ -67,17 +83,22 @@ def _get_form_url(form, url):
return urljoin(form.base_url, url)
def _urlencode(seq, enc):
def _urlencode(seq: Iterable, enc: str) -> str:
values = [(to_bytes(k, enc), to_bytes(v, enc))
for k, vs in seq
for v in (vs if is_listlike(vs) else [vs])]
return urlencode(values, doseq=True)
def _get_form(response, formname, formid, formnumber, formxpath):
"""Find the form element """
root = create_root_node(response.text, lxml.html.HTMLParser,
base_url=get_base_url(response))
def _get_form(
response: TextResponse,
formname: Optional[str],
formid: Optional[str],
formnumber: Optional[int],
formxpath: Optional[str],
) -> FormElement:
"""Find the wanted form element within the given response."""
root = create_root_node(response.text, HTMLParser, base_url=get_base_url(response))
forms = root.xpath('//form')
if not forms:
raise ValueError(f"No <form> element found in {response}")
@ -105,8 +126,7 @@ def _get_form(response, formname, formid, formnumber, formxpath):
break
raise ValueError(f'No <form> element found with {formxpath}')
# If we get here, it means that either formname was None
# or invalid
# If we get here, it means that either formname was None or invalid
if formnumber is not None:
try:
form = forms[formnumber]
@ -116,25 +136,32 @@ def _get_form(response, formname, formid, formnumber, formxpath):
return form
def _get_inputs(form, formdata, dont_click, clickdata, response):
def _get_inputs(
form: FormElement,
formdata: FormdataType,
dont_click: bool,
clickdata: Optional[dict],
) -> List[Tuple[str, str]]:
"""Return a list of key-value pairs for the inputs found in the given form."""
try:
formdata_keys = dict(formdata or ()).keys()
except (ValueError, TypeError):
raise ValueError('formdata should be a dict or iterable of tuples')
if not formdata:
formdata = ()
formdata = []
inputs = form.xpath('descendant::textarea'
'|descendant::select'
'|descendant::input[not(@type) or @type['
' not(re:test(., "^(?:submit|image|reset)$", "i"))'
' and (../@checked or'
' not(re:test(., "^(?:checkbox|radio)$", "i")))]]',
namespaces={
"re": "http://exslt.org/regular-expressions"})
values = [(k, '' if v is None else v)
for k, v in (_value(e) for e in inputs)
if k and k not in formdata_keys]
namespaces={"re": "http://exslt.org/regular-expressions"})
values = [
(k, '' if v is None else v)
for k, v in (_value(e) for e in inputs)
if k and k not in formdata_keys
]
if not dont_click:
clickable = _get_clickable(clickdata, form)
@ -142,13 +169,13 @@ def _get_inputs(form, formdata, dont_click, clickdata, response):
values.append(clickable)
if isinstance(formdata, dict):
formdata = formdata.items()
formdata = formdata.items() # type: ignore[assignment]
values.extend((k, v) for k, v in formdata if v is not None)
return values
def _value(ele):
def _value(ele: HtmlElement):
n = ele.name
v = ele.value
if ele.tag == 'select':
@ -156,7 +183,7 @@ def _value(ele):
return n, v
def _select_value(ele, n, v):
def _select_value(ele: SelectElement, n: str, v: str):
multiple = ele.multiple
if v is None and not multiple:
# Match browser behaviour on simple select tag without options selected
@ -167,11 +194,12 @@ def _select_value(ele, n, v):
# This is a workround to bug in lxml fixed 2.3.1
# fix https://github.com/lxml/lxml/commit/57f49eed82068a20da3db8f1b18ae00c1bab8b12#L1L1139
selected_options = ele.xpath('.//option[@selected]')
v = [(o.get('value') or o.text or '').strip() for o in selected_options]
values = [(o.get('value') or o.text or '').strip() for o in selected_options]
return n, values
return n, v
def _get_clickable(clickdata, form):
def _get_clickable(clickdata: Optional[dict], form: FormElement) -> Optional[Tuple[str, str]]:
"""
Returns the clickable element specified in clickdata,
if the latter is given. If not, it returns the first
@ -183,7 +211,7 @@ def _get_clickable(clickdata, form):
namespaces={"re": "http://exslt.org/regular-expressions"}
))
if not clickables:
return
return None
# If we don't have clickdata, we just use the first clickable element
if clickdata is None:

15
scrapy/http/request/json_request.py
View File

@ -8,7 +8,7 @@ See documentation in docs/topics/request-response.rst
import copy
import json
import warnings
from typing import Tuple
from typing import Optional, Tuple
from scrapy.http.request import Request
from scrapy.utils.deprecate import create_deprecated_class
@ -18,8 +18,8 @@ class JsonRequest(Request):
attributes: Tuple[str, ...] = Request.attributes + ("dumps_kwargs",)
def __init__(self, *args, **kwargs):
dumps_kwargs = copy.deepcopy(kwargs.pop('dumps_kwargs', {}))
def __init__(self, *args, dumps_kwargs: Optional[dict] = None, **kwargs) -> 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
@ -29,10 +29,8 @@ class JsonRequest(Request):
if body_passed and data_passed:
warnings.warn('Both body and data passed. data will be ignored')
elif not body_passed and data_passed:
kwargs['body'] = self._dumps(data)
if 'method' not in kwargs:
kwargs['method'] = 'POST'
@ -41,23 +39,22 @@ class JsonRequest(Request):
self.headers.setdefault('Accept', 'application/json, text/javascript, */*; q=0.01')
@property
def dumps_kwargs(self):
def dumps_kwargs(self) -> dict:
return self._dumps_kwargs
def replace(self, *args, **kwargs):
def replace(self, *args, **kwargs) -> Request:
body_passed = kwargs.get('body', None) is not None
data = kwargs.pop('data', None)
data_passed = data is not None
if body_passed and data_passed:
warnings.warn('Both body and data passed. data will be ignored')
elif not body_passed and data_passed:
kwargs['body'] = self._dumps(data)
return super().replace(*args, **kwargs)
def _dumps(self, data):
def _dumps(self, data: dict) -> str:
"""Convert to JSON """
return json.dumps(data, **self._dumps_kwargs)

4
scrapy/http/request/rpc.py
View File

@ -5,6 +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 scrapy.http.request import Request
from scrapy.utils.python import get_func_args
@ -15,8 +16,7 @@ DUMPS_ARGS = get_func_args(xmlrpclib.dumps)
class XmlRpcRequest(Request):
def __init__(self, *args, **kwargs):
encoding = kwargs.get('encoding', None)
def __init__(self, *args, encoding: Optional[str] = None, **kwargs):
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)

2
scrapy/utils/curl.py
View File

@ -54,7 +54,7 @@ def _parse_headers_and_cookies(parsed_args):
return headers, cookies
def curl_to_request_kwargs(curl_command, ignore_unknown_options=True):
def curl_to_request_kwargs(curl_command: str, ignore_unknown_options: bool = True) -> dict:
"""Convert a cURL command syntax to Request kwargs.
:param str curl_command: string containing the curl command