diff --git a/scrapy/downloadermiddlewares/httpcompression.py b/scrapy/downloadermiddlewares/httpcompression.py index 8e170a1c7..f3647e05f 100644 --- a/scrapy/downloadermiddlewares/httpcompression.py +++ b/scrapy/downloadermiddlewares/httpcompression.py @@ -3,7 +3,7 @@ from __future__ import annotations import warnings from itertools import chain from logging import getLogger -from typing import TYPE_CHECKING, List, Optional, Tuple, Union +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union from scrapy import Request, Spider, signals from scrapy.crawler import Crawler @@ -138,12 +138,12 @@ class HttpCompressionMiddleware: respcls = responsetypes.from_args( headers=response.headers, url=response.url, body=decoded_body ) - kwargs = {"cls": respcls, "body": decoded_body} + kwargs: Dict[str, Any] = {"body": decoded_body} if issubclass(respcls, TextResponse): # force recalculating the encoding until we make sure the # responsetypes guessing is reliable kwargs["encoding"] = None - response = response.replace(**kwargs) + response = response.replace(cls=respcls, **kwargs) if not content_encoding: del response.headers["Content-Encoding"] diff --git a/scrapy/downloadermiddlewares/redirect.py b/scrapy/downloadermiddlewares/redirect.py index aa08827c4..371e2fd3b 100644 --- a/scrapy/downloadermiddlewares/redirect.py +++ b/scrapy/downloadermiddlewares/redirect.py @@ -27,6 +27,7 @@ def _build_redirect_request( redirect_request = source_request.replace( url=url, **kwargs, + cls=None, cookies=None, ) if "_scheme_proxy" in redirect_request.meta: diff --git a/scrapy/http/request/__init__.py b/scrapy/http/request/__init__.py index 77149333c..04589dd37 100644 --- a/scrapy/http/request/__init__.py +++ b/scrapy/http/request/__init__.py @@ -20,9 +20,11 @@ from typing import ( NoReturn, Optional, Tuple, + Type, TypedDict, + TypeVar, Union, - cast, + overload, ) from w3lib.url import safe_url_string @@ -50,6 +52,9 @@ class VerboseCookie(TypedDict): CookiesT = Union[Dict[str, str], List[VerboseCookie]] +RequestTypeVar = TypeVar("RequestTypeVar", bound="Request") + + 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 @@ -189,15 +194,26 @@ class Request(object_ref): def __repr__(self) -> str: return f"<{self.method} {self.url}>" - def copy(self) -> "Request": + def copy(self) -> Self: return self.replace() - def replace(self, *args: Any, **kwargs: Any) -> "Request": + @overload + def replace( + self, *args: Any, cls: Type[RequestTypeVar], **kwargs: Any + ) -> RequestTypeVar: ... + + @overload + def replace(self, *args: Any, cls: None = None, **kwargs: Any) -> Self: ... + + def replace( + self, *args: Any, cls: Optional[Type[Request]] = None, **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 cast(Request, cls(*args, **kwargs)) + if cls is None: + cls = self.__class__ + return cls(*args, **kwargs) @classmethod def from_curl( @@ -237,7 +253,7 @@ class Request(object_ref): request_kwargs.update(kwargs) return cls(**request_kwargs) - def to_dict(self, *, spider: Optional["scrapy.Spider"] = None) -> Dict[str, Any]: + 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. diff --git a/scrapy/http/request/json_request.py b/scrapy/http/request/json_request.py index 405c0b9d0..057a4f897 100644 --- a/scrapy/http/request/json_request.py +++ b/scrapy/http/request/json_request.py @@ -5,12 +5,18 @@ This module implements the JsonRequest class which is a more convenient class See documentation in docs/topics/request-response.rst """ +from __future__ import annotations + import copy import json import warnings -from typing import Any, Dict, Optional, Tuple +from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Type, overload -from scrapy.http.request import Request +from scrapy.http.request import Request, RequestTypeVar + +if TYPE_CHECKING: + # typing.Self requires Python 3.11 + from typing_extensions import Self class JsonRequest(Request): @@ -44,7 +50,17 @@ class JsonRequest(Request): def dumps_kwargs(self) -> Dict[str, Any]: return self._dumps_kwargs - def replace(self, *args: Any, **kwargs: Any) -> Request: + @overload + def replace( + self, *args: Any, cls: Type[RequestTypeVar], **kwargs: Any + ) -> RequestTypeVar: ... + + @overload + def replace(self, *args: Any, cls: None = None, **kwargs: Any) -> Self: ... + + def replace( + self, *args: Any, cls: Optional[Type[Request]] = None, **kwargs: Any + ) -> Request: body_passed = kwargs.get("body", None) is not None data: Any = kwargs.pop("data", None) data_passed: bool = data is not None @@ -54,7 +70,7 @@ class JsonRequest(Request): elif not body_passed and data_passed: kwargs["body"] = self._dumps(data) - return super().replace(*args, **kwargs) + return super().replace(*args, cls=cls, **kwargs) def _dumps(self, data: Any) -> str: """Convert to JSON""" diff --git a/scrapy/http/response/__init__.py b/scrapy/http/response/__init__.py index daf193f59..684439097 100644 --- a/scrapy/http/response/__init__.py +++ b/scrapy/http/response/__init__.py @@ -19,8 +19,10 @@ from typing import ( Mapping, Optional, Tuple, + Type, + TypeVar, Union, - cast, + overload, ) from urllib.parse import urljoin @@ -33,9 +35,15 @@ from scrapy.link import Link from scrapy.utils.trackref import object_ref if TYPE_CHECKING: + # typing.Self requires Python 3.11 + from typing_extensions import Self + from scrapy.selector import SelectorList +ResponseTypeVar = TypeVar("ResponseTypeVar", bound="Response") + + class Response(object_ref): """An object that represents an HTTP response, which is usually downloaded (by the Downloader) and fed to the Spiders for processing. @@ -132,16 +140,27 @@ class Response(object_ref): def __repr__(self) -> str: return f"<{self.status} {self.url}>" - def copy(self) -> Response: + def copy(self) -> Self: """Return a copy of this Response""" return self.replace() - def replace(self, *args: Any, **kwargs: Any) -> Response: + @overload + def replace( + self, *args: Any, cls: Type[ResponseTypeVar], **kwargs: Any + ) -> ResponseTypeVar: ... + + @overload + def replace(self, *args: Any, cls: None = None, **kwargs: Any) -> Self: ... + + def replace( + self, *args: Any, cls: Optional[Type[Response]] = None, **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 cast(Response, cls(*args, **kwargs)) + if cls is None: + cls = self.__class__ + return cls(*args, **kwargs) def urljoin(self, url: str) -> str: """Join this Response's url with a possible relative url to form an diff --git a/tests_typing/test_http_request.mypy-testing b/tests_typing/test_http_request.mypy-testing new file mode 100644 index 000000000..665db9088 --- /dev/null +++ b/tests_typing/test_http_request.mypy-testing @@ -0,0 +1,80 @@ +from typing import Any, Dict + +import pytest + +from scrapy import Request +from scrapy.http import JsonRequest + + +class MyRequest(Request): + pass + + +class MyRequest2(Request): + pass + + +@pytest.mark.mypy_testing +def mypy_test_headers(): + Request("data:,", headers=1) # E: Argument "headers" to "Request" has incompatible type "int"; expected "Union[Mapping[str, Any], Iterable[Tuple[str, Any]], None]" + Request("data:,", headers=None) + Request("data:,", headers={}) + Request("data:,", headers=[]) + Request("data:,", headers={"foo": "bar"}) + Request("data:,", headers={b"foo": "bar"}) + Request("data:,", headers={"foo": b"bar"}) + Request("data:,", headers=[("foo", "bar")]) + Request("data:,", headers=[(b"foo", "bar")]) + Request("data:,", headers=[("foo", b"bar")]) + + +@pytest.mark.mypy_testing +def mypy_test_copy(): + req = Request("data:,") + reveal_type(req) # R: scrapy.http.request.Request + req_copy = req.copy() + reveal_type(req_copy) # R: scrapy.http.request.Request + + +@pytest.mark.mypy_testing +def mypy_test_copy_subclass(): + req = MyRequest("data:,") + reveal_type(req) # R: __main__.MyRequest + req_copy = req.copy() + reveal_type(req_copy) # R: __main__.MyRequest + + +@pytest.mark.mypy_testing +def mypy_test_replace(): + req = Request("data:,") + reveal_type(req) # R: scrapy.http.request.Request + req_copy = req.replace(body=b"a") + reveal_type(req_copy) # R: scrapy.http.request.Request + kwargs: Dict[str, Any] = {} + req_copy2 = req.replace(body=b"a", **kwargs) + reveal_type(req_copy2) # R: Any + + +@pytest.mark.mypy_testing +def mypy_test_replace_subclass(): + req = MyRequest("data:,") + reveal_type(req) # R: __main__.MyRequest + req_copy = req.replace(body=b"a") + reveal_type(req_copy) # R: __main__.MyRequest + req_copy2 = req.replace(body=b"a", cls=MyRequest2) + reveal_type(req_copy2) # R: __main__.MyRequest2 + kwargs: Dict[str, Any] = {} + req_copy3 = req.replace(body=b"a", cls=MyRequest2, **kwargs) + reveal_type(req_copy3) # R: __main__.MyRequest2 + + +@pytest.mark.mypy_testing +def mypy_test_jsonrequest_copy_replace(): + req = JsonRequest("data:,") + reveal_type(req) # R: scrapy.http.request.json_request.JsonRequest + req_copy = req.copy() + reveal_type(req_copy) # R: scrapy.http.request.json_request.JsonRequest + req_copy = req.replace(body=b"a") + reveal_type(req_copy) # R: scrapy.http.request.json_request.JsonRequest + req_copy_my = req.replace(body=b"a", cls=MyRequest) + reveal_type(req_copy_my) # R: __main__.MyRequest diff --git a/tests_typing/test_http_response.mypy-testing b/tests_typing/test_http_response.mypy-testing new file mode 100644 index 000000000..d58ac1027 --- /dev/null +++ b/tests_typing/test_http_response.mypy-testing @@ -0,0 +1,59 @@ +from typing import Any, Dict + +import pytest + +from scrapy.http import HtmlResponse, Response, TextResponse + + +@pytest.mark.mypy_testing +def mypy_test_headers(): + Response("data:,", headers=1) # E: Argument "headers" to "Response" has incompatible type "int"; expected "Union[Mapping[str, Any], Iterable[Tuple[str, Any]], None]" + Response("data:,", headers=None) + Response("data:,", headers={}) + Response("data:,", headers=[]) + Response("data:,", headers={"foo": "bar"}) + Response("data:,", headers={b"foo": "bar"}) + Response("data:,", headers={"foo": b"bar"}) + Response("data:,", headers=[("foo", "bar")]) + Response("data:,", headers=[(b"foo", "bar")]) + Response("data:,", headers=[("foo", b"bar")]) + + +@pytest.mark.mypy_testing +def mypy_test_copy(): + resp = Response("data:,") + reveal_type(resp) # R: scrapy.http.response.Response + resp_copy = resp.copy() + reveal_type(resp_copy) # R: scrapy.http.response.Response + + +@pytest.mark.mypy_testing +def mypy_test_copy_subclass(): + resp = HtmlResponse("data:,") + reveal_type(resp) # R: scrapy.http.response.html.HtmlResponse + resp_copy = resp.copy() + reveal_type(resp_copy) # R: scrapy.http.response.html.HtmlResponse + + +@pytest.mark.mypy_testing +def mypy_test_replace(): + resp = Response("data:,") + reveal_type(resp) # R: scrapy.http.response.Response + resp_copy = resp.replace(body=b"a") + reveal_type(resp_copy) # R: scrapy.http.response.Response + kwargs: Dict[str, Any] = {} + resp_copy2 = resp.replace(body=b"a", **kwargs) + reveal_type(resp_copy2) # R: Any + + +@pytest.mark.mypy_testing +def mypy_test_replace_subclass(): + resp = HtmlResponse("data:,") + reveal_type(resp) # R: scrapy.http.response.html.HtmlResponse + resp_copy = resp.replace(body=b"a") + reveal_type(resp_copy) # R: scrapy.http.response.html.HtmlResponse + resp_copy2 = resp.replace(body=b"a", cls=TextResponse) + reveal_type(resp_copy2) # R: scrapy.http.response.text.TextResponse + kwargs: Dict[str, Any] = {} + resp_copy3 = resp.replace(body=b"a", cls=TextResponse, **kwargs) + reveal_type(resp_copy3) # R: scrapy.http.response.text.TextResponse