diff --git a/scrapy/dupefilters.py b/scrapy/dupefilters.py index ac5478e7c..292c68099 100644 --- a/scrapy/dupefilters.py +++ b/scrapy/dupefilters.py @@ -1,35 +1,47 @@ -import os import logging +import os +from typing import Optional, Set, Type, TypeVar +from twisted.internet.defer import Deferred + +from scrapy.http.request import Request +from scrapy.settings import BaseSettings +from scrapy.spiders import Spider from scrapy.utils.job import job_dir from scrapy.utils.request import referer_str, request_fingerprint -class BaseDupeFilter: +BaseDupeFilterTV = TypeVar("BaseDupeFilterTV", bound="BaseDupeFilter") + +class BaseDupeFilter: @classmethod - def from_settings(cls, settings): + def from_settings(cls: Type[BaseDupeFilterTV], settings: BaseSettings) -> BaseDupeFilterTV: return cls() - def request_seen(self, request): + def request_seen(self, request: Request) -> bool: return False - def open(self): # can return deferred + def open(self) -> Optional[Deferred]: pass - def close(self, reason): # can return a deferred + def close(self, reason: str) -> Optional[Deferred]: pass - def log(self, request, spider): # log that a request has been filtered + def log(self, request: Request, spider: Spider) -> None: + """Log that a request has been filtered""" pass +RFPDupeFilterTV = TypeVar("RFPDupeFilterTV", bound="RFPDupeFilter") + + class RFPDupeFilter(BaseDupeFilter): """Request Fingerprint duplicates filter""" - def __init__(self, path=None, debug=False): + def __init__(self, path: Optional[str] = None, debug: bool = False) -> None: self.file = None - self.fingerprints = set() + self.fingerprints: Set[str] = set() self.logdupes = True self.debug = debug self.logger = logging.getLogger(__name__) @@ -39,26 +51,27 @@ class RFPDupeFilter(BaseDupeFilter): self.fingerprints.update(x.rstrip() for x in self.file) @classmethod - def from_settings(cls, settings): + def from_settings(cls: Type[RFPDupeFilterTV], settings: BaseSettings) -> RFPDupeFilterTV: debug = settings.getbool('DUPEFILTER_DEBUG') return cls(job_dir(settings), debug) - def request_seen(self, request): + def request_seen(self, request: Request) -> bool: fp = self.request_fingerprint(request) if fp in self.fingerprints: return True self.fingerprints.add(fp) if self.file: self.file.write(fp + '\n') + return False - def request_fingerprint(self, request): + def request_fingerprint(self, request: Request) -> str: return request_fingerprint(request) - def close(self, reason): + def close(self, reason: str) -> None: if self.file: self.file.close() - def log(self, request, spider): + def log(self, request: Request, spider: Spider) -> None: if self.debug: msg = "Filtered duplicate request: %(request)s (referer: %(referer)s)" args = {'request': request, 'referer': referer_str(request)} diff --git a/scrapy/utils/request.py b/scrapy/utils/request.py index 66736b42f..541368423 100644 --- a/scrapy/utils/request.py +++ b/scrapy/utils/request.py @@ -24,7 +24,7 @@ def request_fingerprint( request: Request, include_headers: Optional[Iterable[Union[bytes, str]]] = None, keep_fragments: bool = False, -): +) -> str: """ Return the request fingerprint.