RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add typing for scrapy/commands (#6268)

This commit is contained in:
Andrey Rakhmatullin 2024-03-06 17:21:08 +05:00 committed by GitHub
parent bf149356fc
commit 6ecc9e0a34
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 355 additions and 185 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
scrapy/commands/__init__.py
View File

@ -3,61 +3,62 @@ Base class for Scrapy commands
"""
import argparse
import builtins
import os
from pathlib import Path
from typing import Any, Dict, List, Optional
from typing import Any, Dict, Iterable, List, Optional
from twisted.python import failure
from scrapy.crawler import CrawlerProcess
from scrapy.crawler import Crawler, CrawlerProcess
from scrapy.exceptions import UsageError
from scrapy.utils.conf import arglist_to_dict, feed_process_params_from_cli
class ScrapyCommand:
requires_project = False
requires_project: bool = False
crawler_process: Optional[CrawlerProcess] = None
# default settings to be used for this command instead of global defaults
default_settings: Dict[str, Any] = {}
exitcode = 0
exitcode: int = 0
def __init__(self) -> None:
self.settings: Any = None # set in scrapy.cmdline
def set_crawler(self, crawler):
def set_crawler(self, crawler: Crawler) -> None:
if hasattr(self, "_crawler"):
raise RuntimeError("crawler already set")
self._crawler = crawler
self._crawler: Crawler = crawler
def syntax(self):
def syntax(self) -> str:
"""
Command syntax (preferably one-line). Do not include command name.
"""
return ""
def short_desc(self):
def short_desc(self) -> str:
"""
A short description of the command
"""
return ""
def long_desc(self):
def long_desc(self) -> str:
"""A long description of the command. Return short description when not
available. It cannot contain newlines since contents will be formatted
by optparser which removes newlines and wraps text.
"""
return self.short_desc()
def help(self):
def help(self) -> str:
"""An extensive help for the command. It will be shown when using the
"help" command. It can contain newlines since no post-formatting will
be applied to its contents.
"""
return self.long_desc()
def add_options(self, parser):
def add_options(self, parser: argparse.ArgumentParser) -> None:
"""
Populate option parse with options available for this command
"""
@ -92,7 +93,7 @@ class ScrapyCommand:
)
group.add_argument("--pdb", action="store_true", help="enable pdb on failure")
def process_options(self, args, opts):
def process_options(self, args: List[str], opts: argparse.Namespace) -> None:
try:
self.settings.setdict(arglist_to_dict(opts.set), priority="cmdline")
except ValueError:
@ -129,8 +130,8 @@ class BaseRunSpiderCommand(ScrapyCommand):
Common class used to share functionality between the crawl, parse and runspider commands
"""
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
def add_options(self, parser: argparse.ArgumentParser) -> None:
super().add_options(parser)
parser.add_argument(
"-a",
dest="spargs",
@ -162,8 +163,8 @@ class BaseRunSpiderCommand(ScrapyCommand):
help="format to use for dumping items",
)
def process_options(self, args, opts):
ScrapyCommand.process_options(self, args, opts)
def process_options(self, args: List[str], opts: argparse.Namespace) -> None:
super().process_options(args, opts)
try:
opts.spargs = arglist_to_dict(opts.spargs)
except ValueError:
@ -183,7 +184,13 @@ class ScrapyHelpFormatter(argparse.HelpFormatter):
Help Formatter for scrapy command line help messages.
"""
def __init__(self, prog, indent_increment=2, max_help_position=24, width=None):
def __init__(
self,
prog: str,
indent_increment: int = 2,
max_help_position: int = 24,
width: Optional[int] = None,
):
super().__init__(
prog,
indent_increment=indent_increment,
@ -191,11 +198,12 @@ class ScrapyHelpFormatter(argparse.HelpFormatter):
width=width,
)
def _join_parts(self, part_strings):
parts = self.format_part_strings(part_strings)
def _join_parts(self, part_strings: Iterable[str]) -> str:
# scrapy.commands.list shadows builtins.list
parts = self.format_part_strings(builtins.list(part_strings))
return super()._join_parts(parts)
def format_part_strings(self, part_strings):
def format_part_strings(self, part_strings: List[str]) -> List[str]:
"""
Underline and title case command line help message headers.
"""

18
scrapy/commands/bench.py
View File

@ -1,10 +1,14 @@
import argparse
import subprocess # nosec
import sys
import time
from typing import Any, Iterable, List
from urllib.parse import urlencode
import scrapy
from scrapy import Request
from scrapy.commands import ScrapyCommand
from scrapy.http import Response
from scrapy.linkextractors import LinkExtractor
@ -15,26 +19,28 @@ class Command(ScrapyCommand):
"CLOSESPIDER_TIMEOUT": 10,
}
def short_desc(self):
def short_desc(self) -> str:
return "Run quick benchmark test"
def run(self, args, opts):
def run(self, args: List[str], opts: argparse.Namespace) -> None:
with _BenchServer():
assert self.crawler_process
self.crawler_process.crawl(_BenchSpider, total=100000)
self.crawler_process.start()
class _BenchServer:
def __enter__(self):
def __enter__(self) -> None:
from scrapy.utils.test import get_testenv
pargs = [sys.executable, "-u", "-m", "scrapy.utils.benchserver"]
self.proc = subprocess.Popen(
pargs, stdout=subprocess.PIPE, env=get_testenv()
) # nosec
assert self.proc.stdout
self.proc.stdout.readline()
def __exit__(self, exc_type, exc_value, traceback):
def __exit__(self, exc_type, exc_value, traceback) -> None:
self.proc.kill()
self.proc.wait()
time.sleep(0.2)
@ -49,11 +55,11 @@ class _BenchSpider(scrapy.Spider):
baseurl = "http://localhost:8998"
link_extractor = LinkExtractor()
def start_requests(self):
def start_requests(self) -> Iterable[Request]:
qargs = {"total": self.total, "show": self.show}
url = f"{self.baseurl}?{urlencode(qargs, doseq=True)}"
return [scrapy.Request(url, dont_filter=True)]
def parse(self, response):
def parse(self, response: Response) -> Any: # type: ignore[override]
for link in self.link_extractor.extract_links(response):
yield scrapy.Request(link.url, callback=self.parse)

18
scrapy/commands/check.py
View File

@ -1,5 +1,7 @@
import argparse
import time
from collections import defaultdict
from typing import List
from unittest import TextTestResult as _TextTestResult
from unittest import TextTestRunner
@ -10,9 +12,10 @@ from scrapy.utils.misc import load_object, set_environ
class TextTestResult(_TextTestResult):
def printSummary(self, start, stop):
def printSummary(self, start: float, stop: float) -> None:
write = self.stream.write
writeln = self.stream.writeln
# _WritelnDecorator isn't implemented in typeshed yet
writeln = self.stream.writeln # type: ignore[attr-defined]
run = self.testsRun
plural = "s" if run != 1 else ""
@ -42,14 +45,14 @@ class Command(ScrapyCommand):
requires_project = True
default_settings = {"LOG_ENABLED": False}
def syntax(self):
def syntax(self) -> str:
return "[options] <spider>"
def short_desc(self):
def short_desc(self) -> str:
return "Check spider contracts"
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
def add_options(self, parser: argparse.ArgumentParser) -> None:
super().add_options(parser)
parser.add_argument(
"-l",
"--list",
@ -66,7 +69,7 @@ class Command(ScrapyCommand):
help="print contract tests for all spiders",
)
def run(self, args, opts):
def run(self, args: List[str], opts: argparse.Namespace) -> None:
# load contracts
contracts = build_component_list(self.settings.getwithbase("SPIDER_CONTRACTS"))
conman = ContractsManager(load_object(c) for c in contracts)
@ -76,6 +79,7 @@ class Command(ScrapyCommand):
# contract requests
contract_reqs = defaultdict(list)
assert self.crawler_process
spider_loader = self.crawler_process.spider_loader
with set_environ(SCRAPY_CHECK="true"):

14
scrapy/commands/crawl.py
View File

@ -1,3 +1,8 @@
import argparse
from typing import List, cast
from twisted.python.failure import Failure
from scrapy.commands import BaseRunSpiderCommand
from scrapy.exceptions import UsageError
@ -5,13 +10,13 @@ from scrapy.exceptions import UsageError
class Command(BaseRunSpiderCommand):
requires_project = True
def syntax(self):
def syntax(self) -> str:
return "[options] <spider>"
def short_desc(self):
def short_desc(self) -> str:
return "Run a spider"
def run(self, args, opts):
def run(self, args: List[str], opts: argparse.Namespace) -> None:
if len(args) < 1:
raise UsageError()
elif len(args) > 1:
@ -20,10 +25,11 @@ class Command(BaseRunSpiderCommand):
)
spname = args[0]
assert self.crawler_process
crawl_defer = self.crawler_process.crawl(spname, **opts.spargs)
if getattr(crawl_defer, "result", None) is not None and issubclass(
crawl_defer.result.type, Exception
cast(Failure, crawl_defer.result).type, Exception
):
self.exitcode = 1
else:

14
scrapy/commands/edit.py
View File

@ -1,5 +1,7 @@
import argparse
import os
import sys
from typing import List
from scrapy.commands import ScrapyCommand
from scrapy.exceptions import UsageError
@ -9,32 +11,34 @@ class Command(ScrapyCommand):
requires_project = True
default_settings = {"LOG_ENABLED": False}
def syntax(self):
def syntax(self) -> str:
return "<spider>"
def short_desc(self):
def short_desc(self) -> str:
return "Edit spider"
def long_desc(self):
def long_desc(self) -> str:
return (
"Edit a spider using the editor defined in the EDITOR environment"
" variable or else the EDITOR setting"
)
def _err(self, msg):
def _err(self, msg: str) -> None:
sys.stderr.write(msg + os.linesep)
self.exitcode = 1
def run(self, args, opts):
def run(self, args: List[str], opts: argparse.Namespace) -> None:
if len(args) != 1:
raise UsageError()
editor = self.settings["EDITOR"]
assert self.crawler_process
try:
spidercls = self.crawler_process.spider_loader.load(args[0])
except KeyError:
return self._err(f"Spider not found: {args[0]}")
sfile = sys.modules[spidercls.__module__].__file__
assert sfile
sfile = sfile.replace(".pyc", ".py")
self.exitcode = os.system(f'{editor} "{sfile}"') # nosec

23
scrapy/commands/fetch.py
View File

@ -1,13 +1,13 @@
import sys
from argparse import Namespace
from typing import List, Type
from argparse import ArgumentParser, Namespace
from typing import Dict, List, Type
from w3lib.url import is_url
from scrapy import Spider
from scrapy.commands import ScrapyCommand
from scrapy.exceptions import UsageError
from scrapy.http import Request
from scrapy.http import Request, Response
from scrapy.utils.datatypes import SequenceExclude
from scrapy.utils.spider import DefaultSpider, spidercls_for_request
@ -15,20 +15,20 @@ from scrapy.utils.spider import DefaultSpider, spidercls_for_request
class Command(ScrapyCommand):
requires_project = False
def syntax(self):
def syntax(self) -> str:
return "[options] <url>"
def short_desc(self):
def short_desc(self) -> str:
return "Fetch a URL using the Scrapy downloader"
def long_desc(self):
def long_desc(self) -> str:
return (
"Fetch a URL using the Scrapy downloader and print its content"
" to stdout. You may want to use --nolog to disable logging"
)
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
def add_options(self, parser: ArgumentParser) -> None:
super().add_options(parser)
parser.add_argument("--spider", dest="spider", help="use this spider")
parser.add_argument(
"--headers",
@ -44,20 +44,21 @@ class Command(ScrapyCommand):
help="do not handle HTTP 3xx status codes and print response as-is",
)
def _print_headers(self, headers, prefix):
def _print_headers(self, headers: Dict[bytes, List[bytes]], prefix: bytes) -> None:
for key, values in headers.items():
for value in values:
self._print_bytes(prefix + b" " + key + b": " + value)
def _print_response(self, response, opts):
def _print_response(self, response: Response, opts: Namespace) -> None:
if opts.headers:
assert response.request
self._print_headers(response.request.headers, b">")
print(">")
self._print_headers(response.headers, b"<")
else:
self._print_bytes(response.body)
def _print_bytes(self, bytes_):
def _print_bytes(self, bytes_: bytes) -> None:
sys.stdout.buffer.write(bytes_ + b"\n")
def run(self, args: List[str], opts: Namespace) -> None:

31
scrapy/commands/genspider.py
View File

@ -1,9 +1,10 @@
import argparse
import os
import shutil
import string
from importlib import import_module
from pathlib import Path
from typing import Optional, cast
from typing import List, Optional, Union, cast
from urllib.parse import urlparse
import scrapy
@ -12,7 +13,7 @@ from scrapy.exceptions import UsageError
from scrapy.utils.template import render_templatefile, string_camelcase
def sanitize_module_name(module_name):
def sanitize_module_name(module_name: str) -> str:
"""Sanitize the given module name, by replacing dashes and points
with underscores and prefixing it with a letter if it doesn't start
with one
@ -23,7 +24,7 @@ def sanitize_module_name(module_name):
return module_name
def extract_domain(url):
def extract_domain(url: str) -> str:
"""Extract domain name from URL string"""
o = urlparse(url)
if o.scheme == "" and o.netloc == "":
@ -31,7 +32,7 @@ def extract_domain(url):
return o.netloc
def verify_url_scheme(url):
def verify_url_scheme(url: str) -> str:
"""Check url for scheme and insert https if none found."""
parsed = urlparse(url)
if parsed.scheme == "" and parsed.netloc == "":
@ -43,14 +44,14 @@ class Command(ScrapyCommand):
requires_project = False
default_settings = {"LOG_ENABLED": False}
def syntax(self):
def syntax(self) -> str:
return "[options] <name> <domain>"
def short_desc(self):
def short_desc(self) -> str:
return "Generate new spider using pre-defined templates"
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
def add_options(self, parser: argparse.ArgumentParser) -> None:
super().add_options(parser)
parser.add_argument(
"-l",
"--list",
@ -86,7 +87,7 @@ class Command(ScrapyCommand):
help="If the spider already exists, overwrite it with the template",
)
def run(self, args, opts):
def run(self, args: List[str], opts: argparse.Namespace) -> None:
if opts.list:
self._list_templates()
return
@ -115,7 +116,14 @@ class Command(ScrapyCommand):
if opts.edit:
self.exitcode = os.system(f'scrapy edit "{name}"') # nosec
def _genspider(self, module, name, url, template_name, template_file):
def _genspider(
self,
module: str,
name: str,
url: str,
template_name: str,
template_file: Union[str, os.PathLike],
) -> None:
"""Generate the spider module, based on the given template"""
capitalized_module = "".join(s.capitalize() for s in module.split("_"))
domain = extract_domain(url)
@ -130,6 +138,7 @@ class Command(ScrapyCommand):
}
if self.settings.get("NEWSPIDER_MODULE"):
spiders_module = import_module(self.settings["NEWSPIDER_MODULE"])
assert spiders_module.__file__
spiders_dir = Path(spiders_module.__file__).parent.resolve()
else:
spiders_module = None
@ -152,7 +161,7 @@ class Command(ScrapyCommand):
print('Use "scrapy genspider --list" to see all available templates.')
return None
def _list_templates(self):
def _list_templates(self) -> None:
print("Available templates:")
for file in sorted(Path(self.templates_dir).iterdir()):
if file.suffix == ".tmpl":

8
scrapy/commands/list.py
View File

@ -1,3 +1,6 @@
import argparse
from typing import List
from scrapy.commands import ScrapyCommand
@ -5,9 +8,10 @@ class Command(ScrapyCommand):
requires_project = True
default_settings = {"LOG_ENABLED": False}
def short_desc(self):
def short_desc(self) -> str:
return "List available spiders"
def run(self, args, opts):
def run(self, args: List[str], opts: argparse.Namespace) -> None:
assert self.crawler_process
for s in sorted(self.crawler_process.spider_loader.list()):
print(s)

128
scrapy/commands/parse.py
View File

@ -1,16 +1,32 @@
import argparse
import functools
import inspect
import json
import logging
from typing import Dict
from types import CoroutineType
from typing import (
Any,
AsyncGenerator,
Callable,
Dict,
Iterable,
List,
Optional,
Tuple,
TypeVar,
Union,
overload,
)
from itemadapter import ItemAdapter, is_item
from twisted.internet.defer import maybeDeferred
from twisted.internet.defer import Deferred, maybeDeferred
from twisted.python.failure import Failure
from w3lib.url import is_url
from scrapy.commands import BaseRunSpiderCommand
from scrapy.exceptions import UsageError
from scrapy.http import Request
from scrapy.http import Request, Response
from scrapy.spiders import Spider
from scrapy.utils import display
from scrapy.utils.asyncgen import collect_asyncgen
from scrapy.utils.defer import aiter_errback, deferred_from_coro
@ -20,24 +36,26 @@ from scrapy.utils.spider import spidercls_for_request
logger = logging.getLogger(__name__)
_T = TypeVar("_T")
class Command(BaseRunSpiderCommand):
requires_project = True
spider = None
items: Dict[int, list] = {}
requests: Dict[int, list] = {}
items: Dict[int, List[Any]] = {}
requests: Dict[int, List[Request]] = {}
first_response = None
def syntax(self):
def syntax(self) -> str:
return "[options] <url>"
def short_desc(self):
def short_desc(self) -> str:
return "Parse URL (using its spider) and print the results"
def add_options(self, parser):
BaseRunSpiderCommand.add_options(self, parser)
def add_options(self, parser: argparse.ArgumentParser) -> None:
super().add_options(parser)
parser.add_argument(
"--spider",
dest="spider",
@ -106,7 +124,7 @@ class Command(BaseRunSpiderCommand):
)
@property
def max_level(self):
def max_level(self) -> int:
max_items, max_requests = 0, 0
if self.items:
max_items = max(self.items)
@ -114,13 +132,21 @@ class Command(BaseRunSpiderCommand):
max_requests = max(self.requests)
return max(max_items, max_requests)
def handle_exception(self, _failure):
def handle_exception(self, _failure: Failure) -> None:
logger.error(
"An error is caught while iterating the async iterable",
exc_info=failure_to_exc_info(_failure),
)
def iterate_spider_output(self, result):
@overload
def iterate_spider_output(
self, result: Union[AsyncGenerator, CoroutineType]
) -> Deferred: ...
@overload
def iterate_spider_output(self, result: _T) -> Iterable: ...
def iterate_spider_output(self, result: Any) -> Union[Iterable, Deferred]:
if inspect.isasyncgen(result):
d = deferred_from_coro(
collect_asyncgen(aiter_errback(result, self.handle_exception))
@ -133,15 +159,15 @@ class Command(BaseRunSpiderCommand):
return d
return arg_to_iter(deferred_from_coro(result))
def add_items(self, lvl, new_items):
def add_items(self, lvl: int, new_items: List[Any]) -> None:
old_items = self.items.get(lvl, [])
self.items[lvl] = old_items + new_items
def add_requests(self, lvl, new_reqs):
def add_requests(self, lvl: int, new_reqs: List[Request]) -> None:
old_reqs = self.requests.get(lvl, [])
self.requests[lvl] = old_reqs + new_reqs
def print_items(self, lvl=None, colour=True):
def print_items(self, lvl: Optional[int] = None, colour: bool = True) -> None:
if lvl is None:
items = [item for lst in self.items.values() for item in lst]
else:
@ -150,7 +176,7 @@ class Command(BaseRunSpiderCommand):
print("# Scraped Items ", "-" * 60)
display.pprint([ItemAdapter(x).asdict() for x in items], colorize=colour)
def print_requests(self, lvl=None, colour=True):
def print_requests(self, lvl: Optional[int] = None, colour: bool = True) -> None:
if lvl is None:
if self.requests:
requests = self.requests[max(self.requests)]
@ -162,7 +188,7 @@ class Command(BaseRunSpiderCommand):
print("# Requests ", "-" * 65)
display.pprint(requests, colorize=colour)
def print_results(self, opts):
def print_results(self, opts: argparse.Namespace) -> None:
colour = not opts.nocolour
if opts.verbose:
@ -179,7 +205,14 @@ class Command(BaseRunSpiderCommand):
if not opts.nolinks:
self.print_requests(colour=colour)
def _get_items_and_requests(self, spider_output, opts, depth, spider, callback):
def _get_items_and_requests(
self,
spider_output: Iterable[Any],
opts: argparse.Namespace,
depth: int,
spider: Spider,
callback: Callable,
) -> Tuple[List[Any], List[Request], argparse.Namespace, int, Spider, Callable]:
items, requests = [], []
for x in spider_output:
if is_item(x):
@ -188,14 +221,21 @@ class Command(BaseRunSpiderCommand):
requests.append(x)
return items, requests, opts, depth, spider, callback
def run_callback(self, response, callback, cb_kwargs=None):
def run_callback(
self,
response: Response,
callback: Callable,
cb_kwargs: Optional[Dict[str, Any]] = None,
) -> Deferred:
cb_kwargs = cb_kwargs or {}
d = maybeDeferred(self.iterate_spider_output, callback(response, **cb_kwargs))
return d
def get_callback_from_rules(self, spider, response):
def get_callback_from_rules(
self, spider: Spider, response: Response
) -> Union[Callable, str, None]:
if getattr(spider, "rules", None):
for rule in spider.rules:
for rule in spider.rules: # type: ignore[attr-defined]
if rule.link_extractor.matches(response.url):
return rule.callback or "parse"
else:
@ -204,8 +244,10 @@ class Command(BaseRunSpiderCommand):
"please specify a callback to use for parsing",
{"spider": spider.name},
)
return None
def set_spidercls(self, url, opts):
def set_spidercls(self, url: str, opts: argparse.Namespace) -> None:
assert self.crawler_process
spider_loader = self.crawler_process.spider_loader
if opts.spider:
try:
@ -219,13 +261,14 @@ class Command(BaseRunSpiderCommand):
if not self.spidercls:
logger.error("Unable to find spider for: %(url)s", {"url": url})
def _start_requests(spider):
def _start_requests(spider: Spider) -> Iterable[Request]:
yield self.prepare_request(spider, Request(url), opts)
if self.spidercls:
self.spidercls.start_requests = _start_requests
def start_parsing(self, url, opts):
def start_parsing(self, url: str, opts: argparse.Namespace) -> None:
assert self.crawler_process
self.crawler_process.crawl(self.spidercls, **opts.spargs)
self.pcrawler = list(self.crawler_process.crawlers)[0]
self.crawler_process.start()
@ -233,7 +276,12 @@ class Command(BaseRunSpiderCommand):
if not self.first_response:
logger.error("No response downloaded for: %(url)s", {"url": url})
def scraped_data(self, args):
def scraped_data(
self,
args: Tuple[
List[Any], List[Request], argparse.Namespace, int, Spider, Callable
],
) -> List[Any]:
items, requests, opts, depth, spider, callback = args
if opts.pipelines:
itemproc = self.pcrawler.engine.scraper.itemproc
@ -252,8 +300,14 @@ class Command(BaseRunSpiderCommand):
return scraped_data
def _get_callback(self, *, spider, opts, response=None):
cb = None
def _get_callback(
self,
*,
spider: Spider,
opts: argparse.Namespace,
response: Optional[Response] = None,
) -> Callable:
cb: Union[str, Callable, None] = None
if response:
cb = response.meta["_callback"]
if not cb:
@ -270,6 +324,7 @@ class Command(BaseRunSpiderCommand):
cb = "parse"
if not callable(cb):
assert cb is not None
cb_method = getattr(spider, cb, None)
if callable(cb_method):
cb = cb_method
@ -277,10 +332,13 @@ class Command(BaseRunSpiderCommand):
raise ValueError(
f"Cannot find callback {cb!r} in spider: {spider.name}"
)
assert callable(cb)
return cb
def prepare_request(self, spider, request, opts):
def callback(response, **cb_kwargs):
def prepare_request(
self, spider: Spider, request: Request, opts: argparse.Namespace
) -> Request:
def callback(response: Response, **cb_kwargs: Any) -> Deferred:
# memorize first request
if not self.first_response:
self.first_response = response
@ -288,7 +346,7 @@ class Command(BaseRunSpiderCommand):
cb = self._get_callback(spider=spider, opts=opts, response=response)
# parse items and requests
depth = response.meta["_depth"]
depth: int = response.meta["_depth"]
d = self.run_callback(response, cb, cb_kwargs)
d.addCallback(self._get_items_and_requests, opts, depth, spider, callback)
@ -311,13 +369,13 @@ class Command(BaseRunSpiderCommand):
request.callback = callback
return request
def process_options(self, args, opts):
BaseRunSpiderCommand.process_options(self, args, opts)
def process_options(self, args: List[str], opts: argparse.Namespace) -> None:
super().process_options(args, opts)
self.process_request_meta(opts)
self.process_request_cb_kwargs(opts)
def process_request_meta(self, opts):
def process_request_meta(self, opts: argparse.Namespace) -> None:
if opts.meta:
try:
opts.meta = json.loads(opts.meta)
@ -328,7 +386,7 @@ class Command(BaseRunSpiderCommand):
print_help=False,
)
def process_request_cb_kwargs(self, opts):
def process_request_cb_kwargs(self, opts: argparse.Namespace) -> None:
if opts.cbkwargs:
try:
opts.cbkwargs = json.loads(opts.cbkwargs)
@ -339,7 +397,7 @@ class Command(BaseRunSpiderCommand):
print_help=False,
)
def run(self, args, opts):
def run(self, args: List[str], opts: argparse.Namespace) -> None:
# parse arguments
if not len(args) == 1 or not is_url(args[0]):
raise UsageError()

12
scrapy/commands/runspider.py
View File

@ -1,9 +1,10 @@
import argparse
import sys
from importlib import import_module
from os import PathLike
from pathlib import Path
from types import ModuleType
from typing import Union
from typing import List, Union
from scrapy.commands import BaseRunSpiderCommand
from scrapy.exceptions import UsageError
@ -27,16 +28,16 @@ class Command(BaseRunSpiderCommand):
requires_project = False
default_settings = {"SPIDER_LOADER_WARN_ONLY": True}
def syntax(self):
def syntax(self) -> str:
return "[options] <spider_file>"
def short_desc(self):
def short_desc(self) -> str:
return "Run a self-contained spider (without creating a project)"
def long_desc(self):
def long_desc(self) -> str:
return "Run the spider defined in the given file"
def run(self, args, opts):
def run(self, args: List[str], opts: argparse.Namespace) -> None:
if len(args) != 1:
raise UsageError()
filename = Path(args[0])
@ -51,6 +52,7 @@ class Command(BaseRunSpiderCommand):
raise UsageError(f"No spider found in file: {filename}\n")
spidercls = spclasses.pop()
assert self.crawler_process
self.crawler_process.crawl(spidercls, **opts.spargs)
self.crawler_process.start()

13
scrapy/commands/settings.py
View File

@ -1,4 +1,6 @@
import argparse
import json
from typing import List
from scrapy.commands import ScrapyCommand
from scrapy.settings import BaseSettings
@ -8,14 +10,14 @@ class Command(ScrapyCommand):
requires_project = False
default_settings = {"LOG_ENABLED": False, "SPIDER_LOADER_WARN_ONLY": True}
def syntax(self):
def syntax(self) -> str:
return "[options]"
def short_desc(self):
def short_desc(self) -> str:
return "Get settings values"
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
def add_options(self, parser: argparse.ArgumentParser) -> None:
super().add_options(parser)
parser.add_argument(
"--get", dest="get", metavar="SETTING", help="print raw setting value"
)
@ -44,7 +46,8 @@ class Command(ScrapyCommand):
help="print setting value, interpreted as a list",
)
def run(self, args, opts):
def run(self, args: List[str], opts: argparse.Namespace) -> None:
assert self.crawler_process
settings = self.crawler_process.settings
if opts.get:
s = settings.get(opts.get)

19
scrapy/commands/shell.py
View File

@ -4,9 +4,9 @@ Scrapy Shell
See documentation in docs/topics/shell.rst
"""
from argparse import Namespace
from argparse import ArgumentParser, Namespace
from threading import Thread
from typing import List, Type
from typing import Any, Dict, List, Type
from scrapy import Spider
from scrapy.commands import ScrapyCommand
@ -24,20 +24,20 @@ class Command(ScrapyCommand):
"DUPEFILTER_CLASS": "scrapy.dupefilters.BaseDupeFilter",
}
def syntax(self):
def syntax(self) -> str:
return "[url|file]"
def short_desc(self):
def short_desc(self) -> str:
return "Interactive scraping console"
def long_desc(self):
def long_desc(self) -> str:
return (
"Interactive console for scraping the given url or file. "
"Use ./file.html syntax or full path for local file."
)
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
def add_options(self, parser: ArgumentParser) -> None:
super().add_options(parser)
parser.add_argument(
"-c",
dest="code",
@ -52,7 +52,7 @@ class Command(ScrapyCommand):
help="do not handle HTTP 3xx status codes and print response as-is",
)
def update_vars(self, vars):
def update_vars(self, vars: Dict[str, Any]) -> None:
"""You can use this function to update the Scrapy objects that will be
available in the shell
"""
@ -88,7 +88,8 @@ class Command(ScrapyCommand):
shell = Shell(crawler, update_vars=self.update_vars, code=opts.code)
shell.start(url=url, redirect=not opts.no_redirect)
def _start_crawler_thread(self):
def _start_crawler_thread(self) -> None:
assert self.crawler_process
t = Thread(
target=self.crawler_process.start,
kwargs={"stop_after_crawl": False, "install_signal_handlers": False},

22
scrapy/commands/startproject.py
View File

@ -1,3 +1,4 @@
import argparse
import os
import re
import string
@ -5,13 +6,14 @@ from importlib.util import find_spec
from pathlib import Path
from shutil import copy2, copystat, ignore_patterns, move
from stat import S_IWUSR as OWNER_WRITE_PERMISSION
from typing import List, Tuple, Union
import scrapy
from scrapy.commands import ScrapyCommand
from scrapy.exceptions import UsageError
from scrapy.utils.template import render_templatefile, string_camelcase
TEMPLATES_TO_RENDER = (
TEMPLATES_TO_RENDER: Tuple[Tuple[str, ...], ...] = (
("scrapy.cfg",),
("${project_name}", "settings.py.tmpl"),
("${project_name}", "items.py.tmpl"),
@ -22,7 +24,7 @@ TEMPLATES_TO_RENDER = (
IGNORE = ignore_patterns("*.pyc", "__pycache__", ".svn")
def _make_writable(path):
def _make_writable(path: Union[str, os.PathLike]) -> None:
current_permissions = os.stat(path).st_mode
os.chmod(path, current_permissions | OWNER_WRITE_PERMISSION)
@ -31,14 +33,14 @@ class Command(ScrapyCommand):
requires_project = False
default_settings = {"LOG_ENABLED": False, "SPIDER_LOADER_WARN_ONLY": True}
def syntax(self):
def syntax(self) -> str:
return "<project_name> [project_dir]"
def short_desc(self):
def short_desc(self) -> str:
return "Create new project"
def _is_valid_name(self, project_name):
def _module_exists(module_name):
def _is_valid_name(self, project_name: str) -> bool:
def _module_exists(module_name: str) -> bool:
spec = find_spec(module_name)
return spec is not None and spec.loader is not None
@ -53,7 +55,7 @@ class Command(ScrapyCommand):
return True
return False
def _copytree(self, src: Path, dst: Path):
def _copytree(self, src: Path, dst: Path) -> None:
"""
Since the original function always creates the directory, to resolve
the issue a new function had to be created. It's a simple copy and
@ -84,7 +86,7 @@ class Command(ScrapyCommand):
copystat(src, dst)
_make_writable(dst)
def run(self, args, opts):
def run(self, args: List[str], opts: argparse.Namespace) -> None:
if len(args) not in (1, 2):
raise UsageError()
@ -105,7 +107,9 @@ class Command(ScrapyCommand):
return
self._copytree(Path(self.templates_dir), project_dir.resolve())
move(project_dir / "module", project_dir / project_name)
# On 3.8 shutil.move doesn't fully support Path args, but it supports our use case
# See https://bugs.python.org/issue32689
move(project_dir / "module", project_dir / project_name) # type: ignore[arg-type]
for paths in TEMPLATES_TO_RENDER:
tplfile = Path(
project_dir,

13
scrapy/commands/version.py
View File

@ -1,3 +1,6 @@
import argparse
from typing import List
import scrapy
from scrapy.commands import ScrapyCommand
from scrapy.utils.versions import scrapy_components_versions
@ -6,14 +9,14 @@ from scrapy.utils.versions import scrapy_components_versions
class Command(ScrapyCommand):
default_settings = {"LOG_ENABLED": False, "SPIDER_LOADER_WARN_ONLY": True}
def syntax(self):
def syntax(self) -> str:
return "[-v]"
def short_desc(self):
def short_desc(self) -> str:
return "Print Scrapy version"
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
def add_options(self, parser: argparse.ArgumentParser) -> None:
super().add_options(parser)
parser.add_argument(
"--verbose",
"-v",
@ -22,7 +25,7 @@ class Command(ScrapyCommand):
help="also display twisted/python/platform info (useful for bug reports)",
)
def run(self, args, opts):
def run(self, args: List[str], opts: argparse.Namespace) -> None:
if opts.verbose:
versions = scrapy_components_versions()
width = max(len(n) for (n, _) in versions)

15
scrapy/commands/view.py
View File

@ -1,21 +1,28 @@
import argparse
import logging
from scrapy.commands import fetch
from scrapy.http import Response, TextResponse
from scrapy.utils.response import open_in_browser
logger = logging.getLogger(__name__)
class Command(fetch.Command):
def short_desc(self):
def short_desc(self) -> str:
return "Open URL in browser, as seen by Scrapy"
def long_desc(self):
def long_desc(self) -> str:
return (
"Fetch a URL using the Scrapy downloader and show its contents in a browser"
)
def add_options(self, parser):
def add_options(self, parser: argparse.ArgumentParser) -> None:
super().add_options(parser)
parser.add_argument("--headers", help=argparse.SUPPRESS)
def _print_response(self, response, opts):
def _print_response(self, response: Response, opts: argparse.Namespace) -> None:
if not isinstance(response, TextResponse):
logger.error("Cannot view a non-text response.")
return
open_in_browser(response)

70
scrapy/shell.py
View File

@ -6,6 +6,7 @@ See documentation in docs/topics/shell.rst
import os
import signal
from typing import Any, Callable, Dict, Optional, Tuple, Union
from itemadapter import is_item
from twisted.internet import defer, threads
@ -26,18 +27,32 @@ from scrapy.utils.response import open_in_browser
class Shell:
relevant_classes = (Crawler, Spider, Request, Response, Settings)
relevant_classes: Tuple[type, ...] = (Crawler, Spider, Request, Response, Settings)
def __init__(self, crawler, update_vars=None, code=None):
self.crawler = crawler
self.update_vars = update_vars or (lambda x: None)
self.item_class = load_object(crawler.settings["DEFAULT_ITEM_CLASS"])
self.spider = None
self.inthread = not threadable.isInIOThread()
self.code = code
self.vars = {}
def __init__(
self,
crawler: Crawler,
update_vars: Optional[Callable[[Dict[str, Any]], None]] = None,
code: Optional[str] = None,
):
self.crawler: Crawler = crawler
self.update_vars: Callable[[Dict[str, Any]], None] = update_vars or (
lambda x: None
)
self.item_class: type = load_object(crawler.settings["DEFAULT_ITEM_CLASS"])
self.spider: Optional[Spider] = None
self.inthread: bool = not threadable.isInIOThread()
self.code: Optional[str] = code
self.vars: Dict[str, Any] = {}
def start(self, url=None, request=None, response=None, spider=None, redirect=True):
def start(
self,
url: Optional[str] = None,
request: Optional[Request] = None,
response: Optional[Response] = None,
spider: Optional[Spider] = None,
redirect: bool = True,
) -> None:
# disable accidental Ctrl-C key press from shutting down the engine
signal.signal(signal.SIGINT, signal.SIG_IGN)
if url:
@ -77,7 +92,7 @@ class Shell:
self.vars, shells=shells, banner=self.vars.pop("banner", "")
)
def _schedule(self, request, spider):
def _schedule(self, request: Request, spider: Optional[Spider]) -> defer.Deferred:
if is_asyncio_reactor_installed():
# set the asyncio event loop for the current thread
event_loop_path = self.crawler.settings["ASYNCIO_EVENT_LOOP"]
@ -85,10 +100,11 @@ class Shell:
spider = self._open_spider(request, spider)
d = _request_deferred(request)
d.addCallback(lambda x: (x, spider))
assert self.crawler.engine
self.crawler.engine.crawl(request)
return d
def _open_spider(self, request, spider):
def _open_spider(self, request: Request, spider: Optional[Spider]) -> Spider:
if self.spider:
return self.spider
@ -96,11 +112,18 @@ class Shell:
spider = self.crawler.spider or self.crawler._create_spider()
self.crawler.spider = spider
assert self.crawler.engine
self.crawler.engine.open_spider(spider, close_if_idle=False)
self.spider = spider
return spider
def fetch(self, request_or_url, spider=None, redirect=True, **kwargs):
def fetch(
self,
request_or_url: Union[Request, str],
spider: Optional[Spider] = None,
redirect: bool = True,
**kwargs: Any,
) -> None:
from twisted.internet import reactor
if isinstance(request_or_url, Request):
@ -123,7 +146,12 @@ class Shell:
pass
self.populate_vars(response, request, spider)
def populate_vars(self, response=None, request=None, spider=None):
def populate_vars(
self,
response: Optional[Response] = None,
request: Optional[Request] = None,
spider: Optional[Spider] = None,
) -> None:
import scrapy
self.vars["scrapy"] = scrapy
@ -141,10 +169,10 @@ class Shell:
if not self.code:
self.vars["banner"] = self.get_help()
def print_help(self):
def print_help(self) -> None:
print(self.get_help())
def get_help(self):
def get_help(self) -> str:
b = []
b.append("Available Scrapy objects:")
b.append(
@ -168,11 +196,11 @@ class Shell:
return "\n".join(f"[s] {line}" for line in b)
def _is_relevant(self, value):
def _is_relevant(self, value: Any) -> bool:
return isinstance(value, self.relevant_classes) or is_item(value)
def inspect_response(response, spider):
def inspect_response(response: Response, spider: Spider) -> None:
"""Open a shell to inspect the given response"""
# Shell.start removes the SIGINT handler, so save it and re-add it after
# the shell has closed
@ -181,7 +209,7 @@ def inspect_response(response, spider):
signal.signal(signal.SIGINT, sigint_handler)
def _request_deferred(request):
def _request_deferred(request: Request) -> defer.Deferred:
"""Wrap a request inside a Deferred.
This function is harmful, do not use it until you know what you are doing.
@ -195,12 +223,12 @@ def _request_deferred(request):
request_callback = request.callback
request_errback = request.errback
def _restore_callbacks(result):
def _restore_callbacks(result: Any) -> Any:
request.callback = request_callback
request.errback = request_errback
return result
d = defer.Deferred()
d: defer.Deferred = defer.Deferred()
d.addBoth(_restore_callbacks)
if request.callback:
d.addCallbacks(request.callback, request.errback)

48
scrapy/utils/console.py
View File

@ -1,17 +1,27 @@
from functools import wraps
from typing import Any, Callable, Dict, Iterable, Optional
EmbedFuncT = Callable[..., None]
KnownShellsT = Dict[str, Callable[..., EmbedFuncT]]
def _embed_ipython_shell(namespace={}, banner=""):
def _embed_ipython_shell(
namespace: Dict[str, Any] = {}, banner: str = ""
) -> EmbedFuncT:
"""Start an IPython Shell"""
try:
from IPython.terminal.embed import InteractiveShellEmbed
from IPython.terminal.ipapp import load_default_config
except ImportError:
from IPython.frontend.terminal.embed import InteractiveShellEmbed
from IPython.frontend.terminal.ipapp import load_default_config
from IPython.frontend.terminal.embed import ( # type: ignore[no-redef]
InteractiveShellEmbed,
)
from IPython.frontend.terminal.ipapp import ( # type: ignore[no-redef]
load_default_config,
)
@wraps(_embed_ipython_shell)
def wrapper(namespace=namespace, banner=""):
def wrapper(namespace: Dict[str, Any] = namespace, banner: str = "") -> None:
config = load_default_config()
# Always use .instance() to ensure _instance propagation to all parents
# this is needed for <TAB> completion works well for new imports
@ -26,30 +36,36 @@ def _embed_ipython_shell(namespace={}, banner=""):
return wrapper
def _embed_bpython_shell(namespace={}, banner=""):
def _embed_bpython_shell(
namespace: Dict[str, Any] = {}, banner: str = ""
) -> EmbedFuncT:
"""Start a bpython shell"""
import bpython
@wraps(_embed_bpython_shell)
def wrapper(namespace=namespace, banner=""):
def wrapper(namespace: Dict[str, Any] = namespace, banner: str = "") -> None:
bpython.embed(locals_=namespace, banner=banner)
return wrapper
def _embed_ptpython_shell(namespace={}, banner=""):
def _embed_ptpython_shell(
namespace: Dict[str, Any] = {}, banner: str = ""
) -> EmbedFuncT:
"""Start a ptpython shell"""
import ptpython.repl
@wraps(_embed_ptpython_shell)
def wrapper(namespace=namespace, banner=""):
def wrapper(namespace: Dict[str, Any] = namespace, banner: str = "") -> None:
print(banner)
ptpython.repl.embed(locals=namespace)
return wrapper
def _embed_standard_shell(namespace={}, banner=""):
def _embed_standard_shell(
namespace: Dict[str, Any] = {}, banner: str = ""
) -> EmbedFuncT:
"""Start a standard python shell"""
import code
@ -63,13 +79,13 @@ def _embed_standard_shell(namespace={}, banner=""):
readline.parse_and_bind("tab:complete")
@wraps(_embed_standard_shell)
def wrapper(namespace=namespace, banner=""):
def wrapper(namespace: Dict[str, Any] = namespace, banner: str = "") -> None:
code.interact(banner=banner, local=namespace)
return wrapper
DEFAULT_PYTHON_SHELLS = {
DEFAULT_PYTHON_SHELLS: KnownShellsT = {
"ptpython": _embed_ptpython_shell,
"ipython": _embed_ipython_shell,
"bpython": _embed_bpython_shell,
@ -77,7 +93,9 @@ DEFAULT_PYTHON_SHELLS = {
}
def get_shell_embed_func(shells=None, known_shells=None):
def get_shell_embed_func(
shells: Optional[Iterable[str]] = None, known_shells: Optional[KnownShellsT] = None
) -> Any:
"""Return the first acceptable shell-embed function
from a given list of shell names.
"""
@ -95,7 +113,11 @@ def get_shell_embed_func(shells=None, known_shells=None):
continue
def start_python_console(namespace=None, banner="", shells=None):
def start_python_console(
namespace: Optional[Dict[str, Any]] = None,
banner: str = "",
shells: Optional[Iterable[str]] = None,
) -> None:
"""Start Python console bound to the given namespace.
Readline support and tab completion will be used on Unix, if available.
"""

26
scrapy/utils/response.py
View File

@ -3,24 +3,27 @@ This module provides some useful functions for working with
scrapy.http.Response objects
"""
from __future__ import annotations
import os
import re
import tempfile
import webbrowser
from typing import Any, Callable, Iterable, Tuple, Union
from typing import TYPE_CHECKING, Any, Callable, Iterable, Tuple, Union
from weakref import WeakKeyDictionary
from twisted.web import http
from w3lib import html
import scrapy
from scrapy.http.response import Response
from scrapy.utils.python import to_bytes, to_unicode
_baseurl_cache: "WeakKeyDictionary[Response, str]" = WeakKeyDictionary()
if TYPE_CHECKING:
from scrapy.http import Response, TextResponse
_baseurl_cache: WeakKeyDictionary[Response, str] = WeakKeyDictionary()
def get_base_url(response: "scrapy.http.response.text.TextResponse") -> str:
def get_base_url(response: TextResponse) -> str:
"""Return the base url of the given response, joined with the response url"""
if response not in _baseurl_cache:
text = response.text[0:4096]
@ -30,13 +33,13 @@ def get_base_url(response: "scrapy.http.response.text.TextResponse") -> str:
return _baseurl_cache[response]
_metaref_cache: (
"WeakKeyDictionary[Response, Union[Tuple[None, None], Tuple[float, str]]]"
) = WeakKeyDictionary()
_metaref_cache: WeakKeyDictionary[
Response, Union[Tuple[None, None], Tuple[float, str]]
] = WeakKeyDictionary()
def get_meta_refresh(
response: "scrapy.http.response.text.TextResponse",
response: TextResponse,
ignore_tags: Iterable[str] = ("script", "noscript"),
) -> Union[Tuple[None, None], Tuple[float, str]]:
"""Parse the http-equiv refresh parameter from the given response"""
@ -68,10 +71,7 @@ def _remove_html_comments(body):
def open_in_browser(
response: Union[
"scrapy.http.response.html.HtmlResponse",
"scrapy.http.response.text.TextResponse",
],
response: TextResponse,
_openfunc: Callable[[str], Any] = webbrowser.open,
) -> Any:
"""Open *response* in a local web browser, adjusting the `base tag`_ for