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

Merge pull request #6385 from wRAR/typing-generics-collections 

Add parameters to various generics.
This commit is contained in:
Adrián Chaves 2024-06-03 13:29:38 +02:00 committed by GitHub
parent 492c3bce9d 019f23e3b7
commit 2b9e32f1ca
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
30 changed files with 198 additions and 131 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,5 +1,7 @@
"""Download handlers for different schemes"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any, Callable, Dict, Generator, Union, cast
@ -19,15 +21,18 @@ logger = logging.getLogger(__name__)
class DownloadHandlers:
def __init__(self, crawler: "Crawler"):
self._crawler: "Crawler" = crawler
self._schemes: Dict[str, Union[str, Callable]] = (
def __init__(self, crawler: Crawler):
self._crawler: Crawler = crawler
self._schemes: Dict[str, Union[str, Callable[..., Any]]] = (
{}
) # stores acceptable schemes on instancing
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")
handlers: Dict[str, Union[str, Callable[..., Any]]] = without_none_values(
cast(
Dict[str, Union[str, Callable[..., Any]]],
crawler.settings.getwithbase("DOWNLOAD_HANDLERS"),
)
)
for scheme, clspath in handlers.items():
self._schemes[scheme] = clspath

30
scrapy/core/engine.py
View File

@ -5,6 +5,8 @@ For more information see docs/topics/architecture.rst
"""
from __future__ import annotations
import logging
from time import time
from typing import (
@ -51,15 +53,15 @@ class Slot:
self,
start_requests: Iterable[Request],
close_if_idle: bool,
nextcall: CallLaterOnce,
scheduler: "BaseScheduler",
nextcall: CallLaterOnce[None],
scheduler: BaseScheduler,
) -> None:
self.closing: Optional[Deferred] = None
self.inprogress: Set[Request] = set()
self.start_requests: Optional[Iterator[Request]] = iter(start_requests)
self.close_if_idle: bool = close_if_idle
self.nextcall: CallLaterOnce = nextcall
self.scheduler: "BaseScheduler" = scheduler
self.nextcall: CallLaterOnce[None] = nextcall
self.scheduler: BaseScheduler = scheduler
self.heartbeat: LoopingCall = LoopingCall(nextcall.schedule)
def add_request(self, request: Request) -> None:
@ -84,8 +86,12 @@ class Slot:
class ExecutionEngine:
def __init__(self, crawler: "Crawler", spider_closed_callback: Callable) -> None:
self.crawler: "Crawler" = crawler
def __init__(
self,
crawler: Crawler,
spider_closed_callback: Callable[[Spider], Optional[Deferred[None]]],
) -> None:
self.crawler: Crawler = crawler
self.settings: Settings = crawler.settings
self.signals: SignalManager = crawler.signals
assert crawler.logformatter
@ -94,19 +100,21 @@ class ExecutionEngine:
self.spider: Optional[Spider] = None
self.running: bool = False
self.paused: bool = False
self.scheduler_cls: Type["BaseScheduler"] = self._get_scheduler_class(
self.scheduler_cls: Type[BaseScheduler] = self._get_scheduler_class(
crawler.settings
)
downloader_cls: Type[Downloader] = load_object(self.settings["DOWNLOADER"])
self.downloader: Downloader = downloader_cls(crawler)
self.scraper = Scraper(crawler)
self._spider_closed_callback: Callable = spider_closed_callback
self._spider_closed_callback: Callable[[Spider], Optional[Deferred[None]]] = (
spider_closed_callback
)
self.start_time: Optional[float] = None
def _get_scheduler_class(self, settings: BaseSettings) -> Type["BaseScheduler"]:
def _get_scheduler_class(self, settings: BaseSettings) -> Type[BaseScheduler]:
from scrapy.core.scheduler import BaseScheduler
scheduler_cls: Type = load_object(settings["SCHEDULER"])
scheduler_cls: Type[BaseScheduler] = load_object(settings["SCHEDULER"])
if not issubclass(scheduler_cls, BaseScheduler):
raise TypeError(
f"The provided scheduler class ({settings['SCHEDULER']})"
@ -425,7 +433,7 @@ class ExecutionEngine:
dfd = self.slot.close()
def log_failure(msg: str) -> Callable:
def log_failure(msg: str) -> Callable[[Failure], None]:
def errback(failure: Failure) -> None:
logger.error(
msg, exc_info=failure_to_exc_info(failure), extra={"spider": spider}

24
scrapy/core/http2/agent.py
View File

@ -20,6 +20,8 @@ from scrapy.http.request import Request
from scrapy.settings import Settings
from scrapy.spiders import Spider
ConnectionKeyT = Tuple[bytes, bytes, int]
class H2ConnectionPool:
def __init__(self, reactor: ReactorBase, settings: Settings) -> None:
@ -28,13 +30,13 @@ class H2ConnectionPool:
# Store a dictionary which is used to get the respective
# H2ClientProtocolInstance using the key as Tuple(scheme, hostname, port)
self._connections: Dict[Tuple, H2ClientProtocol] = {}
self._connections: Dict[ConnectionKeyT, H2ClientProtocol] = {}
# Save all requests that arrive before the connection is established
self._pending_requests: Dict[Tuple, Deque[Deferred]] = {}
self._pending_requests: Dict[ConnectionKeyT, Deque[Deferred]] = {}
def get_connection(
self, key: Tuple, uri: URI, endpoint: HostnameEndpoint
self, key: ConnectionKeyT, uri: URI, endpoint: HostnameEndpoint
) -> Deferred:
if key in self._pending_requests:
# Received a request while connecting to remote
@ -54,7 +56,7 @@ class H2ConnectionPool:
return self._new_connection(key, uri, endpoint)
def _new_connection(
self, key: Tuple, uri: URI, endpoint: HostnameEndpoint
self, key: ConnectionKeyT, uri: URI, endpoint: HostnameEndpoint
) -> Deferred:
self._pending_requests[key] = deque()
@ -69,7 +71,9 @@ class H2ConnectionPool:
self._pending_requests[key].append(d)
return d
def put_connection(self, conn: H2ClientProtocol, key: Tuple) -> H2ClientProtocol:
def put_connection(
self, conn: H2ClientProtocol, key: ConnectionKeyT
) -> H2ClientProtocol:
self._connections[key] = conn
# Now as we have established a proper HTTP/2 connection
@ -81,7 +85,9 @@ class H2ConnectionPool:
return conn
def _remove_connection(self, errors: List[BaseException], key: Tuple) -> None:
def _remove_connection(
self, errors: List[BaseException], key: ConnectionKeyT
) -> None:
self._connections.pop(key)
# Call the errback of all the pending requests for this connection
@ -122,7 +128,7 @@ class H2Agent:
def get_endpoint(self, uri: URI) -> HostnameEndpoint:
return self.endpoint_factory.endpointForURI(uri)
def get_key(self, uri: URI) -> Tuple:
def get_key(self, uri: URI) -> ConnectionKeyT:
"""
Arguments:
uri - URI obtained directly from request URL
@ -164,6 +170,6 @@ class ScrapyProxyH2Agent(H2Agent):
def get_endpoint(self, uri: URI) -> HostnameEndpoint:
return self.endpoint_factory.endpointForURI(self._proxy_uri)
def get_key(self, uri: URI) -> Tuple:
def get_key(self, uri: URI) -> ConnectionKeyT:
"""We use the proxy uri instead of uri obtained from request url"""
return "http-proxy", self._proxy_uri.host, self._proxy_uri.port
return b"http-proxy", self._proxy_uri.host, self._proxy_uri.port

6
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, Deque, Dict, List, Optional, Union
from h2.config import H2Configuration
from h2.connection import H2Connection
@ -107,7 +107,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin):
# If requests are received before connection is made we keep
# all requests in a pool and send them as the connection is made
self._pending_request_stream_pool: deque = deque()
self._pending_request_stream_pool: Deque[Stream] = deque()
# Save an instance of errors raised which lead to losing the connection
# We pass these instances to the streams ResponseFailed() failure
@ -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(),

8
scrapy/core/scheduler.py
View File

@ -4,7 +4,7 @@ import json
import logging
from abc import abstractmethod
from pathlib import Path
from typing import TYPE_CHECKING, Any, Optional, Type, cast
from typing import TYPE_CHECKING, Any, List, Optional, Type, cast
from twisted.internet.defer import Deferred
@ -362,13 +362,13 @@ class Scheduler(BaseScheduler):
return str(dqdir)
return None
def _read_dqs_state(self, dqdir: str) -> list:
def _read_dqs_state(self, dqdir: str) -> List[int]:
path = Path(dqdir, "active.json")
if not path.exists():
return []
with path.open(encoding="utf-8") as f:
return cast(list, json.load(f))
return cast(List[int], json.load(f))
def _write_dqs_state(self, dqdir: str, state: list) -> None:
def _write_dqs_state(self, dqdir: str, state: List[int]) -> None:
with Path(dqdir, "active.json").open("w", encoding="utf-8") as f:
json.dump(state, f)

25
scrapy/downloadermiddlewares/cookies.py
View File

@ -3,16 +3,7 @@ from __future__ import annotations
import logging
from collections import defaultdict
from http.cookiejar import Cookie
from typing import (
TYPE_CHECKING,
Any,
DefaultDict,
Dict,
Iterable,
Optional,
Sequence,
Union,
)
from typing import TYPE_CHECKING, Any, DefaultDict, Iterable, Optional, Sequence, Union
from tldextract import TLDExtract
@ -21,6 +12,7 @@ from scrapy.crawler import Crawler
from scrapy.exceptions import NotConfigured
from scrapy.http import Response
from scrapy.http.cookies import CookieJar
from scrapy.http.request import VerboseCookie
from scrapy.utils.httpobj import urlparse_cached
from scrapy.utils.python import to_unicode
@ -128,7 +120,7 @@ class CookiesMiddleware:
msg = f"Received cookies from: {response}\n{cookies}"
logger.debug(msg, extra={"spider": spider})
def _format_cookie(self, cookie: Dict[str, Any], request: Request) -> Optional[str]:
def _format_cookie(self, cookie: VerboseCookie, request: Request) -> Optional[str]:
"""
Given a dict consisting of cookie components, return its string representation.
Decode from bytes if necessary.
@ -142,18 +134,19 @@ class CookiesMiddleware:
logger.warning(msg)
return None
continue
if isinstance(cookie[key], (bool, float, int, str)):
decoded[key] = str(cookie[key])
# https://github.com/python/mypy/issues/7178, https://github.com/python/mypy/issues/9168
if isinstance(cookie[key], (bool, float, int, str)): # type: ignore[literal-required]
decoded[key] = str(cookie[key]) # type: ignore[literal-required]
else:
try:
decoded[key] = cookie[key].decode("utf8")
decoded[key] = cookie[key].decode("utf8") # type: ignore[literal-required]
except UnicodeDecodeError:
logger.warning(
"Non UTF-8 encoded cookie found in request %s: %s",
request,
cookie,
)
decoded[key] = cookie[key].decode("latin1", errors="replace")
decoded[key] = cookie[key].decode("latin1", errors="replace") # type: ignore[literal-required]
for flag in ("secure",):
value = cookie.get(flag, _UNSET)
if value is _UNSET or not value:
@ -174,7 +167,7 @@ class CookiesMiddleware:
"""
if not request.cookies:
return []
cookies: Iterable[Dict[str, Any]]
cookies: Iterable[VerboseCookie]
if isinstance(request.cookies, dict):
cookies = tuple({"name": k, "value": v} for k, v in request.cookies.items())
else:

6
scrapy/downloadermiddlewares/stats.py
View File

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Dict, Union
from typing import TYPE_CHECKING, Dict, List, Tuple, Union
from twisted.web import http
@ -17,7 +17,9 @@ if TYPE_CHECKING:
from typing_extensions import Self
def get_header_size(headers: Dict[str, Union[list, tuple]]) -> int:
def get_header_size(
headers: Dict[str, Union[List[Union[str, bytes]], Tuple[Union[str, bytes], ...]]]
) -> int:
size = 0
for key, value in headers.items():
if isinstance(value, (list, tuple)):

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:

9
scrapy/extensions/httpcache.py
View File

@ -315,7 +315,9 @@ class FilesystemCacheStorage:
self.expiration_secs: int = settings.getint("HTTPCACHE_EXPIRATION_SECS")
self.use_gzip: bool = settings.getbool("HTTPCACHE_GZIP")
# https://github.com/python/mypy/issues/10740
self._open: Callable[Concatenate[Union[str, os.PathLike], str, ...], IO] = (
self._open: Callable[
Concatenate[Union[str, os.PathLike], str, ...], IO[bytes]
] = (
gzip.open if self.use_gzip else open # type: ignore[assignment]
)
@ -368,11 +370,12 @@ class FilesystemCacheStorage:
with self._open(rpath / "pickled_meta", "wb") as f:
pickle.dump(metadata, f, protocol=4)
with self._open(rpath / "response_headers", "wb") as f:
f.write(headers_dict_to_raw(response.headers))
# headers_dict_to_raw() needs a better type hint
f.write(cast(bytes, headers_dict_to_raw(response.headers)))
with self._open(rpath / "response_body", "wb") as f:
f.write(response.body)
with self._open(rpath / "request_headers", "wb") as f:
f.write(headers_dict_to_raw(request.headers))
f.write(cast(bytes, headers_dict_to_raw(request.headers)))
with self._open(rpath / "request_body", "wb") as f:
f.write(request.body)

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

@ -20,6 +20,7 @@ from typing import (
NoReturn,
Optional,
Tuple,
TypedDict,
Union,
cast,
)
@ -34,8 +35,19 @@ from scrapy.utils.trackref import object_ref
from scrapy.utils.url import escape_ajax
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
# typing.NotRequired and typing.Self require Python 3.11
from typing_extensions import NotRequired, Self
class VerboseCookie(TypedDict):
name: str
value: str
domain: NotRequired[str]
path: NotRequired[str]
secure: NotRequired[bool]
CookiesT = Union[Dict[str, str], List[VerboseCookie]]
def NO_CALLBACK(*args: Any, **kwargs: Any) -> NoReturn:
@ -97,7 +109,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[CookiesT] = None,
meta: Optional[Dict[str, Any]] = None,
encoding: str = "utf-8",
priority: int = 0,
@ -123,7 +135,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: CookiesT = cookies or {}
self.headers: Headers = Headers(headers or {}, encoding=encoding)
self.dont_filter: bool = dont_filter
@ -254,7 +266,7 @@ class Request(object_ref):
return d
def _find_method(obj: Any, func: Callable) -> str:
def _find_method(obj: Any, func: Callable[..., Any]) -> str:
"""Helper function for Request.to_dict"""
# Only instance methods contain ``__func__``
if obj and hasattr(func, "__func__"):

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)

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

@ -29,7 +29,7 @@ from twisted.internet.ssl import Certificate
from scrapy.exceptions import NotSupported
from scrapy.http.headers import Headers
from scrapy.http.request import Request
from scrapy.http.request import CookiesT, Request
from scrapy.link import Link
from scrapy.utils.trackref import object_ref
@ -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[CookiesT] = 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[CookiesT] = None,
meta: Optional[Dict[str, Any]] = None,
encoding: Optional[str] = "utf-8",
priority: int = 0,

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

@ -36,7 +36,7 @@ from w3lib.encoding import (
)
from w3lib.html import strip_html5_whitespace
from scrapy.http import Request
from scrapy.http.request import CookiesT, Request
from scrapy.http.response import Response
from scrapy.link import Link
from scrapy.utils.python import memoizemethod_noargs, to_unicode
@ -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[CookiesT] = 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[CookiesT] = 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).

4
scrapy/mail.py
View File

@ -97,7 +97,7 @@ class MailSender:
subject: str,
body: str,
cc: Union[str, List[str], None] = None,
attachs: Sequence[Tuple[str, str, IO]] = (),
attachs: Sequence[Tuple[str, str, IO[Any]]] = (),
mimetype: str = "text/plain",
charset: Optional[str] = None,
_callback: Optional[Callable[..., None]] = None,
@ -214,7 +214,7 @@ class MailSender:
return d
def _create_sender_factory(
self, to_addrs: List[str], msg: IO, d: Deferred
self, to_addrs: List[str], msg: IO[bytes], d: Deferred
) -> ESMTPSenderFactory:
from twisted.mail.smtp import ESMTPSenderFactory

2
scrapy/pipelines/files.py
View File

@ -47,7 +47,7 @@ def _to_string(path: Union[str, PathLike]) -> str:
return str(path) # convert a Path object to string
def _md5sum(file: IO) -> str:
def _md5sum(file: IO[bytes]) -> str:
"""Calculate the md5 checksum of a file-like object without reading its
whole content in memory.

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:

10
scrapy/utils/asyncgen.py
View File

@ -1,14 +1,18 @@
from typing import AsyncGenerator, AsyncIterable, Iterable, Union
from typing import AsyncGenerator, AsyncIterable, Iterable, List, TypeVar, Union
_T = TypeVar("_T")
async def collect_asyncgen(result: AsyncIterable) -> list:
async def collect_asyncgen(result: AsyncIterable[_T]) -> List[_T]:
results = []
async for x in result:
results.append(x)
return results
async def as_async_generator(it: Union[Iterable, AsyncIterable]) -> AsyncGenerator:
async def as_async_generator(
it: Union[Iterable[_T], AsyncIterable[_T]]
) -> AsyncGenerator[_T, None]:
"""Wraps an iterable (sync or async) into an async generator."""
if isinstance(it, AsyncIterable):
async for r in it:

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:

4
scrapy/utils/datatypes.py
View File

@ -196,8 +196,8 @@ class LocalWeakReferencedCache(weakref.WeakKeyDictionary):
class SequenceExclude:
"""Object to test if an item is NOT within some sequence."""
def __init__(self, seq: Sequence):
self.seq: Sequence = seq
def __init__(self, seq: Sequence[Any]):
self.seq: Sequence[Any] = seq
def __contains__(self, item: Any) -> bool:
return item not in self.seq

2
scrapy/utils/ftp.py
View File

@ -21,7 +21,7 @@ def ftp_makedirs_cwd(ftp: FTP, path: str, first_call: bool = True) -> None:
def ftp_store_file(
*,
path: str,
file: IO,
file: IO[bytes],
host: str,
port: int,
username: str,

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/misc.py
View File

@ -56,7 +56,7 @@ def arg_to_iter(arg: Any) -> Iterable[Any]:
return [arg]
def load_object(path: Union[str, Callable]) -> Any:
def load_object(path: Union[str, Callable[..., Any]]) -> Any:
"""Load an object given its absolute object path, and return it.
The object can be the import path of a class, function, variable or an
@ -111,7 +111,7 @@ def walk_modules(path: str) -> List[ModuleType]:
return mods
def md5sum(file: IO) -> str:
def md5sum(file: IO[bytes]) -> str:
"""Calculate the md5 checksum of a file-like object without reading its
whole content in memory.
@ -263,7 +263,7 @@ def walk_callable(node: ast.AST) -> Generator[ast.AST, Any, None]:
_generator_callbacks_cache = LocalWeakReferencedCache(limit=128)
def is_generator_with_return_value(callable: Callable) -> bool:
def is_generator_with_return_value(callable: Callable[..., Any]) -> bool:
"""
Returns True if a callable is a generator function which includes a
'return' statement with a value different than None, False otherwise
@ -300,7 +300,9 @@ def is_generator_with_return_value(callable: Callable) -> bool:
return bool(_generator_callbacks_cache[callable])
def warn_on_generator_with_return_value(spider: Spider, callable: Callable) -> None:
def warn_on_generator_with_return_value(
spider: Spider, callable: Callable[..., Any]
) -> None:
"""
Logs a warning if a callable is a generator function and includes
a 'return' statement with a value different than None

26
scrapy/utils/python.py
View File

@ -42,9 +42,11 @@ if TYPE_CHECKING:
_P = ParamSpec("_P")
_T = TypeVar("_T")
_KT = TypeVar("_KT")
_VT = TypeVar("_VT")
def flatten(x: Iterable) -> list:
def flatten(x: Iterable[Any]) -> List[Any]:
"""flatten(sequence) -> list
Returns a single, flat list which contains all elements retrieved
@ -64,7 +66,7 @@ def flatten(x: Iterable) -> list:
return list(iflatten(x))
def iflatten(x: Iterable) -> Iterable:
def iflatten(x: Iterable[Any]) -> Iterable[Any]:
"""iflatten(sequence) -> iterator
Similar to ``.flatten()``, but returns iterator instead"""
@ -99,10 +101,10 @@ def is_listlike(x: Any) -> bool:
return hasattr(x, "__iter__") and not isinstance(x, (str, bytes))
def unique(list_: Iterable, key: Callable[[Any], Any] = lambda x: x) -> list:
def unique(list_: Iterable[_T], key: Callable[[_T], Any] = lambda x: x) -> List[_T]:
"""efficient function to uniquify a list preserving item order"""
seen = set()
result = []
result: List[_T] = []
for item in list_:
seenkey = key(item)
if seenkey in seen:
@ -146,7 +148,7 @@ def to_bytes(
def re_rsearch(
pattern: Union[str, Pattern], text: str, chunk_size: int = 1024
pattern: Union[str, Pattern[str]], text: str, chunk_size: int = 1024
) -> Optional[Tuple[int, int]]:
"""
This function does a reverse search in a text using a regular expression
@ -215,7 +217,7 @@ def binary_is_text(data: bytes) -> bool:
return all(c not in _BINARYCHARS for c in data)
def get_func_args(func: Callable, stripself: bool = False) -> List[str]:
def get_func_args(func: Callable[..., Any], stripself: bool = False) -> List[str]:
"""Return the argument name list of a callable object"""
if not callable(func):
raise TypeError(f"func must be callable, got '{type(func).__name__}'")
@ -245,7 +247,7 @@ def get_func_args(func: Callable, stripself: bool = False) -> List[str]:
return args
def get_spec(func: Callable) -> Tuple[List[str], Dict[str, Any]]:
def get_spec(func: Callable[..., Any]) -> Tuple[List[str], Dict[str, Any]]:
"""Returns (args, kwargs) tuple for a function
>>> import re
>>> get_spec(re.match)
@ -283,7 +285,7 @@ def get_spec(func: Callable) -> Tuple[List[str], Dict[str, Any]]:
def equal_attributes(
obj1: Any, obj2: Any, attributes: Optional[List[Union[str, Callable]]]
obj1: Any, obj2: Any, attributes: Optional[List[Union[str, Callable[[Any], Any]]]]
) -> bool:
"""Compare two objects attributes"""
# not attributes given return False by default
@ -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

8
scrapy/utils/test.py
View File

@ -7,7 +7,7 @@ import os
from importlib import import_module
from pathlib import Path
from posixpath import split
from typing import Any, Coroutine, Dict, List, Optional, Tuple, Type
from typing import Any, Awaitable, Dict, List, Optional, Tuple, Type, TypeVar
from unittest import TestCase, mock
from twisted.internet.defer import Deferred
@ -17,6 +17,8 @@ from scrapy import Spider
from scrapy.crawler import Crawler
from scrapy.utils.boto import is_botocore_available
_T = TypeVar("_T")
def assert_gcs_environ() -> None:
if "GCS_PROJECT_ID" not in os.environ:
@ -118,8 +120,8 @@ def assert_samelines(
testcase.assertEqual(text1.splitlines(), text2.splitlines(), msg)
def get_from_asyncio_queue(value: Any) -> Coroutine:
q: asyncio.Queue = asyncio.Queue()
def get_from_asyncio_queue(value: _T) -> Awaitable[_T]:
q: asyncio.Queue[_T] = asyncio.Queue()
getter = q.get()
q.put_nowait(value)
return getter