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

Add parameters to typing.Dict.

This commit is contained in:
Andrey Rakhmatullin 2024-05-31 21:11:50 +05:00
parent b4293e8f9e
commit da42e8f124
17 changed files with 85 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
scrapy/core/downloader/handlers/__init__.py
View File

@ -27,7 +27,10 @@ class DownloadHandlers:
self._handlers: Dict[str, Any] = {} # stores instanced handlers for schemes
self._notconfigured: Dict[str, str] = {} # remembers failed handlers
handlers: Dict[str, Union[str, Callable]] = without_none_values(
crawler.settings.getwithbase("DOWNLOAD_HANDLERS")
cast(
Dict[str, Union[str, Callable]],
crawler.settings.getwithbase("DOWNLOAD_HANDLERS"),
)
)
for scheme, clspath in handlers.items():
self._schemes[scheme] = clspath

4
scrapy/core/http2/protocol.py
View File

@ -3,7 +3,7 @@ import itertools
import logging
from collections import deque
from ipaddress import IPv4Address, IPv6Address
from typing import Dict, List, Optional, Union
from typing import Any, Dict, List, Optional, Union
from h2.config import H2Configuration
from h2.connection import H2Connection
@ -115,7 +115,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin):
# Some meta data of this connection
# initialized when connection is successfully made
self.metadata: Dict = {
self.metadata: Dict[str, Any] = {
# Peer certificate instance
"certificate": None,
# Address of the server we are connected to which

4
scrapy/core/http2/stream.py
View File

@ -110,7 +110,7 @@ class Stream:
# Metadata of an HTTP/2 connection stream
# initialized when stream is instantiated
self.metadata: Dict = {
self.metadata: Dict[str, Any] = {
"request_content_length": (
0 if self._request.body is None else len(self._request.body)
),
@ -131,7 +131,7 @@ class Stream:
# Private variable used to build the response
# this response is then converted to appropriate Response class
# passed to the response deferred callback
self._response: Dict = {
self._response: Dict[str, Any] = {
# Data received frame by frame from the server is appended
# and passed to the response Deferred when completely received.
"body": BytesIO(),

4
scrapy/extensions/feedexport.py
View File

@ -694,7 +694,9 @@ class FeedExporter:
self.slots = slots
def _load_components(self, setting_prefix: str) -> Dict[str, Any]:
conf = without_none_values(self.settings.getwithbase(setting_prefix))
conf = without_none_values(
cast(Dict[str, str], self.settings.getwithbase(setting_prefix))
)
d = {}
for k, v in conf.items():
try:

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

@ -97,7 +97,7 @@ class Request(object_ref):
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,
cookies: Optional[Union[Dict[str, str], List[Dict[str, str]]]] = None,
meta: Optional[Dict[str, Any]] = None,
encoding: str = "utf-8",
priority: int = 0,
@ -123,7 +123,7 @@ class Request(object_ref):
self.callback: Optional[Callable] = callback
self.errback: Optional[Callable] = errback
self.cookies: Union[dict, List[dict]] = cookies or {}
self.cookies: Union[Dict[str, str], List[Dict[str, str]]] = cookies or {}
self.headers: Headers = Headers(headers or {}, encoding=encoding)
self.dont_filter: bool = dont_filter

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

@ -7,7 +7,17 @@ See documentation in docs/topics/request-response.rst
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Iterable, List, Optional, Tuple, Union, cast
from typing import (
TYPE_CHECKING,
Any,
Dict,
Iterable,
List,
Optional,
Tuple,
Union,
cast,
)
from urllib.parse import urlencode, urljoin, urlsplit, urlunsplit
from lxml.html import FormElement # nosec
@ -26,8 +36,9 @@ if TYPE_CHECKING:
from typing_extensions import Self
FormdataKVType = Tuple[str, Union[str, Iterable[str]]]
FormdataType = Optional[Union[dict, List[FormdataKVType]]]
FormdataVType = Union[str, Iterable[str]]
FormdataKVType = Tuple[str, FormdataVType]
FormdataType = Optional[Union[Dict[str, FormdataVType], List[FormdataKVType]]]
class FormRequest(Request):
@ -62,7 +73,7 @@ class FormRequest(Request):
formid: Optional[str] = None,
formnumber: int = 0,
formdata: FormdataType = None,
clickdata: Optional[dict] = None,
clickdata: Optional[Dict[str, Union[str, int]]] = None,
dont_click: bool = False,
formxpath: Optional[str] = None,
formcss: Optional[str] = None,
@ -156,7 +167,7 @@ def _get_inputs(
form: FormElement,
formdata: FormdataType,
dont_click: bool,
clickdata: Optional[dict],
clickdata: Optional[Dict[str, Union[str, int]]],
) -> List[FormdataKVType]:
"""Return a list of key-value pairs for the inputs found in the given form."""
try:
@ -186,10 +197,8 @@ def _get_inputs(
if clickable and clickable[0] not in formdata and not clickable[0] is None:
values.append(clickable)
if isinstance(formdata, dict):
formdata = formdata.items() # type: ignore[assignment]
values.extend((k, v) for k, v in formdata if v is not None)
formdata_items = formdata.items() if isinstance(formdata, dict) else formdata
values.extend((k, v) for k, v in formdata_items if v is not None)
return values
@ -216,7 +225,7 @@ def _select_value(
def _get_clickable(
clickdata: Optional[dict], form: FormElement
clickdata: Optional[Dict[str, Union[str, int]]], form: FormElement
) -> Optional[Tuple[str, str]]:
"""
Returns the clickable element specified in clickdata,
@ -243,6 +252,7 @@ def _get_clickable(
# because that uniquely identifies the element
nr = clickdata.get("nr", None)
if nr is not None:
assert isinstance(nr, int)
try:
el = list(form.inputs)[nr]
except IndexError:

18
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 Any, Optional, Tuple
from typing import Any, Dict, Optional, Tuple
from scrapy.http.request import Request
@ -17,15 +17,15 @@ class JsonRequest(Request):
attributes: Tuple[str, ...] = Request.attributes + ("dumps_kwargs",)
def __init__(
self, *args: Any, dumps_kwargs: Optional[dict] = None, **kwargs: Any
self, *args: Any, dumps_kwargs: Optional[Dict[str, Any]] = 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
self._dumps_kwargs: Dict[str, Any] = dumps_kwargs
body_passed = kwargs.get("body", None) is not None
data = kwargs.pop("data", None)
data_passed = data is not None
data: Any = kwargs.pop("data", None)
data_passed: bool = data is not None
if body_passed and data_passed:
warnings.warn("Both body and data passed. data will be ignored")
@ -41,13 +41,13 @@ class JsonRequest(Request):
)
@property
def dumps_kwargs(self) -> dict:
def dumps_kwargs(self) -> Dict[str, Any]:
return self._dumps_kwargs
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
data: Any = kwargs.pop("data", None)
data_passed: bool = data is not None
if body_passed and data_passed:
warnings.warn("Both body and data passed. data will be ignored")
@ -56,6 +56,6 @@ class JsonRequest(Request):
return super().replace(*args, **kwargs)
def _dumps(self, data: dict) -> str:
def _dumps(self, data: Any) -> str:
"""Convert to JSON"""
return json.dumps(data, **self._dumps_kwargs)

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

@ -181,7 +181,7 @@ class Response(object_ref):
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,
cookies: Optional[Union[Dict[str, str], List[Dict[str, str]]]] = None,
meta: Optional[Dict[str, Any]] = None,
encoding: Optional[str] = "utf-8",
priority: int = 0,
@ -234,7 +234,7 @@ class Response(object_ref):
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,
cookies: Optional[Union[Dict[str, str], List[Dict[str, str]]]] = None,
meta: Optional[Dict[str, Any]] = None,
encoding: Optional[str] = "utf-8",
priority: int = 0,

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

@ -183,7 +183,7 @@ class TextResponse(Response):
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,
cookies: Optional[Union[Dict[str, str], List[Dict[str, str]]]] = None,
meta: Optional[Dict[str, Any]] = None,
encoding: Optional[str] = None,
priority: int = 0,
@ -236,7 +236,7 @@ class TextResponse(Response):
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,
cookies: Optional[Union[Dict[str, str], List[Dict[str, str]]]] = None,
meta: Optional[Dict[str, Any]] = None,
encoding: Optional[str] = None,
priority: int = 0,

2
scrapy/item.py
View File

@ -27,7 +27,7 @@ if TYPE_CHECKING:
from typing_extensions import Self
class Field(dict):
class Field(Dict[str, Any]):
"""Container of field metadata"""

22
scrapy/logformatter.py
View File

@ -2,7 +2,7 @@ from __future__ import annotations
import logging
import os
from typing import TYPE_CHECKING, Any, Dict, Optional, Union
from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, TypedDict, Union
from twisted.python.failure import Failure
@ -26,6 +26,12 @@ DOWNLOADERRORMSG_SHORT = "Error downloading %(request)s"
DOWNLOADERRORMSG_LONG = "Error downloading %(request)s: %(errmsg)s"
class LogFormatterResult(TypedDict):
level: int
msg: str
args: Union[Dict[str, Any], Tuple[Any, ...]]
class LogFormatter:
"""Class for generating log messages for different actions.
@ -64,7 +70,9 @@ class LogFormatter:
}
"""
def crawled(self, request: Request, response: Response, spider: Spider) -> dict:
def crawled(
self, request: Request, response: Response, spider: Spider
) -> LogFormatterResult:
"""Logs a message when the crawler finds a webpage."""
request_flags = f" {str(request.flags)}" if request.flags else ""
response_flags = f" {str(response.flags)}" if response.flags else ""
@ -84,7 +92,7 @@ class LogFormatter:
def scraped(
self, item: Any, response: Union[Response, Failure], spider: Spider
) -> dict:
) -> LogFormatterResult:
"""Logs a message when an item is scraped by a spider."""
src: Any
if isinstance(response, Failure):
@ -102,7 +110,7 @@ class LogFormatter:
def dropped(
self, item: Any, exception: BaseException, response: Response, spider: Spider
) -> dict:
) -> LogFormatterResult:
"""Logs a message when an item is dropped while it is passing through the item pipeline."""
return {
"level": logging.WARNING,
@ -115,7 +123,7 @@ class LogFormatter:
def item_error(
self, item: Any, exception: BaseException, response: Response, spider: Spider
) -> dict:
) -> LogFormatterResult:
"""Logs a message when an item causes an error while it is passing
through the item pipeline.
@ -135,7 +143,7 @@ class LogFormatter:
request: Request,
response: Union[Response, Failure],
spider: Spider,
) -> dict:
) -> LogFormatterResult:
"""Logs an error message from a spider.
.. versionadded:: 2.0
@ -155,7 +163,7 @@ class LogFormatter:
request: Request,
spider: Spider,
errmsg: Optional[str] = None,
) -> dict:
) -> LogFormatterResult:
"""Logs a download error message from a spider (typically coming from
the engine).

2
scrapy/settings/__init__.py
View File

@ -411,7 +411,7 @@ class BaseSettings(MutableMapping[_SettingsKeyT, Any]):
"""
self._assert_mutability()
if isinstance(values, str):
values = cast(dict, json.loads(values))
values = cast(Dict[_SettingsKeyT, Any], json.loads(values))
if values is not None:
if isinstance(values, BaseSettings):
for name, value in values.items():

6
scrapy/spiders/__init__.py
View File

@ -7,7 +7,7 @@ See documentation in docs/topics/spiders.rst
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any, Iterable, List, Optional, Union, cast
from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Union, cast
from twisted.internet.defer import Deferred
@ -24,7 +24,7 @@ if TYPE_CHECKING:
from typing_extensions import Concatenate, Self
from scrapy.crawler import Crawler
from scrapy.settings import BaseSettings
from scrapy.settings import BaseSettings, _SettingsKeyT
from scrapy.utils.log import SpiderLoggerAdapter
CallbackT = Callable[Concatenate[Response, ...], Any]
@ -36,7 +36,7 @@ class Spider(object_ref):
"""
name: str
custom_settings: Optional[dict] = None
custom_settings: Optional[Dict[_SettingsKeyT, Any]] = None
def __init__(self, name: Optional[str] = None, **kwargs: Any):
if name is not None:

3
scrapy/utils/conf.py
View File

@ -16,6 +16,7 @@ from typing import (
MutableMapping,
Optional,
Union,
cast,
)
from scrapy.exceptions import ScrapyDeprecationWarning, UsageError
@ -173,7 +174,7 @@ def feed_process_params_from_cli(
suitable to be used as the FEEDS setting.
"""
valid_output_formats: Iterable[str] = without_none_values(
settings.getwithbase("FEED_EXPORTERS")
cast(Dict[str, str], settings.getwithbase("FEED_EXPORTERS"))
).keys()
def check_valid_format(output_format: str) -> None:

13
scrapy/utils/log.py
View File

@ -7,6 +7,7 @@ from types import TracebackType
from typing import (
TYPE_CHECKING,
Any,
Dict,
List,
MutableMapping,
Optional,
@ -20,7 +21,8 @@ from twisted.python import log as twisted_log
from twisted.python.failure import Failure
import scrapy
from scrapy.settings import Settings
from scrapy.logformatter import LogFormatterResult
from scrapy.settings import Settings, _SettingsKeyT
from scrapy.utils.versions import scrapy_components_versions
if TYPE_CHECKING:
@ -86,7 +88,8 @@ DEFAULT_LOGGING = {
def configure_logging(
settings: Union[Settings, dict, None] = None, install_root_handler: bool = True
settings: Union[Settings, Dict[_SettingsKeyT, Any], None] = None,
install_root_handler: bool = True,
) -> None:
"""
Initialize logging defaults for Scrapy.
@ -234,7 +237,9 @@ class LogCounterHandler(logging.Handler):
self.crawler.stats.inc_value(sname)
def logformatter_adapter(logkws: dict) -> Tuple[int, str, dict]:
def logformatter_adapter(
logkws: LogFormatterResult,
) -> Tuple[int, str, Union[Dict[str, Any], Tuple[Any, ...]]]:
"""
Helper that takes the dictionary output from the methods in LogFormatter
and adapts it into a tuple of positional arguments for logger.log calls,
@ -245,7 +250,7 @@ def logformatter_adapter(logkws: dict) -> Tuple[int, str, dict]:
message = logkws.get("msg") or ""
# NOTE: This also handles 'args' being an empty dict, that case doesn't
# play well in logger.log calls
args = logkws if not logkws.get("args") else logkws["args"]
args = cast(Dict[str, Any], logkws) if not logkws.get("args") else logkws["args"]
return (level, message, args)

10
scrapy/utils/python.py
View File

@ -42,6 +42,8 @@ if TYPE_CHECKING:
_P = ParamSpec("_P")
_T = TypeVar("_T")
_KT = TypeVar("_KT")
_VT = TypeVar("_VT")
def flatten(x: Iterable) -> list:
@ -303,14 +305,16 @@ def equal_attributes(
@overload
def without_none_values(iterable: Mapping) -> dict: ...
def without_none_values(iterable: Mapping[_KT, _VT]) -> Dict[_KT, _VT]: ...
@overload
def without_none_values(iterable: Iterable) -> Iterable: ...
def without_none_values(iterable: Iterable[_KT]) -> Iterable[_KT]: ...
def without_none_values(iterable: Union[Mapping, Iterable]) -> Union[dict, Iterable]:
def without_none_values(
iterable: Union[Mapping[_KT, _VT], Iterable[_KT]]
) -> Union[Dict[_KT, _VT], Iterable[_KT]]:
"""Return a copy of ``iterable`` with all ``None`` entries removed.
If ``iterable`` is a mapping, return a dictionary where all pairs that have

2
scrapy/utils/request.py
View File

@ -197,7 +197,7 @@ def referer_str(request: Request) -> Optional[str]:
return to_unicode(referrer, errors="replace")
def request_from_dict(d: dict, *, spider: Optional[Spider] = None) -> Request:
def request_from_dict(d: Dict[str, Any], *, spider: Optional[Spider] = None) -> Request:
"""Create a :class:`~scrapy.Request` object from a dict.
If a spider is given, it will try to resolve the callbacks looking at the