Merge pull request #7487 from wRAR/deprecation-removals

Remove most of the 2.13.0 deprecations
This commit is contained in:
Andrey Rakhmatullin 2026-05-04 12:33:21 +05:00 committed by GitHub
commit 9ca206da64
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
16 changed files with 25 additions and 689 deletions

View File

@ -23,9 +23,6 @@ def _py_files(folder):
collect_ignore = [
# may need extra deps
"docs/_ext",
# not a test, but looks like a test
"scrapy/utils/testproc.py",
"scrapy/utils/testsite.py",
# contains scripts to be run by tests/test_crawler.py::AsyncCrawlerProcessSubprocess
*_py_files("tests/AsyncCrawlerProcess"),
# contains scripts to be run by tests/test_crawler.py::AsyncCrawlerRunnerSubprocess

View File

@ -809,7 +809,6 @@ Default:
"scrapy.downloadermiddlewares.defaultheaders.DefaultHeadersMiddleware": 400,
"scrapy.downloadermiddlewares.useragent.UserAgentMiddleware": 500,
"scrapy.downloadermiddlewares.retry.RetryMiddleware": 550,
"scrapy.downloadermiddlewares.ajaxcrawl.AjaxCrawlMiddleware": 560,
"scrapy.downloadermiddlewares.redirect.MetaRefreshMiddleware": 580,
"scrapy.downloadermiddlewares.httpcompression.HttpCompressionMiddleware": 590,
"scrapy.downloadermiddlewares.redirect.RedirectMiddleware": 600,

View File

@ -122,22 +122,9 @@ implicit_reexport = true
module = "scrapy.settings.default_settings"
ignore_errors = true
# deprecated modules
[[tool.mypy.overrides]]
module = [
"scrapy.spiders.init",
"scrapy.utils.testsite",
]
allow_any_generics = true
allow_untyped_calls = true
allow_untyped_defs = true
check_untyped_defs = false
warn_return_any = false
# usually no type hints
[[tool.mypy.overrides]]
module = [
# "IPython.*",
"bpython",
"brotli",
"brotlicffi",

View File

@ -5,16 +5,13 @@ import logging
from abc import abstractmethod
from pathlib import Path
from typing import TYPE_CHECKING, Any
from warnings import warn
# working around https://github.com/sphinx-doc/sphinx/issues/10400
from twisted.internet.defer import Deferred # noqa: TC002
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.spiders import Spider # noqa: TC001
from scrapy.utils.job import job_dir
from scrapy.utils.misc import build_from_crawler, load_object
from scrapy.utils.python import global_object_name
if TYPE_CHECKING:
# requires queuelib >= 1.6.2
@ -450,28 +447,13 @@ class Scheduler(BaseScheduler):
"""Create a new priority queue instance, with in-memory storage"""
assert self.crawler
assert self.pqclass
try:
return build_from_crawler(
self.pqclass,
self.crawler,
downstream_queue_cls=self.mqclass,
key="",
start_queue_cls=self._smqclass,
)
except TypeError: # pragma: no cover
warn(
f"The __init__ method of {global_object_name(self.pqclass)} "
"does not support a `start_queue_cls` keyword-only "
"parameter.",
ScrapyDeprecationWarning,
stacklevel=2,
)
return build_from_crawler(
self.pqclass,
self.crawler,
downstream_queue_cls=self.mqclass,
key="",
)
return build_from_crawler(
self.pqclass,
self.crawler,
downstream_queue_cls=self.mqclass,
key="",
start_queue_cls=self._smqclass,
)
def _dq(self) -> ScrapyPriorityQueue:
"""Create a new priority queue instance, with disk storage"""
@ -479,30 +461,14 @@ class Scheduler(BaseScheduler):
assert self.dqdir
assert self.pqclass
state = self._read_dqs_state(self.dqdir)
try:
q = build_from_crawler(
self.pqclass,
self.crawler,
downstream_queue_cls=self.dqclass,
key=self.dqdir,
startprios=state,
start_queue_cls=self._sdqclass,
)
except TypeError: # pragma: no cover
warn(
f"The __init__ method of {global_object_name(self.pqclass)} "
"does not support a `start_queue_cls` keyword-only "
"parameter.",
ScrapyDeprecationWarning,
stacklevel=2,
)
q = build_from_crawler(
self.pqclass,
self.crawler,
downstream_queue_cls=self.dqclass,
key=self.dqdir,
startprios=state,
)
q = build_from_crawler(
self.pqclass,
self.crawler,
downstream_queue_cls=self.dqclass,
key=self.dqdir,
startprios=state,
start_queue_cls=self._sdqclass,
)
if q:
logger.info(
"Resuming crawl (%(queuesize)d requests scheduled)",

View File

@ -23,7 +23,6 @@ from scrapy.exceptions import (
from scrapy.http import Request, Response
from scrapy.pipelines import ItemPipelineManager
from scrapy.utils.asyncio import _parallel_asyncio, is_asyncio_available
from scrapy.utils.decorators import _warn_spider_arg
from scrapy.utils.defer import (
_defer_sleep_async,
_schedule_coro,
@ -178,9 +177,7 @@ class Scraper:
self.itemproc.open_spider(self.crawler.spider)
)
def close_spider(
self, spider: Spider | None = None
) -> Deferred[None]: # pragma: no cover
def close_spider(self) -> Deferred[None]: # pragma: no cover
warnings.warn(
"Scraper.close_spider() is deprecated, use close_spider_async() instead",
ScrapyDeprecationWarning,
@ -217,9 +214,8 @@ class Scraper:
self.slot.closing.callback(self.crawler.spider)
@inlineCallbacks
@_warn_spider_arg
def enqueue_scrape(
self, result: Response | Failure, request: Request, spider: Spider | None = None
self, result: Response | Failure, request: Request
) -> Generator[Deferred[Any], Any, None]:
if self.slot is None:
raise RuntimeError("Scraper slot not assigned")
@ -349,13 +345,11 @@ class Scraper:
)
return await ensure_awaitable(iterate_spider_output(output))
@_warn_spider_arg
def handle_spider_error(
self,
_failure: Failure,
request: Request,
response: Response | Failure,
spider: Spider | None = None,
) -> None:
"""Handle an exception raised by a spider callback or errback."""
assert self.crawler.spider
@ -391,7 +385,6 @@ class Scraper:
result: Iterable[_T] | AsyncIterator[_T],
request: Request,
response: Response | Failure,
spider: Spider | None = None,
) -> Deferred[None]: # pragma: no cover
"""Pass items/requests produced by a callback to ``_process_spidermw_output()`` in parallel."""
warnings.warn(

View File

@ -1,114 +0,0 @@
from __future__ import annotations
import logging
import re
from typing import TYPE_CHECKING
from warnings import warn
from w3lib import html
from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
from scrapy.http import HtmlResponse, Response
from scrapy.utils.url import escape_ajax
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
from scrapy import Request, Spider
from scrapy.crawler import Crawler
from scrapy.settings import BaseSettings
logger = logging.getLogger(__name__)
class AjaxCrawlMiddleware:
"""
Handle 'AJAX crawlable' pages marked as crawlable via meta tag.
"""
def __init__(self, settings: BaseSettings):
if not settings.getbool("AJAXCRAWL_ENABLED"):
raise NotConfigured
warn(
"scrapy.downloadermiddlewares.ajaxcrawl.AjaxCrawlMiddleware is deprecated"
" and will be removed in a future Scrapy version.",
ScrapyDeprecationWarning,
stacklevel=2,
)
# XXX: Google parses at least first 100k bytes; scrapy's redirect
# middleware parses first 4k. 4k turns out to be insufficient
# for this middleware, and parsing 100k could be slow.
# We use something in between (32K) by default.
self.lookup_bytes: int = settings.getint("AJAXCRAWL_MAXSIZE")
@classmethod
def from_crawler(cls, crawler: Crawler) -> Self:
return cls(crawler.settings)
def process_response(
self, request: Request, response: Response, spider: Spider
) -> Request | Response:
if not isinstance(response, HtmlResponse) or response.status != 200:
return response
if request.method != "GET":
# other HTTP methods are either not safe or don't have a body
return response
if "ajax_crawlable" in request.meta: # prevent loops
return response
if not self._has_ajax_crawlable_variant(response):
return response
ajax_crawl_request = request.replace(url=escape_ajax(request.url + "#!"))
logger.debug(
"Downloading AJAX crawlable %(ajax_crawl_request)s instead of %(request)s",
{"ajax_crawl_request": ajax_crawl_request, "request": request},
extra={"spider": spider},
)
ajax_crawl_request.meta["ajax_crawlable"] = True
return ajax_crawl_request
def _has_ajax_crawlable_variant(self, response: Response) -> bool:
"""
Return True if a page without hash fragment could be "AJAX crawlable".
"""
body = response.text[: self.lookup_bytes]
return _has_ajaxcrawlable_meta(body)
_ajax_crawlable_re: re.Pattern[str] = re.compile(
r'<meta\s+name=["\']fragment["\']\s+content=["\']!["\']/?>'
)
def _has_ajaxcrawlable_meta(text: str) -> bool:
"""
>>> _has_ajaxcrawlable_meta('<html><head><meta name="fragment" content="!"/></head><body></body></html>')
True
>>> _has_ajaxcrawlable_meta("<html><head><meta name='fragment' content='!'></head></html>")
True
>>> _has_ajaxcrawlable_meta('<html><head><!--<meta name="fragment" content="!"/>--></head><body></body></html>')
False
>>> _has_ajaxcrawlable_meta('<html></html>')
False
"""
# Stripping scripts and comments is slow (about 20x slower than
# just checking if a string is in text); this is a quick fail-fast
# path that should work for most pages.
if "fragment" not in text:
return False
if "content" not in text:
return False
text = html.remove_tags_with_content(text, ("script", "noscript"))
text = html.replace_entities(text)
text = html.remove_comments(text)
return _ajax_crawlable_re.search(text) is not None

View File

@ -284,7 +284,6 @@ DOWNLOADER_MIDDLEWARES_BASE = {
"scrapy.downloadermiddlewares.defaultheaders.DefaultHeadersMiddleware": 400,
"scrapy.downloadermiddlewares.useragent.UserAgentMiddleware": 500,
"scrapy.downloadermiddlewares.retry.RetryMiddleware": 550,
"scrapy.downloadermiddlewares.ajaxcrawl.AjaxCrawlMiddleware": 560,
"scrapy.downloadermiddlewares.redirect.MetaRefreshMiddleware": 580,
"scrapy.downloadermiddlewares.httpcompression.HttpCompressionMiddleware": 590,
"scrapy.downloadermiddlewares.redirect.RedirectMiddleware": 600,

View File

@ -1,64 +0,0 @@
from __future__ import annotations
import warnings
from typing import TYPE_CHECKING, Any, cast
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.spiders import Spider
from scrapy.utils.spider import iterate_spider_output
if TYPE_CHECKING:
from collections.abc import AsyncIterator, Iterable
from scrapy import Request
from scrapy.http import Response
class InitSpider(Spider):
"""Base Spider with initialization facilities
.. warning:: This class is deprecated. Copy its code into your project if needed.
It will be removed in a future Scrapy version.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
warnings.warn(
"InitSpider is deprecated. Copy its code from Scrapy's source if needed. "
"Will be removed in a future version.",
ScrapyDeprecationWarning,
stacklevel=2,
)
async def start(self) -> AsyncIterator[Any]:
with warnings.catch_warnings():
warnings.filterwarnings(
"ignore", category=ScrapyDeprecationWarning, module=r"^scrapy\.spiders$"
)
for item_or_request in self.start_requests():
yield item_or_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: Response | None = 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) -> 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
several requests for initializing your spider, you can do so by using
different callbacks. The only requirement is that the final callback
(of the last initialization request) must be self.initialized.
The default implementation calls self.initialized immediately, and
means that no initialization is needed. This method should be
overridden only when you need to perform requests to initialize your
spider
"""
return self.initialized()

View File

@ -7,20 +7,14 @@ from __future__ import annotations
import asyncio
import os
import warnings
from ftplib import FTP
from importlib import import_module
from pathlib import Path
from posixpath import split
from typing import TYPE_CHECKING, Any, TypeVar, cast
from unittest import mock
from twisted.trial.unittest import SkipTest
from twisted.web.client import Agent
from scrapy.crawler import AsyncCrawlerRunner, CrawlerRunner, CrawlerRunnerBase
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.utils.boto import is_botocore_available
from scrapy.utils.deprecate import create_deprecated_class
from scrapy.utils.reactor import is_asyncio_reactor_installed, is_reactor_installed
from scrapy.utils.spider import DefaultSpider
@ -37,80 +31,6 @@ if TYPE_CHECKING:
_T = TypeVar("_T")
def assert_gcs_environ() -> None: # pragma: no cover
warnings.warn(
"The assert_gcs_environ() function is deprecated and will be removed in a future version of Scrapy."
" Check GCS_PROJECT_ID directly.",
category=ScrapyDeprecationWarning,
stacklevel=2,
)
if "GCS_PROJECT_ID" not in os.environ:
raise SkipTest("GCS_PROJECT_ID not found")
def skip_if_no_boto() -> None: # pragma: no cover
warnings.warn(
"The skip_if_no_boto() function is deprecated and will be removed in a future version of Scrapy."
" Check scrapy.utils.boto.is_botocore_available() directly.",
category=ScrapyDeprecationWarning,
stacklevel=2,
)
if not is_botocore_available():
raise SkipTest("missing botocore library")
def get_gcs_content_and_delete(
bucket: Any, path: str
) -> tuple[bytes, list[dict[str, str]], Any]: # pragma: no cover
from google.cloud import storage # noqa: PLC0415
warnings.warn(
"The get_gcs_content_and_delete() function is deprecated and will be removed in a future version of Scrapy.",
category=ScrapyDeprecationWarning,
stacklevel=2,
)
client = storage.Client(project=os.environ.get("GCS_PROJECT_ID"))
bucket = client.get_bucket(bucket)
blob = bucket.get_blob(path)
content = blob.download_as_string()
acl = list(blob.acl) # loads acl before it will be deleted
bucket.delete_blob(path)
return content, acl, blob
def get_ftp_content_and_delete(
path: str,
host: str,
port: int,
username: str,
password: str,
use_active_mode: bool = False,
) -> bytes: # pragma: no cover
warnings.warn(
"The get_ftp_content_and_delete() function is deprecated and will be removed in a future version of Scrapy.",
category=ScrapyDeprecationWarning,
stacklevel=2,
)
ftp = FTP()
ftp.connect(host, port)
ftp.login(username, password)
if use_active_mode:
ftp.set_pasv(False)
ftp_data: list[bytes] = []
def buffer_data(data: bytes) -> None:
ftp_data.append(data)
ftp.retrbinary(f"RETR {path}", buffer_data)
dirname, filename = split(path)
ftp.cwd(dirname)
ftp.delete(filename)
return b"".join(ftp_data)
TestSpider = create_deprecated_class("TestSpider", DefaultSpider)
def get_reactor_settings() -> dict[str, Any]:
"""Return a settings dict that works with the installed reactor.
@ -185,29 +105,6 @@ def get_from_asyncio_queue(value: _T) -> Awaitable[_T]:
return getter
def mock_google_cloud_storage() -> tuple[Any, Any, Any]: # pragma: no cover
"""Creates autospec mocks for google-cloud-storage Client, Bucket and Blob
classes and set their proper return values.
"""
from google.cloud.storage import Blob, Bucket, Client # noqa: PLC0415
warnings.warn(
"The mock_google_cloud_storage() function is deprecated and will be removed in a future version of Scrapy.",
category=ScrapyDeprecationWarning,
stacklevel=2,
)
client_mock = mock.create_autospec(Client)
bucket_mock = mock.create_autospec(Bucket)
client_mock.get_bucket.return_value = bucket_mock
blob_mock = mock.create_autospec(Blob)
bucket_mock.blob.return_value = blob_mock
return (client_mock, bucket_mock, blob_mock)
def get_web_client_agent_req(url: str) -> Deferred[TxResponse]: # pragma: no cover
warnings.warn(
"The get_web_client_agent_req() function is deprecated"

View File

@ -1,78 +0,0 @@
# pragma: no file cover
from __future__ import annotations
import os
import sys
import warnings
from typing import TYPE_CHECKING, ClassVar, cast
from twisted.internet.defer import Deferred
from twisted.internet.protocol import ProcessProtocol
from scrapy.exceptions import ScrapyDeprecationWarning
if TYPE_CHECKING:
from collections.abc import Iterable
from twisted.internet.error import ProcessTerminated
from twisted.python.failure import Failure
warnings.warn(
"The scrapy.utils.testproc module is deprecated.",
ScrapyDeprecationWarning,
stacklevel=2,
)
class ProcessTest:
command: str | None = None
prefix: ClassVar[list[str]] = [sys.executable, "-m", "scrapy.cmdline"]
cwd = os.getcwd() # trial chdirs to temp dir # noqa: PTH109
def execute(
self,
args: Iterable[str],
check_code: bool = True,
settings: str | None = None,
) -> Deferred[TestProcessProtocol]:
from twisted.internet import reactor
env = os.environ.copy()
if settings is not None:
env["SCRAPY_SETTINGS_MODULE"] = settings
assert self.command
cmd = [*self.prefix, self.command, *args]
pp = TestProcessProtocol()
pp.deferred.addCallback(self._process_finished, cmd, check_code)
reactor.spawnProcess(pp, cmd[0], cmd, env=env, path=self.cwd)
return pp.deferred
def _process_finished(
self, pp: TestProcessProtocol, cmd: list[str], check_code: bool
) -> tuple[int, bytes, bytes]:
if pp.exitcode and check_code:
msg = f"process {cmd} exit with code {pp.exitcode}"
msg += f"\n>>> stdout <<<\n{pp.out.decode()}"
msg += "\n"
msg += f"\n>>> stderr <<<\n{pp.err.decode()}"
raise RuntimeError(msg)
return cast("int", pp.exitcode), pp.out, pp.err
class TestProcessProtocol(ProcessProtocol):
def __init__(self) -> None:
self.deferred: Deferred[TestProcessProtocol] = Deferred()
self.out: bytes = b""
self.err: bytes = b""
self.exitcode: int | None = None
def outReceived(self, data: bytes) -> None:
self.out += data
def errReceived(self, data: bytes) -> None:
self.err += data
def processEnded(self, status: Failure) -> None:
self.exitcode = cast("ProcessTerminated", status.value).exitCode
self.deferred.callback(self)

View File

@ -1,65 +0,0 @@
# pragma: no file cover
import warnings
from urllib.parse import urljoin
from twisted.web import resource, server, static, util
from scrapy.exceptions import ScrapyDeprecationWarning
warnings.warn(
"The scrapy.utils.testsite module is deprecated.",
ScrapyDeprecationWarning,
stacklevel=2,
)
class SiteTest:
def setUp(self):
from twisted.internet import reactor
super().setUp()
self.site = reactor.listenTCP(0, test_site(), interface="127.0.0.1")
self.baseurl = f"http://localhost:{self.site.getHost().port}/"
def tearDown(self):
super().tearDown()
self.site.stopListening()
def url(self, path: str) -> str:
return urljoin(self.baseurl, path)
class NoMetaRefreshRedirect(util.Redirect):
def render(self, request: server.Request) -> bytes:
content = util.Redirect.render(self, request)
return content.replace(
b'http-equiv="refresh"', b'http-no-equiv="do-not-refresh-me"'
)
def test_site():
r = resource.Resource()
r.putChild(b"text", static.Data(b"Works", "text/plain"))
r.putChild(
b"html",
static.Data(
b"<body><p class='one'>Works</p><p class='two'>World</p></body>",
"text/html",
),
)
r.putChild(
b"enc-gb18030",
static.Data(b"<p>gb18030 encoding</p>", "text/html; charset=gb18030"),
)
r.putChild(b"redirect", util.Redirect(b"/redirected"))
r.putChild(b"redirect-no-meta-refresh", NoMetaRefreshRedirect(b"/redirected"))
r.putChild(b"redirected", static.Data(b"Redirected here", "text/plain"))
return server.Site(r)
if __name__ == "__main__":
from twisted.internet import reactor # pylint: disable=ungrouped-imports
port = reactor.listenTCP(0, test_site(), interface="127.0.0.1")
print(f"http://localhost:{port.getHost().port}/")
reactor.run()

View File

@ -6,36 +6,10 @@ library.
from __future__ import annotations
import re
import warnings
from importlib import import_module
from typing import TYPE_CHECKING, Any, TypeAlias
from urllib.parse import ParseResult, urldefrag, urlparse, urlunparse
from warnings import warn
from w3lib.url import __all__ as _public_w3lib_objects
from w3lib.url import add_or_replace_parameter as _add_or_replace_parameter
from w3lib.url import any_to_uri as _any_to_uri
from w3lib.url import parse_url as _parse_url
from scrapy.exceptions import ScrapyDeprecationWarning
_DEPRECATED_NAMES: frozenset[str] = frozenset(
{"_unquotepath", "_safe_chars", "parse_url", *_public_w3lib_objects}
)
def __getattr__(name: str) -> Any:
if name in _DEPRECATED_NAMES:
obj_type = "attribute" if name == "_safe_chars" else "function"
warnings.warn(
f"The scrapy.utils.url.{name} {obj_type} is deprecated, use w3lib.url.{name} instead.",
ScrapyDeprecationWarning,
stacklevel=2,
)
return getattr(import_module("w3lib.url"), name)
raise AttributeError
from typing import TYPE_CHECKING, TypeAlias
from urllib.parse import ParseResult, urlparse, urlunparse
from w3lib.url import any_to_uri, parse_url
if TYPE_CHECKING:
from collections.abc import Iterable
@ -47,7 +21,7 @@ UrlT: TypeAlias = str | bytes | ParseResult
def url_is_from_any_domain(url: UrlT, domains: Iterable[str]) -> bool:
"""Return True if the url belongs to any of the given domains"""
host = _parse_url(url).netloc.lower()
host = parse_url(url).netloc.lower()
if not host:
return False
return any((host == d) or (host.endswith(f".{d}")) for d in map(str.lower, domains))
@ -66,43 +40,10 @@ def url_is_from_spider(url: UrlT, spider: type[Spider]) -> bool:
def url_has_any_extension(url: UrlT, extensions: Iterable[str]) -> bool:
"""Return True if the url ends with one of the extensions provided"""
lowercase_path = _parse_url(url).path.lower()
lowercase_path = parse_url(url).path.lower()
return any(lowercase_path.endswith(ext) for ext in extensions)
def escape_ajax(url: str) -> str:
"""
Return the crawlable url
>>> escape_ajax("www.example.com/ajax.html#!key=value")
'www.example.com/ajax.html?_escaped_fragment_=key%3Dvalue'
>>> escape_ajax("www.example.com/ajax.html?k1=v1&k2=v2#!key=value")
'www.example.com/ajax.html?k1=v1&k2=v2&_escaped_fragment_=key%3Dvalue'
>>> escape_ajax("www.example.com/ajax.html?#!key=value")
'www.example.com/ajax.html?_escaped_fragment_=key%3Dvalue'
>>> escape_ajax("www.example.com/ajax.html#!")
'www.example.com/ajax.html?_escaped_fragment_='
URLs that are not "AJAX crawlable" (according to Google) returned as-is:
>>> escape_ajax("www.example.com/ajax.html#key=value")
'www.example.com/ajax.html#key=value'
>>> escape_ajax("www.example.com/ajax.html#")
'www.example.com/ajax.html#'
>>> escape_ajax("www.example.com/ajax.html")
'www.example.com/ajax.html'
"""
warn(
"escape_ajax() is deprecated and will be removed in a future Scrapy version.",
ScrapyDeprecationWarning,
stacklevel=2,
)
defrag, frag = urldefrag(url)
if not frag.startswith("!"):
return url
return _add_or_replace_parameter(defrag, "_escaped_fragment_", frag[1:])
def add_http_if_no_scheme(url: str) -> str:
"""Add http as the default scheme if it is missing from the url."""
match = re.match(r"^\w+://", url, flags=re.IGNORECASE)
@ -160,7 +101,7 @@ def guess_scheme(url: str) -> str:
"""Add an URL scheme if missing: file:// for filepath-like input or
http:// otherwise."""
if _is_filesystem_path(url):
return _any_to_uri(url)
return any_to_uri(url)
return add_http_if_no_scheme(url)

View File

@ -3,11 +3,9 @@ from __future__ import annotations
import platform
import sys
from importlib.metadata import version
from warnings import warn
import lxml.etree
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.settings.default_settings import LOG_VERSIONS
from scrapy.utils.ssl import get_openssl_version
@ -32,15 +30,3 @@ def get_versions(
) -> list[tuple[str, str]]:
software = software or _DEFAULT_SOFTWARE
return [(item, _version(item)) for item in software]
def scrapy_components_versions() -> list[tuple[str, str]]: # pragma: no cover
warn(
(
"scrapy.utils.versions.scrapy_components_versions() is deprecated, "
"use scrapy.utils.versions.get_versions() instead."
),
ScrapyDeprecationWarning,
stacklevel=2,
)
return get_versions()

View File

@ -1,62 +0,0 @@
import pytest
from scrapy.downloadermiddlewares.ajaxcrawl import AjaxCrawlMiddleware
from scrapy.http import HtmlResponse, Request, Response
from scrapy.spiders import Spider
from scrapy.utils.test import get_crawler
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
class TestAjaxCrawlMiddleware:
def setup_method(self):
crawler = get_crawler(Spider, {"AJAXCRAWL_ENABLED": True})
self.spider = crawler._create_spider("foo")
self.mw = AjaxCrawlMiddleware.from_crawler(crawler)
def _ajaxcrawlable_body(self):
return b'<html><head><meta name="fragment" content="!"/></head><body></body></html>'
def _req_resp(self, url, req_kwargs=None, resp_kwargs=None):
req = Request(url, **(req_kwargs or {}))
resp = HtmlResponse(url, request=req, **(resp_kwargs or {}))
return req, resp
def test_non_get(self):
req, resp = self._req_resp("http://example.com/", {"method": "HEAD"})
resp2 = self.mw.process_response(req, resp, self.spider)
assert resp == resp2
def test_binary_response(self):
req = Request("http://example.com/")
resp = Response("http://example.com/", body=b"foobar\x00\x01\x02", request=req)
resp2 = self.mw.process_response(req, resp, self.spider)
assert resp is resp2
def test_ajaxcrawl(self):
req, resp = self._req_resp(
"http://example.com/",
{"meta": {"foo": "bar"}},
{"body": self._ajaxcrawlable_body()},
)
req2 = self.mw.process_response(req, resp, self.spider)
assert req2.url == "http://example.com/?_escaped_fragment_="
assert req2.meta["foo"] == "bar"
def test_ajaxcrawl_loop(self):
req, resp = self._req_resp(
"http://example.com/", {}, {"body": self._ajaxcrawlable_body()}
)
req2 = self.mw.process_response(req, resp, self.spider)
resp2 = HtmlResponse(req2.url, body=resp.body, request=req2)
resp3 = self.mw.process_response(req2, resp2, self.spider)
assert isinstance(resp3, HtmlResponse), (resp3.__class__, resp3)
assert resp3.request.url == "http://example.com/?_escaped_fragment_="
assert resp3 is resp2
def test_noncrawlable_body(self):
req, resp = self._req_resp(
"http://example.com/", {}, {"body": b"<html></html>"}
)
resp2 = self.mw.process_response(req, resp, self.spider)
assert resp is resp2

View File

@ -11,10 +11,9 @@ from scrapy.crawler import Crawler
from scrapy.http import Response, TextResponse, XmlResponse
from scrapy.settings import Settings
from scrapy.spiders import CSVFeedSpider, Spider, XMLFeedSpider
from scrapy.spiders.init import InitSpider
from scrapy.utils.test import get_crawler, get_reactor_settings
from tests import get_testdata
from tests.utils.decorators import coroutine_test, inline_callbacks_test
from tests.utils.decorators import inline_callbacks_test
class TestSpider:
@ -122,27 +121,6 @@ class TestSpider:
mock_logger.log.assert_called_once_with("INFO", "test log msg")
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
class TestInitSpider(TestSpider):
spider_class = InitSpider
@coroutine_test
async def test_start_urls(self):
responses = []
class TestSpider(self.spider_class):
name = "test"
start_urls = ["data:,"]
async def parse(self, response):
responses.append(response)
crawler = get_crawler(TestSpider)
await crawler.crawl_async()
assert len(responses) == 1
assert responses[0].url == "data:,"
class TestXMLFeedSpider(TestSpider):
spider_class = XMLFeedSpider

View File

@ -1,13 +1,9 @@
import warnings
from importlib import import_module
import pytest
from scrapy.linkextractors import IGNORED_EXTENSIONS
from scrapy.spiders import Spider
from scrapy.utils.url import ( # type: ignore[attr-defined]
from scrapy.utils.url import (
_is_filesystem_path,
_public_w3lib_objects,
add_http_if_no_scheme,
guess_scheme,
strip_url,
@ -446,23 +442,3 @@ class TestStripUrl:
)
def test__is_filesystem_path(path: str, expected: bool) -> None:
assert _is_filesystem_path(path) == expected
@pytest.mark.parametrize(
"obj_name",
[
"_unquotepath",
"_safe_chars",
"parse_url",
*_public_w3lib_objects,
],
)
def test_deprecated_imports_from_w3lib(obj_name: str) -> None:
with warnings.catch_warnings(record=True) as warns:
obj_type = "attribute" if obj_name == "_safe_chars" else "function"
message = f"The scrapy.utils.url.{obj_name} {obj_type} is deprecated, use w3lib.url.{obj_name} instead."
getattr(import_module("scrapy.utils.url"), obj_name)
assert isinstance(warns[0].message, Warning)
assert message in warns[0].message.args