mirror of https://github.com/scrapy/scrapy.git
Remove deprecated scrapy.downloadermiddlewares.decompression
This commit is contained in:
Laerte Pereira
parent
991121fa91
commit
7a5cefbcfa
|
|
@ -1,94 +0,0 @@
|
|||
""" This module implements the DecompressionMiddleware which tries to recognise
|
||||
and extract the potentially compressed responses that may arrive.
|
||||
"""
|
||||
|
||||
import bz2
|
||||
import gzip
|
||||
import logging
|
||||
import tarfile
|
||||
import zipfile
|
||||
from io import BytesIO
|
||||
from tempfile import mktemp
|
||||
from warnings import warn
|
||||
|
||||
from scrapy.exceptions import ScrapyDeprecationWarning
|
||||
from scrapy.responsetypes import responsetypes
|
||||
|
||||
warn(
|
||||
"scrapy.downloadermiddlewares.decompression is deprecated",
|
||||
ScrapyDeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DecompressionMiddleware:
|
||||
"""This middleware tries to recognise and extract the possibly compressed
|
||||
responses that may arrive."""
|
||||
|
||||
def __init__(self):
|
||||
self._formats = {
|
||||
"tar": self._is_tar,
|
||||
"zip": self._is_zip,
|
||||
"gz": self._is_gzip,
|
||||
"bz2": self._is_bzip2,
|
||||
}
|
||||
|
||||
def _is_tar(self, response):
|
||||
archive = BytesIO(response.body)
|
||||
try:
|
||||
tar_file = tarfile.open(name=mktemp(), fileobj=archive)
|
||||
except tarfile.ReadError:
|
||||
return
|
||||
|
||||
body = tar_file.extractfile(tar_file.members[0]).read()
|
||||
respcls = responsetypes.from_args(filename=tar_file.members[0].name, body=body)
|
||||
return response.replace(body=body, cls=respcls)
|
||||
|
||||
def _is_zip(self, response):
|
||||
archive = BytesIO(response.body)
|
||||
try:
|
||||
zip_file = zipfile.ZipFile(archive)
|
||||
except zipfile.BadZipFile:
|
||||
return
|
||||
|
||||
namelist = zip_file.namelist()
|
||||
body = zip_file.read(namelist[0])
|
||||
respcls = responsetypes.from_args(filename=namelist[0], body=body)
|
||||
return response.replace(body=body, cls=respcls)
|
||||
|
||||
def _is_gzip(self, response):
|
||||
archive = BytesIO(response.body)
|
||||
try:
|
||||
body = gzip.GzipFile(fileobj=archive).read()
|
||||
except OSError:
|
||||
return
|
||||
|
||||
respcls = responsetypes.from_args(body=body)
|
||||
return response.replace(body=body, cls=respcls)
|
||||
|
||||
def _is_bzip2(self, response):
|
||||
try:
|
||||
body = bz2.decompress(response.body)
|
||||
except OSError:
|
||||
return
|
||||
|
||||
respcls = responsetypes.from_args(body=body)
|
||||
return response.replace(body=body, cls=respcls)
|
||||
|
||||
def process_response(self, request, response, spider):
|
||||
if not response.body:
|
||||
return response
|
||||
|
||||
for fmt, func in self._formats.items():
|
||||
new_response = func(response)
|
||||
if new_response:
|
||||
logger.debug(
|
||||
"Decompressed response with format: %(responsefmt)s",
|
||||
{"responsefmt": fmt},
|
||||
extra={"spider": spider},
|
||||
)
|
||||
return new_response
|
||||
return response
|
||||
|
|
@ -1,53 +0,0 @@
|
|||
from unittest import TestCase, main
|
||||
|
||||
from scrapy.downloadermiddlewares.decompression import DecompressionMiddleware
|
||||
from scrapy.http import Response, XmlResponse
|
||||
from scrapy.spiders import Spider
|
||||
from scrapy.utils.test import assert_samelines
|
||||
from tests import get_testdata
|
||||
|
||||
|
||||
def _test_data(formats):
|
||||
uncompressed_body = get_testdata("compressed", "feed-sample1.xml")
|
||||
test_responses = {}
|
||||
for format in formats:
|
||||
body = get_testdata("compressed", "feed-sample1." + format)
|
||||
test_responses[format] = Response("http://foo.com/bar", body=body)
|
||||
return uncompressed_body, test_responses
|
||||
|
||||
|
||||
class DecompressionMiddlewareTest(TestCase):
|
||||
test_formats = ["tar", "xml.bz2", "xml.gz", "zip"]
|
||||
uncompressed_body, test_responses = _test_data(test_formats)
|
||||
|
||||
def setUp(self):
|
||||
self.mw = DecompressionMiddleware()
|
||||
self.spider = Spider("foo")
|
||||
|
||||
def test_known_compression_formats(self):
|
||||
for fmt in self.test_formats:
|
||||
rsp = self.test_responses[fmt]
|
||||
new = self.mw.process_response(None, rsp, self.spider)
|
||||
error_msg = f"Failed {fmt}, response type {type(new).__name__}"
|
||||
assert isinstance(new, XmlResponse), error_msg
|
||||
assert_samelines(self, new.body, self.uncompressed_body, fmt)
|
||||
|
||||
def test_plain_response(self):
|
||||
rsp = Response(url="http://test.com", body=self.uncompressed_body)
|
||||
new = self.mw.process_response(None, rsp, self.spider)
|
||||
assert new is rsp
|
||||
assert_samelines(self, new.body, rsp.body)
|
||||
|
||||
def test_empty_response(self):
|
||||
rsp = Response(url="http://test.com", body=b"")
|
||||
new = self.mw.process_response(None, rsp, self.spider)
|
||||
assert new is rsp
|
||||
assert not rsp.body
|
||||
assert not new.body
|
||||
|
||||
def tearDown(self):
|
||||
del self.mw
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Reference in New Issue