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

Full typing for scrapy/extensions, part 2. (#6279)

This commit is contained in:
Andrey Rakhmatullin 2024-03-11 14:09:09 +05:00 committed by GitHub
parent 6e84648c07
commit 642af40704
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 165 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
docs/topics/telnetconsole.rst
View File

@ -172,8 +172,8 @@ TELNETCONSOLE_PORT
Default: ``[6023, 6073]``
The port range to use for the telnet console. If set to ``None`` or ``0``, a
dynamically assigned port is used.
The port range to use for the telnet console. If set to ``None``, a dynamically
assigned port is used.
.. setting:: TELNETCONSOLE_HOST

89
scrapy/extensions/periodic_log.py
View File

@ -1,12 +1,22 @@
from __future__ import annotations
import logging
from datetime import datetime, timezone
from json import JSONEncoder
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
from twisted.internet import task
from scrapy import signals
from scrapy import Spider, signals
from scrapy.crawler import Crawler
from scrapy.exceptions import NotConfigured
from scrapy.statscollectors import StatsCollector
from scrapy.utils.serialize import ScrapyJSONEncoder
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
logger = logging.getLogger(__name__)
@ -15,32 +25,34 @@ class PeriodicLog:
def __init__(
self,
stats,
interval=60.0,
ext_stats={},
ext_delta={},
ext_timing_enabled=False,
stats: StatsCollector,
interval: float = 60.0,
ext_stats: Dict[str, Any] = {},
ext_delta: Dict[str, Any] = {},
ext_timing_enabled: bool = False,
):
self.stats = stats
self.interval = interval
self.multiplier = 60.0 / self.interval
self.task = None
self.encoder = ScrapyJSONEncoder(sort_keys=True, indent=4)
self.ext_stats_enabled = bool(ext_stats)
self.ext_stats_include = ext_stats.get("include", [])
self.ext_stats_exclude = ext_stats.get("exclude", [])
self.ext_delta_enabled = bool(ext_delta)
self.ext_delta_include = ext_delta.get("include", [])
self.ext_delta_exclude = ext_delta.get("exclude", [])
self.ext_timing_enabled = ext_timing_enabled
self.stats: StatsCollector = stats
self.interval: float = interval
self.multiplier: float = 60.0 / self.interval
self.task: Optional[task.LoopingCall] = None
self.encoder: JSONEncoder = ScrapyJSONEncoder(sort_keys=True, indent=4)
self.ext_stats_enabled: bool = bool(ext_stats)
self.ext_stats_include: List[str] = ext_stats.get("include", [])
self.ext_stats_exclude: List[str] = ext_stats.get("exclude", [])
self.ext_delta_enabled: bool = bool(ext_delta)
self.ext_delta_include: List[str] = ext_delta.get("include", [])
self.ext_delta_exclude: List[str] = ext_delta.get("exclude", [])
self.ext_timing_enabled: bool = ext_timing_enabled
@classmethod
def from_crawler(cls, crawler):
interval = crawler.settings.getfloat("LOGSTATS_INTERVAL")
def from_crawler(cls, crawler: Crawler) -> Self:
interval: float = crawler.settings.getfloat("LOGSTATS_INTERVAL")
if not interval:
raise NotConfigured
try:
ext_stats = crawler.settings.getdict("PERIODIC_LOG_STATS")
ext_stats: Optional[Dict[str, Any]] = crawler.settings.getdict(
"PERIODIC_LOG_STATS"
)
except (TypeError, ValueError):
ext_stats = (
{"enabled": True}
@ -48,7 +60,9 @@ class PeriodicLog:
else None
)
try:
ext_delta = crawler.settings.getdict("PERIODIC_LOG_DELTA")
ext_delta: Optional[Dict[str, Any]] = crawler.settings.getdict(
"PERIODIC_LOG_DELTA"
)
except (TypeError, ValueError):
ext_delta = (
{"enabled": True}
@ -56,11 +70,14 @@ class PeriodicLog:
else None
)
ext_timing_enabled = crawler.settings.getbool(
ext_timing_enabled: bool = crawler.settings.getbool(
"PERIODIC_LOG_TIMING_ENABLED", False
)
if not (ext_stats or ext_delta or ext_timing_enabled):
raise NotConfigured
assert crawler.stats
assert ext_stats is not None
assert ext_delta is not None
o = cls(
crawler.stats,
interval,
@ -72,16 +89,16 @@ class PeriodicLog:
crawler.signals.connect(o.spider_closed, signal=signals.spider_closed)
return o
def spider_opened(self, spider):
self.time_prev = datetime.now(tz=timezone.utc)
self.delta_prev = {}
self.stats_prev = {}
def spider_opened(self, spider: Spider) -> None:
self.time_prev: datetime = datetime.now(tz=timezone.utc)
self.delta_prev: Dict[str, Union[int, float]] = {}
self.stats_prev: Dict[str, Union[int, float]] = {}
self.task = task.LoopingCall(self.log)
self.task.start(self.interval)
def log(self):
data = {}
def log(self) -> None:
data: Dict[str, Any] = {}
if self.ext_timing_enabled:
data.update(self.log_timing())
if self.ext_delta_enabled:
@ -90,8 +107,8 @@ class PeriodicLog:
data.update(self.log_crawler_stats())
logger.info(self.encoder.encode(data))
def log_delta(self):
num_stats = {
def log_delta(self) -> Dict[str, Any]:
num_stats: Dict[str, Union[int, float]] = {
k: v
for k, v in self.stats._stats.items()
if isinstance(v, (int, float))
@ -101,7 +118,7 @@ class PeriodicLog:
self.delta_prev = num_stats
return {"delta": delta}
def log_timing(self):
def log_timing(self) -> Dict[str, Any]:
now = datetime.now(tz=timezone.utc)
time = {
"log_interval": self.interval,
@ -113,7 +130,7 @@ class PeriodicLog:
self.time_prev = now
return {"time": time}
def log_crawler_stats(self):
def log_crawler_stats(self) -> Dict[str, Any]:
stats = {
k: v
for k, v in self.stats._stats.items()
@ -121,7 +138,9 @@ class PeriodicLog:
}
return {"stats": stats}
def param_allowed(self, stat_name, include, exclude):
def param_allowed(
self, stat_name: str, include: List[str], exclude: List[str]
) -> bool:
if not include and not exclude:
return True
for p in exclude:
@ -134,7 +153,7 @@ class PeriodicLog:
return True
return False
def spider_closed(self, spider, reason):
def spider_closed(self, spider: Spider, reason: str) -> None:
self.log()
if self.task and self.task.running:
self.task.stop()

4
scrapy/extensions/postprocessing.py
View File

@ -6,7 +6,7 @@ from bz2 import BZ2File
from gzip import GzipFile
from io import IOBase
from lzma import LZMAFile
from typing import Any, BinaryIO, Dict, List
from typing import Any, BinaryIO, Dict, List, cast
from scrapy.utils.misc import load_object
@ -142,7 +142,7 @@ class PostProcessingManager(IOBase):
:return: returns number of bytes written
:rtype: int
"""
return self.head_plugin.write(data)
return cast(int, self.head_plugin.write(data))
def tell(self) -> int:
return self.file.tell()

26
scrapy/extensions/spiderstate.py
View File

@ -1,19 +1,27 @@
from __future__ import annotations
import pickle # nosec
from pathlib import Path
from typing import TYPE_CHECKING, Optional
from scrapy import signals
from scrapy import Spider, signals
from scrapy.crawler import Crawler
from scrapy.exceptions import NotConfigured
from scrapy.utils.job import job_dir
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
class SpiderState:
"""Store and load spider state during a scraping job"""
def __init__(self, jobdir=None):
self.jobdir = jobdir
def __init__(self, jobdir: Optional[str] = None):
self.jobdir: Optional[str] = jobdir
@classmethod
def from_crawler(cls, crawler):
def from_crawler(cls, crawler: Crawler) -> Self:
jobdir = job_dir(crawler.settings)
if not jobdir:
raise NotConfigured
@ -23,18 +31,20 @@ class SpiderState:
crawler.signals.connect(obj.spider_opened, signal=signals.spider_opened)
return obj
def spider_closed(self, spider):
def spider_closed(self, spider: Spider) -> None:
if self.jobdir:
with Path(self.statefn).open("wb") as f:
assert hasattr(spider, "state") # set in spider_opened
pickle.dump(spider.state, f, protocol=4)
def spider_opened(self, spider):
def spider_opened(self, spider: Spider) -> None:
if self.jobdir and Path(self.statefn).exists():
with Path(self.statefn).open("rb") as f:
spider.state = pickle.load(f) # nosec
spider.state = pickle.load(f) # type: ignore[attr-defined] # nosec
else:
spider.state = {}
spider.state = {} # type: ignore[attr-defined]
@property
def statefn(self) -> str:
assert self.jobdir
return str(Path(self.jobdir, "spider.state"))

31
scrapy/extensions/statsmailer.py
View File

@ -4,28 +4,41 @@ StatsMailer extension sends an email when a spider finishes scraping.
Use STATSMAILER_RCPTS setting to enable and give the recipient mail address
"""
from scrapy import signals
from __future__ import annotations
from typing import TYPE_CHECKING, List, Optional
from twisted.internet.defer import Deferred
from scrapy import Spider, signals
from scrapy.crawler import Crawler
from scrapy.exceptions import NotConfigured
from scrapy.mail import MailSender
from scrapy.statscollectors import StatsCollector
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
class StatsMailer:
def __init__(self, stats, recipients, mail):
self.stats = stats
self.recipients = recipients
self.mail = mail
def __init__(self, stats: StatsCollector, recipients: List[str], mail: MailSender):
self.stats: StatsCollector = stats
self.recipients: List[str] = recipients
self.mail: MailSender = mail
@classmethod
def from_crawler(cls, crawler):
recipients = crawler.settings.getlist("STATSMAILER_RCPTS")
def from_crawler(cls, crawler: Crawler) -> Self:
recipients: List[str] = crawler.settings.getlist("STATSMAILER_RCPTS")
if not recipients:
raise NotConfigured
mail = MailSender.from_settings(crawler.settings)
mail: MailSender = MailSender.from_settings(crawler.settings)
assert crawler.stats
o = cls(crawler.stats, recipients, mail)
crawler.signals.connect(o.spider_closed, signal=signals.spider_closed)
return o
def spider_closed(self, spider):
def spider_closed(self, spider: Spider) -> Optional[Deferred]:
spider_stats = self.stats.get_stats(spider)
body = "Global stats\n\n"
body += "\n".join(f"{k:<50} : {v}" for k, v in self.stats.get_stats().items())

37
scrapy/extensions/telnet.py
View File

@ -4,13 +4,17 @@ Scrapy Telnet Console extension
See documentation in docs/topics/telnetconsole.rst
"""
from __future__ import annotations
import binascii
import logging
import os
import pprint
import traceback
from typing import TYPE_CHECKING, Any, Dict, List
from twisted.internet import protocol
from twisted.internet.tcp import Port
try:
from twisted.conch import manhole, telnet
@ -22,12 +26,16 @@ except (ImportError, SyntaxError):
TWISTED_CONCH_AVAILABLE = False
from scrapy import signals
from scrapy.crawler import Crawler
from scrapy.exceptions import NotConfigured
from scrapy.utils.decorators import defers
from scrapy.utils.engine import print_engine_status
from scrapy.utils.reactor import listen_tcp
from scrapy.utils.trackref import print_live_refs
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
logger = logging.getLogger(__name__)
# signal to update telnet variables
@ -36,7 +44,7 @@ update_telnet_vars = object()
class TelnetConsole(protocol.ServerFactory):
def __init__(self, crawler):
def __init__(self, crawler: Crawler):
if not crawler.settings.getbool("TELNETCONSOLE_ENABLED"):
raise NotConfigured
if not TWISTED_CONCH_AVAILABLE:
@ -44,14 +52,14 @@ class TelnetConsole(protocol.ServerFactory):
"TELNETCONSOLE_ENABLED setting is True but required twisted "
"modules failed to import:\n" + _TWISTED_CONCH_TRACEBACK
)
self.crawler = crawler
self.noisy = False
self.portrange = [
self.crawler: Crawler = crawler
self.noisy: bool = False
self.portrange: List[int] = [
int(x) for x in crawler.settings.getlist("TELNETCONSOLE_PORT")
]
self.host = crawler.settings["TELNETCONSOLE_HOST"]
self.username = crawler.settings["TELNETCONSOLE_USERNAME"]
self.password = crawler.settings["TELNETCONSOLE_PASSWORD"]
self.host: str = crawler.settings["TELNETCONSOLE_HOST"]
self.username: str = crawler.settings["TELNETCONSOLE_USERNAME"]
self.password: str = crawler.settings["TELNETCONSOLE_PASSWORD"]
if not self.password:
self.password = binascii.hexlify(os.urandom(8)).decode("utf8")
@ -61,11 +69,11 @@ class TelnetConsole(protocol.ServerFactory):
self.crawler.signals.connect(self.stop_listening, signals.engine_stopped)
@classmethod
def from_crawler(cls, crawler):
def from_crawler(cls, crawler: Crawler) -> Self:
return cls(crawler)
def start_listening(self):
self.port = listen_tcp(self.portrange, self.host, self)
def start_listening(self) -> None:
self.port: Port = listen_tcp(self.portrange, self.host, self)
h = self.port.getHost()
logger.info(
"Telnet console listening on %(host)s:%(port)d",
@ -73,10 +81,10 @@ class TelnetConsole(protocol.ServerFactory):
extra={"crawler": self.crawler},
)
def stop_listening(self):
def stop_listening(self) -> None:
self.port.stopListening()
def protocol(self):
def protocol(self) -> telnet.TelnetTransport: # type: ignore[override]
class Portal:
"""An implementation of IPortal"""
@ -95,9 +103,10 @@ class TelnetConsole(protocol.ServerFactory):
return telnet.TelnetTransport(telnet.AuthenticatingTelnetProtocol, Portal())
def _get_telnet_vars(self):
def _get_telnet_vars(self) -> Dict[str, Any]:
# Note: if you add entries here also update topics/telnetconsole.rst
telnet_vars = {
assert self.crawler.engine
telnet_vars: Dict[str, Any] = {
"engine": self.crawler.engine,
"spider": self.crawler.engine.spider,
"slot": self.crawler.engine.slot,

49
scrapy/extensions/throttle.py
View File

@ -1,19 +1,29 @@
import logging
from __future__ import annotations
from scrapy import signals
import logging
from typing import TYPE_CHECKING, Optional, Tuple
from scrapy import Request, Spider, signals
from scrapy.core.downloader import Slot
from scrapy.crawler import Crawler
from scrapy.exceptions import NotConfigured
from scrapy.http import Response
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
logger = logging.getLogger(__name__)
class AutoThrottle:
def __init__(self, crawler):
self.crawler = crawler
def __init__(self, crawler: Crawler):
self.crawler: Crawler = crawler
if not crawler.settings.getbool("AUTOTHROTTLE_ENABLED"):
raise NotConfigured
self.debug = crawler.settings.getbool("AUTOTHROTTLE_DEBUG")
self.target_concurrency = crawler.settings.getfloat(
self.debug: bool = crawler.settings.getbool("AUTOTHROTTLE_DEBUG")
self.target_concurrency: float = crawler.settings.getfloat(
"AUTOTHROTTLE_TARGET_CONCURRENCY"
)
if self.target_concurrency <= 0.0:
@ -27,27 +37,29 @@ class AutoThrottle:
)
@classmethod
def from_crawler(cls, crawler):
def from_crawler(cls, crawler: Crawler) -> Self:
return cls(crawler)
def _spider_opened(self, spider):
def _spider_opened(self, spider: Spider) -> None:
self.mindelay = self._min_delay(spider)
self.maxdelay = self._max_delay(spider)
spider.download_delay = self._start_delay(spider)
spider.download_delay = self._start_delay(spider) # type: ignore[attr-defined]
def _min_delay(self, spider):
def _min_delay(self, spider: Spider) -> float:
s = self.crawler.settings
return getattr(spider, "download_delay", s.getfloat("DOWNLOAD_DELAY"))
def _max_delay(self, spider):
def _max_delay(self, spider: Spider) -> float:
return self.crawler.settings.getfloat("AUTOTHROTTLE_MAX_DELAY")
def _start_delay(self, spider):
def _start_delay(self, spider: Spider) -> float:
return max(
self.mindelay, self.crawler.settings.getfloat("AUTOTHROTTLE_START_DELAY")
)
def _response_downloaded(self, response, request, spider):
def _response_downloaded(
self, response: Response, request: Request, spider: Spider
) -> None:
key, slot = self._get_slot(request, spider)
latency = request.meta.get("download_latency")
if latency is None or slot is None:
@ -74,11 +86,16 @@ class AutoThrottle:
extra={"spider": spider},
)
def _get_slot(self, request, spider):
key = request.meta.get("download_slot")
def _get_slot(
self, request: Request, spider: Spider
) -> Tuple[Optional[str], Optional[Slot]]:
key: Optional[str] = request.meta.get("download_slot")
if key is None:
return None, None
assert self.crawler.engine
return key, self.crawler.engine.downloader.slots.get(key)
def _adjust_delay(self, slot, latency, response):
def _adjust_delay(self, slot: Slot, latency: float, response: Response) -> None:
"""Define delay adjustment policy"""
# If a server needs `latency` seconds to respond then

12
scrapy/mail.py
View File

@ -14,10 +14,11 @@ from email.mime.nonmultipart import MIMENonMultipart
from email.mime.text import MIMEText
from email.utils import formatdate
from io import BytesIO
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Optional
from twisted import version as twisted_version
from twisted.internet import defer, ssl
from twisted.internet import ssl
from twisted.internet.defer import Deferred
from twisted.python.versions import Version
from scrapy.settings import BaseSettings
@ -85,9 +86,10 @@ class MailSender:
mimetype="text/plain",
charset=None,
_callback=None,
):
) -> Optional[Deferred]:
from twisted.internet import reactor
msg: MIMEBase
if attachs:
msg = MIMEMultipart()
else:
@ -134,7 +136,7 @@ class MailSender:
"mailattachs": len(attachs),
},
)
return
return None
dfd = self._sendmail(rcpts, msg.as_string().encode(charset or "utf-8"))
dfd.addCallbacks(
@ -178,7 +180,7 @@ class MailSender:
from twisted.internet import reactor
msg = BytesIO(msg)
d = defer.Deferred()
d = Deferred()
factory = self._create_sender_factory(to_addrs, msg, d)

8
scrapy/utils/reactor.py
View File

@ -2,17 +2,19 @@ import asyncio
import sys
from asyncio import AbstractEventLoop, AbstractEventLoopPolicy
from contextlib import suppress
from typing import Any, Callable, Dict, Optional, Sequence, Type
from typing import Any, Callable, Dict, List, Optional, Sequence, Type
from warnings import catch_warnings, filterwarnings, warn
from twisted.internet import asyncioreactor, error
from twisted.internet.base import DelayedCall
from twisted.internet.protocol import ServerFactory
from twisted.internet.tcp import Port
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.utils.misc import load_object
def listen_tcp(portrange, host, factory):
def listen_tcp(portrange: List[int], host: str, factory: ServerFactory) -> Port: # type: ignore[return]
"""Like reactor.listenTCP but tries different ports in a range."""
from twisted.internet import reactor
@ -20,8 +22,6 @@ def listen_tcp(portrange, host, factory):
raise ValueError(f"invalid portrange: {portrange}")
if not portrange:
return reactor.listenTCP(0, factory, interface=host)
if not hasattr(portrange, "__iter__"):
return reactor.listenTCP(portrange, factory, interface=host)
if len(portrange) == 1:
return reactor.listenTCP(portrange[0], factory, interface=host)
for x in range(portrange[0], portrange[1] + 1):