Full typing for scrapy/dupefilters.py.

This commit is contained in:
Andrey Rakhmatullin 2023-08-07 00:41:11 +04:00
parent dc6e142096
commit 89503ae3f1
1 changed files with 20 additions and 14 deletions

View File

@ -1,6 +1,8 @@
from __future__ import annotations
import logging
from pathlib import Path
from typing import Optional, Set, Type, TypeVar
from typing import TYPE_CHECKING, Optional, Set
from warnings import warn
from twisted.internet.defer import Deferred
@ -12,14 +14,16 @@ from scrapy.utils.deprecate import ScrapyDeprecationWarning
from scrapy.utils.job import job_dir
from scrapy.utils.request import RequestFingerprinter, referer_str
BaseDupeFilterTV = TypeVar("BaseDupeFilterTV", bound="BaseDupeFilter")
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
from scrapy.crawler import Crawler
class BaseDupeFilter:
@classmethod
def from_settings(
cls: Type[BaseDupeFilterTV], settings: BaseSettings
) -> BaseDupeFilterTV:
def from_settings(cls, settings: BaseSettings) -> Self:
return cls()
def request_seen(self, request: Request) -> bool:
@ -36,9 +40,6 @@ class BaseDupeFilter:
pass
RFPDupeFilterTV = TypeVar("RFPDupeFilterTV", bound="RFPDupeFilter")
class RFPDupeFilter(BaseDupeFilter):
"""Request Fingerprint duplicates filter"""
@ -47,10 +48,12 @@ class RFPDupeFilter(BaseDupeFilter):
path: Optional[str] = None,
debug: bool = False,
*,
fingerprinter=None,
fingerprinter: Optional[RequestFingerprinter] = None,
) -> None:
self.file = None
self.fingerprinter = fingerprinter or RequestFingerprinter()
self.fingerprinter: RequestFingerprinter = (
fingerprinter or RequestFingerprinter()
)
self.fingerprints: Set[str] = set()
self.logdupes = True
self.debug = debug
@ -62,8 +65,11 @@ class RFPDupeFilter(BaseDupeFilter):
@classmethod
def from_settings(
cls: Type[RFPDupeFilterTV], settings: BaseSettings, *, fingerprinter=None
) -> RFPDupeFilterTV:
cls,
settings: BaseSettings,
*,
fingerprinter: Optional[RequestFingerprinter] = None,
) -> Self:
debug = settings.getbool("DUPEFILTER_DEBUG")
try:
return cls(job_dir(settings), debug, fingerprinter=fingerprinter)
@ -75,11 +81,11 @@ class RFPDupeFilter(BaseDupeFilter):
ScrapyDeprecationWarning,
)
result = cls(job_dir(settings), debug)
result.fingerprinter = fingerprinter
result.fingerprinter = fingerprinter or RequestFingerprinter()
return result
@classmethod
def from_crawler(cls, crawler):
def from_crawler(cls, crawler: Crawler) -> Self:
try:
return cls.from_settings(
crawler.settings,