More typing for scrapy/utils.

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)
 

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):

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}'. "

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
     """

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

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

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)

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)

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})"

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 []