From d119790041580a303f1325f8b321f56dcce58882 Mon Sep 17 00:00:00 2001 From: shengchenyang <15538221825@163.com> Date: Fri, 17 May 2024 16:36:58 +0800 Subject: [PATCH] style: add type hint --- ayugespidertools/commands/version.py | 7 ++- ayugespidertools/common/typevars.py | 5 +- ayugespidertools/utils/cmdline.py | 60 ++++++++++++++------ tests/test_commands/test_commands_crawl.py | 5 +- tests/test_commands/test_commands_version.py | 9 ++- tests/test_spider.py | 2 +- 6 files changed, 61 insertions(+), 27 deletions(-) diff --git a/ayugespidertools/commands/version.py b/ayugespidertools/commands/version.py index 2ac0a99..afd87a2 100644 --- a/ayugespidertools/commands/version.py +++ b/ayugespidertools/commands/version.py @@ -1,11 +1,14 @@ +import argparse +from typing import List + from scrapy.commands.version import Command from ayugespidertools import __version__ class AyuCommand(Command): - def short_desc(self): + def short_desc(self) -> str: return "Print AyugeSpiderTools version" - def run(self, args, opts): + def run(self, args: List[str], opts: argparse.Namespace) -> None: print(f"AyugeSpiderTools {__version__}") diff --git a/ayugespidertools/common/typevars.py b/ayugespidertools/common/typevars.py index e5a9ce7..37fc886 100644 --- a/ayugespidertools/common/typevars.py +++ b/ayugespidertools/common/typevars.py @@ -6,11 +6,10 @@ from sqlalchemy import create_engine if TYPE_CHECKING: - import logging - from loguru import Logger + from scrapy.utils.log import SpiderLoggerAdapter - slogT = Union[Logger, logging.LoggerAdapter] + slogT = Union[Logger, SpiderLoggerAdapter] NoneType = type(None) I_Str = TypeVar("I_Str", int, str) diff --git a/ayugespidertools/utils/cmdline.py b/ayugespidertools/utils/cmdline.py index 7a36242..d0b9280 100644 --- a/ayugespidertools/utils/cmdline.py +++ b/ayugespidertools/utils/cmdline.py @@ -1,30 +1,42 @@ +from __future__ import annotations + import argparse import cProfile import inspect import os import sys from importlib.metadata import entry_points +from typing import TYPE_CHECKING, Callable, Dict, Iterable, List, Optional, Tuple, Type from scrapy.commands import BaseRunSpiderCommand, ScrapyCommand, ScrapyHelpFormatter from scrapy.crawler import CrawlerProcess from scrapy.exceptions import UsageError +from scrapy.settings import BaseSettings, Settings from scrapy.utils.misc import walk_modules from scrapy.utils.project import get_project_settings, inside_project from scrapy.utils.python import garbage_collect from ayugespidertools import __version__ +if TYPE_CHECKING: + # typing.ParamSpec requires Python 3.10 + from typing_extensions import ParamSpec + + _P = ParamSpec("_P") + class ScrapyArgumentParser(argparse.ArgumentParser): - def _parse_optional(self, arg_string): - # if starts with -: it means that is a parameter not an argument + def _parse_optional( + self, arg_string: str + ) -> Optional[Tuple[Optional[argparse.Action], str, Optional[str]]]: + # if starts with -: it means that is a parameter not a argument if arg_string[:2] == "-:": return None return super()._parse_optional(arg_string) -def _iter_command_classes(module_name): +def _iter_command_classes(module_name: str) -> Iterable[Type[ScrapyCommand]]: # TODO: add `name` attribute to commands and merge this function with # scrapy.utils.spider.iter_spider_classes for module in walk_modules(module_name): @@ -38,8 +50,8 @@ def _iter_command_classes(module_name): yield obj -def _get_commands_from_module(module, inproject): - d = {} +def _get_commands_from_module(module: str, inproject: bool) -> Dict[str, ScrapyCommand]: + d: Dict[str, ScrapyCommand] = {} for cmd in _iter_command_classes(module): if inproject or not cmd.requires_project: cmdname = cmd.__module__.split(".")[-1] @@ -47,8 +59,10 @@ def _get_commands_from_module(module, inproject): return d -def _get_commands_from_entry_points(inproject, group="ayugespidertools.commands"): - cmds = {} +def _get_commands_from_entry_points( + inproject: bool, group: str = "ayugespidertools.commands" +) -> Dict[str, ScrapyCommand]: + cmds: Dict[str, ScrapyCommand] = {} if sys.version_info >= (3, 10): eps = entry_points(group=group) else: @@ -62,7 +76,9 @@ def _get_commands_from_entry_points(inproject, group="ayugespidertools.commands" return cmds -def _get_commands_dict(settings, inproject): +def _get_commands_dict( + settings: BaseSettings, inproject: bool +) -> Dict[str, ScrapyCommand]: cmds = _get_commands_from_module("ayugespidertools.commands", inproject) cmds.update(_get_commands_from_entry_points(inproject)) cmds_module = settings["COMMANDS_MODULE"] @@ -71,16 +87,17 @@ def _get_commands_dict(settings, inproject): return cmds -def _pop_command_name(argv): +def _pop_command_name(argv: List[str]) -> Optional[str]: i = 0 for arg in argv[1:]: if not arg.startswith("-"): del argv[i] return arg i += 1 + return None -def _print_header(settings, inproject): +def _print_header(settings: BaseSettings, inproject: bool) -> None: if inproject: print( f"AyugeSpiderTools {__version__} - active project: {settings['BOT_NAME']}\n" @@ -89,7 +106,7 @@ def _print_header(settings, inproject): print(f"AyugeSpiderTools {__version__} - no active project\n") -def _print_commands(settings, inproject): +def _print_commands(settings: BaseSettings, inproject: bool) -> None: _print_header(settings, inproject) print("Usage:") print(" ayuge [options] [args]\n") @@ -104,13 +121,20 @@ def _print_commands(settings, inproject): print('Use "ayuge -h" to see more info about a command') -def _print_unknown_command(settings, cmdname, inproject): +def _print_unknown_command( + settings: BaseSettings, cmdname: str, inproject: bool +) -> None: _print_header(settings, inproject) print(f"Unknown command: {cmdname}\n") print('Use "ayuge" to see available commands') -def _run_print_help(parser, func, *a, **kw): +def _run_print_help( + parser: argparse.ArgumentParser, + func: Callable[_P, None], + *a: _P.args, + **kw: _P.kwargs, +) -> None: try: func(*a, **kw) except UsageError as e: @@ -121,7 +145,9 @@ def _run_print_help(parser, func, *a, **kw): sys.exit(2) -def execute(argv=None, settings=None): +def execute( + argv: Optional[List[str]] = None, settings: Optional[Settings] = None +) -> None: if argv is None: argv = sys.argv @@ -163,14 +189,16 @@ def execute(argv=None, settings=None): sys.exit(cmd.exitcode) -def _run_command(cmd, args, opts): +def _run_command(cmd: ScrapyCommand, args: List[str], opts: argparse.Namespace) -> None: if opts.profile: _run_command_profiled(cmd, args, opts) else: cmd.run(args, opts) -def _run_command_profiled(cmd, args, opts): +def _run_command_profiled( + cmd: ScrapyCommand, args: List[str], opts: argparse.Namespace +) -> None: if opts.profile: sys.stderr.write( f"ayugespidertools: writing cProfile stats to {opts.profile!r}\n" diff --git a/tests/test_commands/test_commands_crawl.py b/tests/test_commands/test_commands_crawl.py index 0e1d00a..cb7a0d0 100644 --- a/tests/test_commands/test_commands_crawl.py +++ b/tests/test_commands/test_commands_crawl.py @@ -2,10 +2,9 @@ import re import subprocess import sys -import tempfile from pathlib import Path from shutil import rmtree -from tempfile import mkdtemp +from tempfile import TemporaryFile, mkdtemp from threading import Timer from typing import Optional, Union @@ -28,7 +27,7 @@ def tearDown(self): rmtree(self.temp_path) def call(self, *new_args, **kwargs): - with tempfile.TemporaryFile() as out: + with TemporaryFile() as out: args = (sys.executable, "-m", "ayugespidertools.utils.cmdline") + new_args return subprocess.call( args, stdout=out, stderr=out, cwd=self.cwd, env=self.env, **kwargs diff --git a/tests/test_commands/test_commands_version.py b/tests/test_commands/test_commands_version.py index 67b5e8e..f3e8b8b 100644 --- a/tests/test_commands/test_commands_version.py +++ b/tests/test_commands/test_commands_version.py @@ -1,3 +1,4 @@ +import argparse import sys from io import StringIO from pathlib import Path @@ -15,8 +16,12 @@ def test_version(): cmd = AyuCommand() - output = StringIO() - cmd.run([], {"stdout": output}) + namespace = argparse.Namespace() + options = {"stdout": StringIO()} + for key, value in options.items(): + setattr(namespace, key, value) + + cmd.run([], namespace) assert cmd.short_desc() == "Print AyugeSpiderTools version" diff --git a/tests/test_spider.py b/tests/test_spider.py index e17de59..9a616b9 100644 --- a/tests/test_spider.py +++ b/tests/test_spider.py @@ -236,7 +236,7 @@ class _CrawlSpider(self.spider_class): rules = (Rule(LinkExtractor(), process_links="dummy_process_links"),) def dummy_process_links(self, links): - return links + yield from links spider = _CrawlSpider() output = list(spider._requests_to_follow(response))