diff --git a/scrapy/exceptions.py b/scrapy/exceptions.py index fedd02805..6e83e4a00 100644 --- a/scrapy/exceptions.py +++ b/scrapy/exceptions.py @@ -4,6 +4,7 @@ Scrapy core exceptions These exceptions are documented in docs/topics/exceptions.rst. Please don't add new exceptions here without documenting them there. """ +from typing import Any # Internal @@ -77,7 +78,7 @@ class NotSupported(Exception): class UsageError(Exception): """To indicate a command-line usage error""" - def __init__(self, *a, **kw): + def __init__(self, *a: Any, **kw: Any): self.print_help = kw.pop("print_help", True) super().__init__(*a, **kw) diff --git a/scrapy/extensions/feedexport.py b/scrapy/extensions/feedexport.py index c8022ff57..2bbcaf3ad 100644 --- a/scrapy/extensions/feedexport.py +++ b/scrapy/extensions/feedexport.py @@ -11,7 +11,7 @@ import warnings from datetime import datetime from pathlib import Path, PureWindowsPath from tempfile import NamedTemporaryFile -from typing import IO, Any, Callable, List, Optional, Tuple, Union +from typing import IO, Any, Callable, Dict, List, Optional, Tuple, Union from urllib.parse import unquote, urlparse from twisted.internet import defer, threads @@ -282,15 +282,22 @@ class GCSFeedStorage(BlockingFeedStorage): class FTPFeedStorage(BlockingFeedStorage): - def __init__(self, uri, use_active_mode=False, *, feed_options=None): + def __init__( + self, + uri: str, + use_active_mode: bool = False, + *, + feed_options: Optional[Dict[str, Any]] = None, + ): u = urlparse(uri) - self.host = u.hostname - self.port = int(u.port or "21") - self.username = u.username - self.password = unquote(u.password or "") - self.path = u.path - self.use_active_mode = use_active_mode - self.overwrite = not feed_options or feed_options.get("overwrite", True) + assert u.hostname + self.host: str = u.hostname + self.port: int = int(u.port or "21") + self.username: str = u.username or "" + self.password: str = unquote(u.password or "") + self.path: str = u.path + self.use_active_mode: bool = use_active_mode + self.overwrite: bool = not feed_options or feed_options.get("overwrite", True) @classmethod def from_crawler(cls, crawler, uri, *, feed_options=None): diff --git a/scrapy/utils/conf.py b/scrapy/utils/conf.py index 1889f7571..641dfa4a2 100644 --- a/scrapy/utils/conf.py +++ b/scrapy/utils/conf.py @@ -5,7 +5,18 @@ import warnings from configparser import ConfigParser from operator import itemgetter from pathlib import Path -from typing import Any, Dict, List, Optional, Union +from typing import ( + Any, + Callable, + Collection, + Dict, + Iterable, + List, + Mapping, + MutableMapping, + Optional, + Union, +) from scrapy.exceptions import ScrapyDeprecationWarning, UsageError from scrapy.settings import BaseSettings @@ -13,17 +24,21 @@ from scrapy.utils.deprecate import update_classpath from scrapy.utils.python import without_none_values -def build_component_list(compdict, custom=None, convert=update_classpath): +def build_component_list( + compdict: MutableMapping[Any, Any], + custom: Any = None, + convert: Callable[[Any], Any] = update_classpath, +) -> List[Any]: """Compose a component list from a { class: order } dictionary.""" - def _check_components(complist): + def _check_components(complist: Collection[Any]) -> None: if len({convert(c) for c in complist}) != len(complist): raise ValueError( f"Some paths in {complist!r} convert to the same object, " "please update your settings" ) - def _map_keys(compdict): + def _map_keys(compdict: Mapping[Any, Any]) -> Union[BaseSettings, Dict[Any, Any]]: if isinstance(compdict, BaseSettings): compbs = BaseSettings() for k, v in compdict.items(): @@ -41,7 +56,7 @@ def build_component_list(compdict, custom=None, convert=update_classpath): _check_components(compdict) return {convert(k): v for k, v in compdict.items()} - def _validate_values(compdict): + def _validate_values(compdict: Mapping[Any, Any]) -> None: """Fail if a value in the components dict is not a real number or None.""" for name, value in compdict.items(): if value is not None and not isinstance(value, numbers.Real): @@ -60,7 +75,7 @@ def build_component_list(compdict, custom=None, convert=update_classpath): ) if isinstance(custom, (list, tuple)): _check_components(custom) - return type(custom)(convert(c) for c in custom) + return type(custom)(convert(c) for c in custom) # type: ignore[return-value] compdict.update(custom) _validate_values(compdict) @@ -68,7 +83,7 @@ def build_component_list(compdict, custom=None, convert=update_classpath): return [k for k, v in sorted(compdict.items(), key=itemgetter(1))] -def arglist_to_dict(arglist): +def arglist_to_dict(arglist: List[str]) -> Dict[str, str]: """Convert a list of arguments like ['arg1=val1', 'arg2=val2', ...] to a dict """ @@ -91,7 +106,7 @@ def closest_scrapy_cfg( return closest_scrapy_cfg(path.parent, path) -def init_env(project="default", set_syspath=True): +def init_env(project: str = "default", set_syspath: bool = True) -> None: """Initialize environment to use command-line tool from inside a project dir. This sets the Scrapy settings module and modifies the Python path to be able to locate the project module. @@ -106,7 +121,7 @@ def init_env(project="default", set_syspath=True): sys.path.append(projdir) -def get_config(use_closest=True): +def get_config(use_closest: bool = True) -> ConfigParser: """Get Scrapy config file as a ConfigParser""" sources = get_sources(use_closest) cfg = ConfigParser() @@ -114,7 +129,7 @@ def get_config(use_closest=True): return cfg -def get_sources(use_closest=True) -> List[str]: +def get_sources(use_closest: bool = True) -> List[str]: xdg_config_home = ( os.environ.get("XDG_CONFIG_HOME") or Path("~/.config").expanduser() ) @@ -129,7 +144,9 @@ def get_sources(use_closest=True) -> List[str]: return sources -def feed_complete_default_values_from_settings(feed, settings): +def feed_complete_default_values_from_settings( + feed: Dict[str, Any], settings: BaseSettings +) -> Dict[str, Any]: out = feed.copy() out.setdefault("batch_item_count", settings.getint("FEED_EXPORT_BATCH_ITEM_COUNT")) out.setdefault("encoding", settings["FEED_EXPORT_ENCODING"]) @@ -145,21 +162,21 @@ def feed_complete_default_values_from_settings(feed, settings): def feed_process_params_from_cli( - settings, + settings: BaseSettings, output: List[str], - output_format=None, + output_format: Optional[str] = None, overwrite_output: Optional[List[str]] = None, -): +) -> Dict[str, Dict[str, Any]]: """ Receives feed export params (from the 'crawl' or 'runspider' commands), checks for inconsistencies in their quantities and returns a dictionary suitable to be used as the FEEDS setting. """ - valid_output_formats = without_none_values( + valid_output_formats: Iterable[str] = without_none_values( settings.getwithbase("FEED_EXPORTERS") ).keys() - def check_valid_format(output_format): + def check_valid_format(output_format: str) -> None: if output_format not in valid_output_formats: raise UsageError( f"Unrecognized output format '{output_format}'. " diff --git a/scrapy/utils/ftp.py b/scrapy/utils/ftp.py index 6bf6e9195..c77681a53 100644 --- a/scrapy/utils/ftp.py +++ b/scrapy/utils/ftp.py @@ -1,9 +1,10 @@ import posixpath from ftplib import FTP, error_perm from posixpath import dirname +from typing import IO -def ftp_makedirs_cwd(ftp, path, first_call=True): +def ftp_makedirs_cwd(ftp: FTP, path: str, first_call: bool = True) -> None: """Set the current directory of the FTP connection given in the ``ftp`` argument (as a ftplib.FTP object), creating all parent directories if they don't exist. The ftplib.FTP object must be already connected and logged in. @@ -18,8 +19,16 @@ def ftp_makedirs_cwd(ftp, path, first_call=True): def ftp_store_file( - *, path, file, host, port, username, password, use_active_mode=False, overwrite=True -): + *, + path: str, + file: IO, + host: str, + port: int, + username: str, + password: str, + use_active_mode: bool = False, + overwrite: bool = True, +) -> None: """Opens a FTP connection with passed credentials,sets current directory to the directory extracted from given path, then uploads the file to server """ diff --git a/scrapy/utils/job.py b/scrapy/utils/job.py index 858affc03..c49f7d758 100644 --- a/scrapy/utils/job.py +++ b/scrapy/utils/job.py @@ -5,7 +5,7 @@ from scrapy.settings import BaseSettings def job_dir(settings: BaseSettings) -> Optional[str]: - path = settings["JOBDIR"] + path: str = settings["JOBDIR"] if path and not Path(path).exists(): Path(path).mkdir(parents=True) return path diff --git a/scrapy/utils/ossignal.py b/scrapy/utils/ossignal.py index 7646264a8..f835a2221 100644 --- a/scrapy/utils/ossignal.py +++ b/scrapy/utils/ossignal.py @@ -1,4 +1,5 @@ import signal +from typing import Callable signal_names = {} for signame in dir(signal): @@ -8,7 +9,7 @@ for signame in dir(signal): signal_names[signum] = signame -def install_shutdown_handlers(function, override_sigint=True): +def install_shutdown_handlers(function: Callable, override_sigint: bool = True) -> None: """Install the given function as a signal handler for all common shutdown signals (such as SIGINT, SIGTERM, etc). If override_sigint is ``False`` the SIGINT handler won't be install if there is already a handler in place diff --git a/scrapy/utils/project.py b/scrapy/utils/project.py index 652b74759..a2c224b90 100644 --- a/scrapy/utils/project.py +++ b/scrapy/utils/project.py @@ -11,7 +11,7 @@ ENVVAR = "SCRAPY_SETTINGS_MODULE" DATADIR_CFG_SECTION = "datadir" -def inside_project(): +def inside_project() -> bool: scrapy_module = os.environ.get(ENVVAR) if scrapy_module: try: @@ -25,7 +25,7 @@ def inside_project(): return bool(closest_scrapy_cfg()) -def project_data_dir(project="default") -> str: +def project_data_dir(project: str = "default") -> str: """Return the current project data dir, creating it if it doesn't exist""" if not inside_project(): raise NotConfigured("Not inside a project") @@ -44,7 +44,7 @@ def project_data_dir(project="default") -> str: return str(d) -def data_path(path: str, createdir=False) -> str: +def data_path(path: str, createdir: bool = False) -> str: """ Return the given path joined with the .scrapy data directory. If given an absolute path, return it unmodified. @@ -60,7 +60,7 @@ def data_path(path: str, createdir=False) -> str: return str(path_obj) -def get_project_settings(): +def get_project_settings() -> Settings: if ENVVAR not in os.environ: project = os.environ.get("SCRAPY_PROJECT", "default") init_env(project) diff --git a/scrapy/utils/sitemap.py b/scrapy/utils/sitemap.py index 2622c2775..3d2ecc9a7 100644 --- a/scrapy/utils/sitemap.py +++ b/scrapy/utils/sitemap.py @@ -4,7 +4,7 @@ Module for processing Sitemaps. Note: The main purpose of this module is to provide support for the SitemapSpider, its API is subject to change without notice. """ - +from typing import Any, Dict, Generator, Iterator, Optional from urllib.parse import urljoin import lxml.etree @@ -14,7 +14,7 @@ class Sitemap: """Class to parse Sitemap (type=urlset) and Sitemap Index (type=sitemapindex) files""" - def __init__(self, xmltext): + def __init__(self, xmltext: str): xmlp = lxml.etree.XMLParser( recover=True, remove_comments=True, resolve_entities=False ) @@ -22,9 +22,9 @@ class Sitemap: rt = self._root.tag self.type = self._root.tag.split("}", 1)[1] if "}" in rt else rt - def __iter__(self): + def __iter__(self) -> Iterator[Dict[str, Any]]: for elem in self._root.getchildren(): - d = {} + d: Dict[str, Any] = {} for el in elem.getchildren(): tag = el.tag name = tag.split("}", 1)[1] if "}" in tag else tag @@ -39,11 +39,13 @@ class Sitemap: yield d -def sitemap_urls_from_robots(robots_text, base_url=None): +def sitemap_urls_from_robots( + robots_text: str, base_url: Optional[str] = None +) -> Generator[str, Any, None]: """Return an iterator over all sitemap urls contained in the given robots.txt file """ for line in robots_text.splitlines(): if line.lstrip().lower().startswith("sitemap:"): url = line.split(":", 1)[1].strip() - yield urljoin(base_url, url) + yield urljoin(base_url or "", url) diff --git a/scrapy/utils/ssl.py b/scrapy/utils/ssl.py index 03ae4ba9e..d520ef809 100644 --- a/scrapy/utils/ssl.py +++ b/scrapy/utils/ssl.py @@ -1,4 +1,4 @@ -from typing import Any, Optional, cast +from typing import Any, Optional import OpenSSL._util as pyOpenSSLutil import OpenSSL.SSL @@ -58,9 +58,6 @@ def get_temp_key_info(ssl_object: Any) -> Optional[str]: def get_openssl_version() -> str: - # https://github.com/python/typeshed/issues/10024 - system_openssl_bytes = cast( - bytes, OpenSSL.SSL.SSLeay_version(OpenSSL.SSL.SSLEAY_VERSION) - ) + system_openssl_bytes = OpenSSL.SSL.SSLeay_version(OpenSSL.SSL.SSLEAY_VERSION) system_openssl = system_openssl_bytes.decode("ascii", errors="replace") return f"{OpenSSL.version.__version__} ({system_openssl})" diff --git a/scrapy/utils/trackref.py b/scrapy/utils/trackref.py index 01b980c93..9ff9a273f 100644 --- a/scrapy/utils/trackref.py +++ b/scrapy/utils/trackref.py @@ -12,9 +12,14 @@ alias to object in that case). from collections import defaultdict from operator import itemgetter from time import time -from typing import DefaultDict +from typing import TYPE_CHECKING, Any, DefaultDict, Iterable from weakref import WeakKeyDictionary +if TYPE_CHECKING: + # typing.Self requires Python 3.11 + from typing_extensions import Self + + NoneType = type(None) live_refs: DefaultDict[type, WeakKeyDictionary] = defaultdict(WeakKeyDictionary) @@ -24,13 +29,14 @@ class object_ref: __slots__ = () - def __new__(cls, *args, **kwargs): + def __new__(cls, *args: Any, **kwargs: Any) -> "Self": obj = object.__new__(cls) live_refs[cls][obj] = time() return obj -def format_live_refs(ignore=NoneType): +# using Any as it's hard to type type(None) +def format_live_refs(ignore: Any = NoneType) -> str: """Return a tabular representation of tracked objects""" s = "Live References\n\n" now = time() @@ -44,12 +50,12 @@ def format_live_refs(ignore=NoneType): return s -def print_live_refs(*a, **kw): +def print_live_refs(*a: Any, **kw: Any) -> None: """Print tracked objects""" print(format_live_refs(*a, **kw)) -def get_oldest(class_name): +def get_oldest(class_name: str) -> Any: """Get the oldest object for a specific class name""" for cls, wdict in live_refs.items(): if cls.__name__ == class_name: @@ -58,8 +64,9 @@ def get_oldest(class_name): return min(wdict.items(), key=itemgetter(1))[0] -def iter_all(class_name): +def iter_all(class_name: str) -> Iterable[Any]: """Iterate over all objects of the same class by its class name""" for cls, wdict in live_refs.items(): if cls.__name__ == class_name: return wdict.keys() + return []