Request type hints

This commit is contained in:
Eugenio Lacuesta 2021-05-01 16:34:39 -03:00
parent e27eff47ac
commit 19c7415aae
No known key found for this signature in database
GPG Key ID: DA3EF2D0913E9810
3 changed files with 42 additions and 31 deletions

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:

View File

@ -4,22 +4,38 @@ requests in Scrapy.
See documentation in docs/topics/request-response.rst
"""
from typing import Callable, List, Optional, Type, TypeVar, Union
from w3lib.url import safe_url_string
from scrapy.http.common import obsolete_setter
from scrapy.http.headers import Headers
from scrapy.utils.curl import curl_to_request_kwargs
from scrapy.utils.python import to_bytes
from scrapy.utils.trackref import object_ref
from scrapy.utils.url import escape_ajax
from scrapy.http.common import obsolete_setter
from scrapy.utils.curl import curl_to_request_kwargs
RequestTypeVar = TypeVar("RequestTypeVar", bound="Request")
class Request(object_ref):
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)
@ -44,23 +60,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)
@ -74,34 +90,28 @@ 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) -> RequestTypeVar:
return self.replace()
def replace(self, *args, **kwargs):
"""Create a new Request with the same attributes except for those
given new values.
"""
def replace(self, *args, **kwargs) -> RequestTypeVar:
"""Create a new Request with the same attributes except for those given new values"""
for x in ['url', 'method', 'headers', 'body', 'cookies', 'meta', 'flags',
'encoding', 'priority', 'dont_filter', 'callback', 'errback', 'cb_kwargs']:
kwargs.setdefault(x, getattr(self, x))
@ -109,7 +119,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
@ -136,8 +148,7 @@ class Request(object_ref):
To translate a cURL command into a Scrapy request,
you may use `curl2scrapy <https://michael-shub.github.io/curl2scrapy/>`_.
"""
"""
request_kwargs = curl_to_request_kwargs(curl_command, ignore_unknown_options)
request_kwargs.update(kwargs)
return cls(**request_kwargs)

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