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

Full typing for scrapy/spiders. (#6356)

This commit is contained in:
Andrey Rakhmatullin 2024-05-13 14:07:51 +04:00 committed by GitHub
parent 4ed5c5ae91
commit b8e333c8ce
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 172 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
scrapy/spiders/__init__.py
View File

@ -39,7 +39,7 @@ class Spider(object_ref):
def __init__(self, name: Optional[str] = None, **kwargs: Any):
if name is not None:
self.name = name
self.name: str = name
elif not getattr(self, "name", None):
raise ValueError(f"{type(self).__name__} must have a name")
self.__dict__.update(kwargs)
@ -67,8 +67,8 @@ class Spider(object_ref):
return spider
def _set_crawler(self, crawler: Crawler) -> None:
self.crawler = crawler
self.settings = crawler.settings
self.crawler: Crawler = crawler
self.settings: BaseSettings = crawler.settings
crawler.signals.connect(self.close, signals.spider_closed)
def start_requests(self) -> Iterable[Request]:

138
scrapy/spiders/crawl.py
View File

@ -8,9 +8,27 @@ See documentation in docs/topics/spiders.rst
from __future__ import annotations
import copy
from typing import TYPE_CHECKING, AsyncIterable, Awaitable, Sequence
from typing import (
TYPE_CHECKING,
Any,
AsyncIterable,
Awaitable,
Callable,
Dict,
Iterable,
List,
Optional,
Sequence,
Set,
TypeVar,
Union,
cast,
)
from twisted.python.failure import Failure
from scrapy.http import HtmlResponse, Request, Response
from scrapy.link import Link
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import Spider
from scrapy.utils.asyncgen import collect_asyncgen
@ -20,20 +38,32 @@ if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
from scrapy.crawler import Crawler
def _identity(x):
_T = TypeVar("_T")
ProcessLinksT = Callable[[List[Link]], List[Link]]
ProcessRequestT = Callable[[Request, Response], Optional[Request]]
def _identity(x: _T) -> _T:
return x
def _identity_process_request(request, response):
def _identity_process_request(
request: Request, response: Response
) -> Optional[Request]:
return request
def _get_method(method, spider):
def _get_method(
method: Union[Callable, str, None], spider: Spider
) -> Optional[Callable]:
if callable(method):
return method
if isinstance(method, str):
return getattr(spider, method, None)
return None
_default_link_extractor = LinkExtractor()
@ -42,37 +72,46 @@ _default_link_extractor = LinkExtractor()
class Rule:
def __init__(
self,
link_extractor=None,
callback=None,
cb_kwargs=None,
follow=None,
process_links=None,
process_request=None,
errback=None,
link_extractor: Optional[LinkExtractor] = None,
callback: Union[Callable, str, None] = None,
cb_kwargs: Optional[Dict[str, Any]] = None,
follow: Optional[bool] = None,
process_links: Union[ProcessLinksT, str, None] = None,
process_request: Union[ProcessRequestT, str, None] = None,
errback: Union[Callable[[Failure], Any], str, None] = None,
):
self.link_extractor = link_extractor or _default_link_extractor
self.callback = callback
self.errback = errback
self.cb_kwargs = cb_kwargs or {}
self.process_links = process_links or _identity
self.process_request = process_request or _identity_process_request
self.follow = follow if follow is not None else not callback
self.link_extractor: LinkExtractor = link_extractor or _default_link_extractor
self.callback: Union[Callable, str, None] = callback
self.errback: Union[Callable[[Failure], Any], str, None] = errback
self.cb_kwargs: Dict[str, Any] = cb_kwargs or {}
self.process_links: Union[ProcessLinksT, str] = process_links or _identity
self.process_request: Union[ProcessRequestT, str] = (
process_request or _identity_process_request
)
self.follow: bool = follow if follow is not None else not callback
def _compile(self, spider):
def _compile(self, spider: Spider) -> None:
# this replaces method names with methods and we can't express this in type hints
self.callback = _get_method(self.callback, spider)
self.errback = _get_method(self.errback, spider)
self.process_links = _get_method(self.process_links, spider)
self.process_request = _get_method(self.process_request, spider)
self.errback = cast(Callable[[Failure], Any], _get_method(self.errback, spider))
self.process_links = cast(
ProcessLinksT, _get_method(self.process_links, spider)
)
self.process_request = cast(
ProcessRequestT, _get_method(self.process_request, spider)
)
class CrawlSpider(Spider):
rules: Sequence[Rule] = ()
_rules: List[Rule]
_follow_links: bool
def __init__(self, *a, **kw):
def __init__(self, *a: Any, **kw: Any):
super().__init__(*a, **kw)
self._compile_rules()
def _parse(self, response, **kwargs):
def _parse(self, response: Response, **kwargs: Any) -> Any:
return self._parse_response(
response=response,
callback=self.parse_start_url,
@ -80,13 +119,13 @@ class CrawlSpider(Spider):
follow=True,
)
def parse_start_url(self, response, **kwargs):
def parse_start_url(self, response: Response, **kwargs: Any) -> Any:
return []
def process_results(self, response: Response, results: list):
def process_results(self, response: Response, results: Any) -> Any:
return results
def _build_request(self, rule_index, link):
def _build_request(self, rule_index: int, link: Link) -> Request:
return Request(
url=link.url,
callback=self._callback,
@ -94,32 +133,43 @@ class CrawlSpider(Spider):
meta={"rule": rule_index, "link_text": link.text},
)
def _requests_to_follow(self, response):
def _requests_to_follow(self, response: Response) -> Iterable[Optional[Request]]:
if not isinstance(response, HtmlResponse):
return
seen = set()
seen: Set[Link] = set()
for rule_index, rule in enumerate(self._rules):
links = [
links: List[Link] = [
lnk
for lnk in rule.link_extractor.extract_links(response)
if lnk not in seen
]
for link in rule.process_links(links):
for link in cast(ProcessLinksT, rule.process_links)(links):
seen.add(link)
request = self._build_request(rule_index, link)
yield rule.process_request(request, response)
yield cast(ProcessRequestT, rule.process_request)(request, response)
def _callback(self, response, **cb_kwargs):
rule = self._rules[response.meta["rule"]]
def _callback(self, response: Response, **cb_kwargs: Any) -> Any:
rule = self._rules[cast(int, response.meta["rule"])]
return self._parse_response(
response, rule.callback, {**rule.cb_kwargs, **cb_kwargs}, rule.follow
response,
cast(Callable, rule.callback),
{**rule.cb_kwargs, **cb_kwargs},
rule.follow,
)
def _errback(self, failure):
rule = self._rules[failure.request.meta["rule"]]
return self._handle_failure(failure, rule.errback)
def _errback(self, failure: Failure) -> Iterable[Any]:
rule = self._rules[cast(int, failure.request.meta["rule"])] # type: ignore[attr-defined]
return self._handle_failure(
failure, cast(Callable[[Failure], Any], rule.errback)
)
async def _parse_response(self, response, callback, cb_kwargs, follow=True):
async def _parse_response(
self,
response: Response,
callback: Optional[Callable],
cb_kwargs: Dict[str, Any],
follow: bool = True,
) -> AsyncIterable[Any]:
if callback:
cb_res = callback(response, **cb_kwargs) or ()
if isinstance(cb_res, AsyncIterable):
@ -134,21 +184,23 @@ class CrawlSpider(Spider):
for request_or_item in self._requests_to_follow(response):
yield request_or_item
def _handle_failure(self, failure, errback):
def _handle_failure(
self, failure: Failure, errback: Optional[Callable[[Failure], Any]]
) -> Iterable[Any]:
if errback:
results = errback(failure) or ()
yield from iterate_spider_output(results)
def _compile_rules(self):
def _compile_rules(self) -> None:
self._rules = []
for rule in self.rules:
self._rules.append(copy.copy(rule))
self._rules[-1]._compile(self)
@classmethod
def from_crawler(cls, crawler, *args, **kwargs) -> Self:
def from_crawler(cls, crawler: Crawler, *args: Any, **kwargs: Any) -> Self:
spider = super().from_crawler(crawler, *args, **kwargs)
spider._follow_links = crawler.settings.getbool( # type: ignore[attr-defined]
spider._follow_links = crawler.settings.getbool(
"CRAWLSPIDER_FOLLOW_LINKS", True
)
return spider

48
scrapy/spiders/feed.py
View File

@ -5,7 +5,10 @@ for scraping from an XML feed.
See documentation in docs/topics/spiders.rst
"""
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
from scrapy.exceptions import NotConfigured, NotSupported
from scrapy.http import Response, TextResponse
from scrapy.selector import Selector
from scrapy.spiders import Spider
from scrapy.utils.iterators import csviter, xmliter_lxml
@ -22,11 +25,13 @@ class XMLFeedSpider(Spider):
use iternodes, since it's a faster and cleaner.
"""
iterator = "iternodes"
itertag = "item"
namespaces = ()
iterator: str = "iternodes"
itertag: str = "item"
namespaces: Sequence[Tuple[str, str]] = ()
def process_results(self, response, results):
def process_results(
self, response: Response, results: Iterable[Any]
) -> Iterable[Any]:
"""This overridable method is called for each result (item or request)
returned by the spider, and it's intended to perform any last time
processing required before returning the results to the framework core,
@ -36,20 +41,20 @@ class XMLFeedSpider(Spider):
"""
return results
def adapt_response(self, response):
def adapt_response(self, response: Response) -> Response:
"""You can override this function in order to make any changes you want
to into the feed before parsing it. This function must return a
response.
"""
return response
def parse_node(self, response, selector):
def parse_node(self, response: Response, selector: Selector) -> Any:
"""This method must be overridden with your custom spider functionality"""
if hasattr(self, "parse_item"): # backward compatibility
return self.parse_item(response, selector)
raise NotImplementedError
def parse_nodes(self, response, nodes):
def parse_nodes(self, response: Response, nodes: Iterable[Selector]) -> Any:
"""This method is called for the nodes matching the provided tag name
(itertag). Receives the response and an Selector for each node.
Overriding this method is mandatory. Otherwise, you spider won't work.
@ -61,20 +66,25 @@ class XMLFeedSpider(Spider):
ret = iterate_spider_output(self.parse_node(response, selector))
yield from self.process_results(response, ret)
def _parse(self, response, **kwargs):
def _parse(self, response: Response, **kwargs: Any) -> Any:
if not hasattr(self, "parse_node"):
raise NotConfigured(
"You must define parse_node method in order to scrape this XML feed"
)
response = self.adapt_response(response)
nodes: Iterable[Selector]
if self.iterator == "iternodes":
nodes = self._iternodes(response)
elif self.iterator == "xml":
if not isinstance(response, TextResponse):
raise ValueError("Response content isn't text")
selector = Selector(response, type="xml")
self._register_namespaces(selector)
nodes = selector.xpath(f"//{self.itertag}")
elif self.iterator == "html":
if not isinstance(response, TextResponse):
raise ValueError("Response content isn't text")
selector = Selector(response, type="html")
self._register_namespaces(selector)
nodes = selector.xpath(f"//{self.itertag}")
@ -83,12 +93,12 @@ class XMLFeedSpider(Spider):
return self.parse_nodes(response, nodes)
def _iternodes(self, response):
def _iternodes(self, response: Response) -> Iterable[Selector]:
for node in xmliter_lxml(response, self.itertag):
self._register_namespaces(node)
yield node
def _register_namespaces(self, selector):
def _register_namespaces(self, selector: Selector) -> None:
for prefix, uri in self.namespaces:
selector.register_namespace(prefix, uri)
@ -102,27 +112,29 @@ class CSVFeedSpider(Spider):
and the file's headers.
"""
delimiter = (
delimiter: Optional[str] = (
None # When this is None, python's csv module's default delimiter is used
)
quotechar = (
quotechar: Optional[str] = (
None # When this is None, python's csv module's default quotechar is used
)
headers = None
headers: Optional[List[str]] = None
def process_results(self, response, results):
def process_results(
self, response: Response, results: Iterable[Any]
) -> Iterable[Any]:
"""This method has the same purpose as the one in XMLFeedSpider"""
return results
def adapt_response(self, response):
def adapt_response(self, response: Response) -> Response:
"""This method has the same purpose as the one in XMLFeedSpider"""
return response
def parse_row(self, response, row):
def parse_row(self, response: Response, row: Dict[str, str]) -> Any:
"""This method must be overridden with your custom spider functionality"""
raise NotImplementedError
def parse_rows(self, response):
def parse_rows(self, response: Response) -> Any:
"""Receives a response and a dict (representing each row) with a key for
each provided (or detected) header of the CSV file. This spider also
gives the opportunity to override adapt_response and
@ -135,7 +147,7 @@ class CSVFeedSpider(Spider):
ret = iterate_spider_output(self.parse_row(response, row))
yield from self.process_results(response, ret)
def _parse(self, response, **kwargs):
def _parse(self, response: Response, **kwargs: Any) -> Any:
if not hasattr(self, "parse_row"):
raise NotConfigured(
"You must define parse_row method in order to scrape this CSV feed"

14
scrapy/spiders/init.py
View File

@ -1,3 +1,7 @@
from typing import Any, Iterable, Optional, cast
from scrapy import Request
from scrapy.http import Response
from scrapy.spiders import Spider
from scrapy.utils.spider import iterate_spider_output
@ -5,17 +9,17 @@ from scrapy.utils.spider import iterate_spider_output
class InitSpider(Spider):
"""Base Spider with initialization facilities"""
def start_requests(self):
self._postinit_reqs = super().start_requests()
return iterate_spider_output(self.init_request())
def start_requests(self) -> Iterable[Request]:
self._postinit_reqs: Iterable[Request] = super().start_requests()
return cast(Iterable[Request], iterate_spider_output(self.init_request()))
def initialized(self, response=None):
def initialized(self, response: Optional[Response] = None) -> Any:
"""This method must be set as the callback of your last initialization
request. See self.init_request() docstring for more info.
"""
return self.__dict__.pop("_postinit_reqs")
def init_request(self):
def init_request(self) -> Any:
"""This function should return one initialization request, with the
self.initialized method as callback. When the self.initialized method
is called this spider is considered initialized. If you need to perform

49
scrapy/spiders/sitemap.py
View File

@ -2,9 +2,21 @@ from __future__ import annotations
import logging
import re
from typing import TYPE_CHECKING, Any
from typing import (
TYPE_CHECKING,
Any,
Callable,
Dict,
Iterable,
List,
Optional,
Sequence,
Tuple,
Union,
cast,
)
from scrapy.http import Request, XmlResponse
from scrapy.http import Request, Response, XmlResponse
from scrapy.spiders import Spider
from scrapy.utils._compression import _DecompressionMaxSizeExceeded
from scrapy.utils.gz import gunzip, gzip_magic_number
@ -20,10 +32,12 @@ logger = logging.getLogger(__name__)
class SitemapSpider(Spider):
sitemap_urls = ()
sitemap_rules = [("", "parse")]
sitemap_follow = [""]
sitemap_alternate_links = False
sitemap_urls: Sequence[str] = ()
sitemap_rules: Sequence[
Tuple[Union[re.Pattern[str], str], Union[str, Callable]]
] = [("", "parse")]
sitemap_follow: Sequence[Union[re.Pattern[str], str]] = [""]
sitemap_alternate_links: bool = False
_max_size: int
_warn_size: int
@ -38,27 +52,29 @@ class SitemapSpider(Spider):
)
return spider
def __init__(self, *a, **kw):
def __init__(self, *a: Any, **kw: Any):
super().__init__(*a, **kw)
self._cbs = []
self._cbs: List[Tuple[re.Pattern[str], Callable]] = []
for r, c in self.sitemap_rules:
if isinstance(c, str):
c = getattr(self, c)
c = cast(Callable, getattr(self, c))
self._cbs.append((regex(r), c))
self._follow = [regex(x) for x in self.sitemap_follow]
self._follow: List[re.Pattern[str]] = [regex(x) for x in self.sitemap_follow]
def start_requests(self):
def start_requests(self) -> Iterable[Request]:
for url in self.sitemap_urls:
yield Request(url, self._parse_sitemap)
def sitemap_filter(self, entries):
def sitemap_filter(
self, entries: Iterable[Dict[str, Any]]
) -> Iterable[Dict[str, Any]]:
"""This method can be used to filter sitemap entries by their
attributes, for example, you can filter locs with lastmod greater
than a given date (see docs).
"""
yield from entries
def _parse_sitemap(self, response):
def _parse_sitemap(self, response: Response) -> Iterable[Request]:
if response.url.endswith("/robots.txt"):
for url in sitemap_urls_from_robots(response.text, base_url=response.url):
yield Request(url, callback=self._parse_sitemap)
@ -86,7 +102,7 @@ class SitemapSpider(Spider):
yield Request(loc, callback=c)
break
def _get_sitemap_body(self, response):
def _get_sitemap_body(self, response: Response) -> Optional[bytes]:
"""Return the sitemap body contained in the given response,
or None if the response is not a sitemap.
"""
@ -117,15 +133,16 @@ class SitemapSpider(Spider):
# in other word, here, we have plain XML
if response.url.endswith(".xml") or response.url.endswith(".xml.gz"):
return response.body
return None
def regex(x):
def regex(x: Union[re.Pattern[str], str]) -> re.Pattern[str]:
if isinstance(x, str):
return re.compile(x)
return x
def iterloc(it, alt=False):
def iterloc(it: Iterable[Dict[str, Any]], alt: bool = False) -> Iterable[str]:
for d in it:
yield d["loc"]

4
scrapy/utils/sitemap.py
View File

@ -5,7 +5,7 @@ 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 typing import Any, Dict, Generator, Iterator, Optional, Union
from urllib.parse import urljoin
import lxml.etree # nosec
@ -15,7 +15,7 @@ class Sitemap:
"""Class to parse Sitemap (type=urlset) and Sitemap Index
(type=sitemapindex) files"""
def __init__(self, xmltext: str):
def __init__(self, xmltext: Union[str, bytes]):
xmlp = lxml.etree.XMLParser(
recover=True, remove_comments=True, resolve_entities=False
)