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/extensions, part 3. (#6325)

This commit is contained in:
Andrey Rakhmatullin 2024-04-29 12:43:45 +05:00 committed by GitHub
parent a166e97399
commit 57acad3c38
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 328 additions and 180 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

331
scrapy/extensions/feedexport.py
View File

@ -4,6 +4,8 @@ Feed Exports extension
See documentation in docs/topics/feed-exports.rst
"""
from __future__ import annotations
import logging
import re
import sys
@ -11,18 +13,36 @@ import warnings
from datetime import datetime, timezone
from pathlib import Path, PureWindowsPath
from tempfile import NamedTemporaryFile
from typing import IO, Any, Callable, Dict, List, Optional, Tuple, Type, Union
from typing import (
IO,
TYPE_CHECKING,
Any,
Callable,
Dict,
Iterable,
List,
Optional,
Protocol,
Tuple,
Type,
TypeVar,
Union,
cast,
)
from urllib.parse import unquote, urlparse
from twisted.internet import defer, threads
from twisted.internet.defer import DeferredList
from twisted.internet import threads
from twisted.internet.defer import Deferred, DeferredList, maybeDeferred
from twisted.python.failure import Failure
from w3lib.url import file_uri_to_path
from zope.interface import Interface, implementer
from scrapy import Spider, signals
from scrapy.crawler import Crawler
from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
from scrapy.exporters import BaseItemExporter
from scrapy.extensions.postprocessing import PostProcessingManager
from scrapy.settings import BaseSettings, Settings
from scrapy.utils.boto import is_botocore_available
from scrapy.utils.conf import feed_complete_default_values_from_settings
from scrapy.utils.defer import maybe_deferred_to_future
@ -32,6 +52,12 @@ from scrapy.utils.log import failure_to_exc_info
from scrapy.utils.misc import build_from_crawler, load_object
from scrapy.utils.python import without_none_values
if TYPE_CHECKING:
from _typeshed import OpenBinaryMode
# typing.Self requires Python 3.11
from typing_extensions import Self
logger = logging.getLogger(__name__)
try:
@ -41,8 +67,19 @@ try:
except ImportError:
IS_BOTO3_AVAILABLE = False
UriParamsCallableT = Callable[[Dict[str, Any], Spider], Optional[Dict[str, Any]]]
def build_storage(builder, uri, *args, feed_options=None, preargs=(), **kwargs):
_StorageT = TypeVar("_StorageT", bound="FeedStorageProtocol")
def build_storage(
builder: Callable[..., _StorageT],
uri: str,
*args: Any,
feed_options: Optional[Dict[str, Any]] = None,
preargs: Iterable[Any] = (),
**kwargs: Any,
) -> _StorageT:
kwargs["feed_options"] = feed_options
return builder(*preargs, uri, *args, **kwargs)
@ -56,10 +93,10 @@ class ItemFilter:
:type feed_options: dict
"""
feed_options: Optional[dict]
item_classes: Tuple
feed_options: Optional[Dict[str, Any]]
item_classes: Tuple[type, ...]
def __init__(self, feed_options: Optional[dict]) -> None:
def __init__(self, feed_options: Optional[Dict[str, Any]]) -> None:
self.feed_options = feed_options
if feed_options is not None:
self.item_classes = tuple(
@ -98,28 +135,49 @@ class IFeedStorage(Interface):
"""Store the given file stream"""
class FeedStorageProtocol(Protocol):
"""Reimplementation of ``IFeedStorage`` that can be used in type hints."""
def __init__(self, uri: str, *, feed_options: Optional[Dict[str, Any]] = None):
"""Initialize the storage with the parameters given in the URI and the
feed-specific options (see :setting:`FEEDS`)"""
def open(self, spider: Spider) -> IO[bytes]:
"""Open the storage for the given spider. It must return a file-like
object that will be used for the exporters"""
def store(self, file: IO[bytes]) -> Optional[Deferred]:
"""Store the given file stream"""
@implementer(IFeedStorage)
class BlockingFeedStorage:
def open(self, spider):
def open(self, spider: Spider) -> IO[bytes]:
path = spider.crawler.settings["FEED_TEMPDIR"]
if path and not Path(path).is_dir():
raise OSError("Not a Directory: " + str(path))
return NamedTemporaryFile(prefix="feed-", dir=path)
def store(self, file):
def store(self, file: IO[bytes]) -> Optional[Deferred]:
return threads.deferToThread(self._store_in_thread, file)
def _store_in_thread(self, file):
def _store_in_thread(self, file: IO[bytes]) -> None:
raise NotImplementedError
@implementer(IFeedStorage)
class StdoutFeedStorage:
def __init__(self, uri, _stdout=None, *, feed_options=None):
def __init__(
self,
uri: str,
_stdout: Optional[IO[bytes]] = None,
*,
feed_options: Optional[Dict[str, Any]] = None,
):
if not _stdout:
_stdout = sys.stdout.buffer
self._stdout = _stdout
self._stdout: IO[bytes] = _stdout
if feed_options and feed_options.get("overwrite", False) is True:
logger.warning(
"Standard output (stdout) storage does not support "
@ -128,54 +186,58 @@ class StdoutFeedStorage:
"it to False."
)
def open(self, spider):
def open(self, spider: Spider) -> IO[bytes]:
return self._stdout
def store(self, file):
def store(self, file: IO[bytes]) -> Optional[Deferred]:
pass
@implementer(IFeedStorage)
class FileFeedStorage:
def __init__(self, uri, *, feed_options=None):
self.path = file_uri_to_path(uri)
def __init__(self, uri: str, *, feed_options: Optional[Dict[str, Any]] = None):
self.path: str = file_uri_to_path(uri)
feed_options = feed_options or {}
self.write_mode = "wb" if feed_options.get("overwrite", False) else "ab"
self.write_mode: OpenBinaryMode = (
"wb" if feed_options.get("overwrite", False) else "ab"
)
def open(self, spider) -> IO[Any]:
def open(self, spider: Spider) -> IO[bytes]:
dirname = Path(self.path).parent
if dirname and not dirname.exists():
dirname.mkdir(parents=True)
return Path(self.path).open(self.write_mode)
def store(self, file):
def store(self, file: IO[bytes]) -> Optional[Deferred]:
file.close()
return None
class S3FeedStorage(BlockingFeedStorage):
def __init__(
self,
uri,
access_key=None,
secret_key=None,
acl=None,
endpoint_url=None,
uri: str,
access_key: Optional[str] = None,
secret_key: Optional[str] = None,
acl: Optional[str] = None,
endpoint_url: Optional[str] = None,
*,
feed_options=None,
session_token=None,
region_name=None,
feed_options: Optional[Dict[str, Any]] = None,
session_token: Optional[str] = None,
region_name: Optional[str] = None,
):
if not is_botocore_available():
raise NotConfigured("missing botocore library")
u = urlparse(uri)
self.bucketname = u.hostname
self.access_key = u.username or access_key
self.secret_key = u.password or secret_key
self.session_token = session_token
self.keyname = u.path[1:] # remove first "/"
self.acl = acl
self.endpoint_url = endpoint_url
self.region_name = region_name
assert u.hostname
self.bucketname: str = u.hostname
self.access_key: Optional[str] = u.username or access_key
self.secret_key: Optional[str] = u.password or secret_key
self.session_token: Optional[str] = session_token
self.keyname: str = u.path[1:] # remove first "/"
self.acl: Optional[str] = acl
self.endpoint_url: Optional[str] = endpoint_url
self.region_name: Optional[str] = region_name
if IS_BOTO3_AVAILABLE:
import boto3.session
@ -218,7 +280,13 @@ class S3FeedStorage(BlockingFeedStorage):
)
@classmethod
def from_crawler(cls, crawler, uri, *, feed_options=None):
def from_crawler(
cls,
crawler: Crawler,
uri: str,
*,
feed_options: Optional[Dict[str, Any]] = None,
) -> Self:
return build_storage(
cls,
uri,
@ -231,8 +299,9 @@ class S3FeedStorage(BlockingFeedStorage):
feed_options=feed_options,
)
def _store_in_thread(self, file):
def _store_in_thread(self, file: IO[bytes]) -> None:
file.seek(0)
kwargs: Dict[str, Any]
if IS_BOTO3_AVAILABLE:
kwargs = {"ExtraArgs": {"ACL": self.acl}} if self.acl else {}
self.s3_client.upload_fileobj(
@ -247,22 +316,23 @@ class S3FeedStorage(BlockingFeedStorage):
class GCSFeedStorage(BlockingFeedStorage):
def __init__(self, uri, project_id, acl):
self.project_id = project_id
self.acl = acl
def __init__(self, uri: str, project_id: Optional[str], acl: Optional[str]):
self.project_id: Optional[str] = project_id
self.acl: Optional[str] = acl
u = urlparse(uri)
self.bucket_name = u.hostname
self.blob_name = u.path[1:] # remove first "/"
assert u.hostname
self.bucket_name: str = u.hostname
self.blob_name: str = u.path[1:] # remove first "/"
@classmethod
def from_crawler(cls, crawler, uri):
def from_crawler(cls, crawler: Crawler, uri: str) -> Self:
return cls(
uri,
crawler.settings["GCS_PROJECT_ID"],
crawler.settings["FEED_STORAGE_GCS_ACL"] or None,
)
def _store_in_thread(self, file):
def _store_in_thread(self, file: IO[bytes]) -> None:
file.seek(0)
from google.cloud.storage import Client
@ -292,7 +362,13 @@ class FTPFeedStorage(BlockingFeedStorage):
self.overwrite: bool = not feed_options or feed_options.get("overwrite", True)
@classmethod
def from_crawler(cls, crawler, uri, *, feed_options=None):
def from_crawler(
cls,
crawler: Crawler,
uri: str,
*,
feed_options: Optional[Dict[str, Any]] = None,
) -> Self:
return build_storage(
cls,
uri,
@ -300,7 +376,7 @@ class FTPFeedStorage(BlockingFeedStorage):
feed_options=feed_options,
)
def _store_in_thread(self, file):
def _store_in_thread(self, file: IO[bytes]) -> None:
ftp_store_file(
path=self.path,
file=file,
@ -316,46 +392,51 @@ class FTPFeedStorage(BlockingFeedStorage):
class FeedSlot:
def __init__(
self,
storage,
uri,
format,
store_empty,
batch_id,
uri_template,
filter,
feed_options,
spider,
storage: FeedStorageProtocol,
uri: str,
format: str,
store_empty: bool,
batch_id: int,
uri_template: str,
filter: ItemFilter,
feed_options: Dict[str, Any],
spider: Spider,
exporters: Dict[str, Type[BaseItemExporter]],
settings,
crawler,
settings: BaseSettings,
crawler: Crawler,
):
self.file = None
self.file: Optional[IO[bytes]] = None
self.exporter: Optional[BaseItemExporter] = None
self.storage = storage
self.storage: FeedStorageProtocol = storage
# feed params
self.batch_id = batch_id
self.format = format
self.store_empty = store_empty
self.uri_template = uri_template
self.uri = uri
self.filter = filter
self.batch_id: int = batch_id
self.format: str = format
self.store_empty: bool = store_empty
self.uri_template: str = uri_template
self.uri: str = uri
self.filter: ItemFilter = filter
# exporter params
self.feed_options = feed_options
self.spider = spider
self.feed_options: Dict[str, Any] = feed_options
self.spider: Spider = spider
self.exporters: Dict[str, Type[BaseItemExporter]] = exporters
self.settings = settings
self.crawler = crawler
self.settings: BaseSettings = settings
self.crawler: Crawler = crawler
# flags
self.itemcount = 0
self._exporting = False
self._fileloaded = False
self.itemcount: int = 0
self._exporting: bool = False
self._fileloaded: bool = False
def start_exporting(self):
def start_exporting(self) -> None:
if not self._fileloaded:
self.file = self.storage.open(self.spider)
if "postprocessing" in self.feed_options:
self.file = PostProcessingManager(
self.feed_options["postprocessing"], self.file, self.feed_options
self.file = cast(
IO[bytes],
PostProcessingManager(
self.feed_options["postprocessing"],
self.file,
self.feed_options,
),
)
self.exporter = self._get_exporter(
file=self.file,
@ -368,17 +449,23 @@ class FeedSlot:
self._fileloaded = True
if not self._exporting:
assert self.exporter
self.exporter.start_exporting()
self._exporting = True
def _get_instance(self, objcls, *args, **kwargs):
def _get_instance(
self, objcls: Type[BaseItemExporter], *args: Any, **kwargs: Any
) -> BaseItemExporter:
return build_from_crawler(objcls, self.crawler, *args, **kwargs)
def _get_exporter(self, file, format, *args, **kwargs) -> BaseItemExporter:
def _get_exporter(
self, file: IO[bytes], format: str, *args: Any, **kwargs: Any
) -> BaseItemExporter:
return self._get_instance(self.exporters[format], file, *args, **kwargs)
def finish_exporting(self):
def finish_exporting(self) -> None:
if self._exporting:
assert self.exporter
self.exporter.finish_exporting()
self._exporting = False
@ -390,22 +477,22 @@ _FeedSlot = create_deprecated_class(
class FeedExporter:
_pending_deferreds: List[defer.Deferred] = []
_pending_deferreds: List[Deferred] = []
@classmethod
def from_crawler(cls, crawler):
def from_crawler(cls, crawler: Crawler) -> Self:
exporter = cls(crawler)
crawler.signals.connect(exporter.open_spider, signals.spider_opened)
crawler.signals.connect(exporter.close_spider, signals.spider_closed)
crawler.signals.connect(exporter.item_scraped, signals.item_scraped)
return exporter
def __init__(self, crawler):
self.crawler = crawler
self.settings = crawler.settings
def __init__(self, crawler: Crawler):
self.crawler: Crawler = crawler
self.settings: Settings = crawler.settings
self.feeds = {}
self.slots = []
self.filters = {}
self.slots: List[FeedSlot] = []
self.filters: Dict[str, ItemFilter] = {}
if not self.settings["FEEDS"] and not self.settings["FEED_URI"]:
raise NotConfigured
@ -437,8 +524,12 @@ class FeedExporter:
)
self.filters[uri] = self._load_filter(feed_options)
self.storages = self._load_components("FEED_STORAGES")
self.exporters = self._load_components("FEED_EXPORTERS")
self.storages: Dict[str, Type[FeedStorageProtocol]] = self._load_components(
"FEED_STORAGES"
)
self.exporters: Dict[str, Type[BaseItemExporter]] = self._load_components(
"FEED_EXPORTERS"
)
for uri, feed_options in self.feeds.items():
if not self._storage_supported(uri, feed_options):
raise NotConfigured
@ -447,7 +538,7 @@ class FeedExporter:
if not self._exporter_supported(feed_options["format"]):
raise NotConfigured
def open_spider(self, spider):
def open_spider(self, spider: Spider) -> None:
for uri, feed_options in self.feeds.items():
uri_params = self._get_uri_params(spider, feed_options["uri_params"])
self.slots.append(
@ -460,7 +551,7 @@ class FeedExporter:
)
)
async def close_spider(self, spider):
async def close_spider(self, spider: Spider) -> None:
for slot in self.slots:
self._close_slot(slot, spider)
@ -473,8 +564,9 @@ class FeedExporter:
self.crawler.signals.send_catch_log_deferred(signals.feed_exporter_closed)
)
def _close_slot(self, slot, spider):
def get_file(slot_):
def _close_slot(self, slot: FeedSlot, spider: Spider) -> Optional[Deferred]:
def get_file(slot_: FeedSlot) -> IO[bytes]:
assert slot_.file
if isinstance(slot_.file, PostProcessingManager):
slot_.file.close()
return slot_.file.file
@ -492,7 +584,7 @@ class FeedExporter:
return None
logmsg = f"{slot.format} feed ({slot.itemcount} items) in: {slot.uri}"
d = defer.maybeDeferred(slot.storage.store, get_file(slot))
d: Deferred = maybeDeferred(slot.storage.store, get_file(slot))
d.addCallback(
self._handle_store_success, logmsg, spider, type(slot.storage).__name__
@ -510,20 +602,33 @@ class FeedExporter:
return d
def _handle_store_error(self, f, logmsg, spider, slot_type):
def _handle_store_error(
self, f: Failure, logmsg: str, spider: Spider, slot_type: str
) -> None:
logger.error(
"Error storing %s",
logmsg,
exc_info=failure_to_exc_info(f),
extra={"spider": spider},
)
assert self.crawler.stats
self.crawler.stats.inc_value(f"feedexport/failed_count/{slot_type}")
def _handle_store_success(self, f, logmsg, spider, slot_type):
def _handle_store_success(
self, f: Failure, logmsg: str, spider: Spider, slot_type: str
) -> None:
logger.info("Stored %s", logmsg, extra={"spider": spider})
assert self.crawler.stats
self.crawler.stats.inc_value(f"feedexport/success_count/{slot_type}")
def _start_new_batch(self, batch_id, uri, feed_options, spider, uri_template):
def _start_new_batch(
self,
batch_id: int,
uri: str,
feed_options: Dict[str, Any],
spider: Spider,
uri_template: str,
) -> FeedSlot:
"""
Redirect the output data stream to a new file.
Execute multiple times if FEED_EXPORT_BATCH_ITEM_COUNT setting or FEEDS.batch_item_count is specified
@ -546,11 +651,11 @@ class FeedExporter:
spider=spider,
exporters=self.exporters,
settings=self.settings,
crawler=getattr(self, "crawler", None),
crawler=self.crawler,
)
return slot
def item_scraped(self, item, spider):
def item_scraped(self, item: Any, spider: Spider) -> None:
slots = []
for slot in self.slots:
if not slot.filter.accepts(item):
@ -560,6 +665,7 @@ class FeedExporter:
continue
slot.start_exporting()
assert slot.exporter
slot.exporter.export_item(item)
slot.itemcount += 1
# create new slot for each slot with itemcount == FEED_EXPORT_BATCH_ITEM_COUNT and close the old one
@ -584,7 +690,7 @@ class FeedExporter:
slots.append(slot)
self.slots = slots
def _load_components(self, setting_prefix):
def _load_components(self, setting_prefix: str) -> Dict[str, Any]:
conf = without_none_values(self.settings.getwithbase(setting_prefix))
d = {}
for k, v in conf.items():
@ -594,12 +700,13 @@ class FeedExporter:
pass
return d
def _exporter_supported(self, format):
def _exporter_supported(self, format: str) -> bool:
if format in self.exporters:
return True
logger.error("Unknown feed format: %(format)s", {"format": format})
return False
def _settings_are_valid(self):
def _settings_are_valid(self) -> bool:
"""
If FEED_EXPORT_BATCH_ITEM_COUNT setting or FEEDS.batch_item_count is specified uri has to contain
%(batch_time)s or %(batch_id)d to distinguish different files of partial output
@ -617,7 +724,7 @@ class FeedExporter:
return False
return True
def _storage_supported(self, uri, feed_options):
def _storage_supported(self, uri: str, feed_options: Dict[str, Any]) -> bool:
scheme = urlparse(uri).scheme
if scheme in self.storages or PureWindowsPath(uri).drive:
try:
@ -630,8 +737,11 @@ class FeedExporter:
)
else:
logger.error("Unknown feed storage scheme: %(scheme)s", {"scheme": scheme})
return False
def _get_storage(self, uri, feed_options):
def _get_storage(
self, uri: str, feed_options: Dict[str, Any]
) -> FeedStorageProtocol:
"""Fork of create_instance specific to feed storage classes
It supports not passing the *feed_options* parameters to classes that
@ -640,11 +750,14 @@ class FeedExporter:
feedcls = self.storages.get(urlparse(uri).scheme, self.storages["file"])
crawler = getattr(self, "crawler", None)
def build_instance(builder, *preargs):
def build_instance(
builder: Type[FeedStorageProtocol], *preargs: Any
) -> FeedStorageProtocol:
return build_storage(
builder, uri, feed_options=feed_options, preargs=preargs
)
instance: FeedStorageProtocol
if crawler and hasattr(feedcls, "from_crawler"):
instance = build_instance(feedcls.from_crawler, crawler)
method_name = "from_crawler"
@ -661,9 +774,9 @@ class FeedExporter:
def _get_uri_params(
self,
spider: Spider,
uri_params_function: Optional[Union[str, Callable[[dict, Spider], dict]]],
uri_params_function: Union[str, UriParamsCallableT, None],
slot: Optional[FeedSlot] = None,
) -> dict:
) -> Dict[str, Any]:
params = {}
for k in dir(spider):
params[k] = getattr(spider, k)
@ -671,7 +784,7 @@ class FeedExporter:
params["time"] = utc_now.replace(microsecond=0).isoformat().replace(":", "-")
params["batch_time"] = utc_now.isoformat().replace(":", "-")
params["batch_id"] = slot.batch_id + 1 if slot is not None else 1
uripar_function = (
uripar_function: UriParamsCallableT = (
load_object(uri_params_function)
if uri_params_function
else lambda params, _: params
@ -679,7 +792,9 @@ class FeedExporter:
new_params = uripar_function(params, spider)
return new_params if new_params is not None else params
def _load_filter(self, feed_options):
def _load_filter(self, feed_options: Dict[str, Any]) -> ItemFilter:
# load the item filter if declared else load the default filter class
item_filter_class = load_object(feed_options.get("item_filter", ItemFilter))
item_filter_class: Type[ItemFilter] = load_object(
feed_options.get("item_filter", ItemFilter)
)
return item_filter_class(feed_options)

153
scrapy/extensions/httpcache.py
View File

@ -1,10 +1,13 @@
import gzip
import logging
import os
import pickle # nosec
from email.utils import mktime_tz, parsedate_tz
from importlib import import_module
from pathlib import Path
from time import time
from types import ModuleType
from typing import IO, TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union, cast
from weakref import WeakKeyDictionary
from w3lib.http import headers_dict_to_raw, headers_raw_to_dict
@ -12,49 +15,65 @@ from w3lib.http import headers_dict_to_raw, headers_raw_to_dict
from scrapy.http import Headers, Response
from scrapy.http.request import Request
from scrapy.responsetypes import responsetypes
from scrapy.settings import BaseSettings
from scrapy.spiders import Spider
from scrapy.utils.httpobj import urlparse_cached
from scrapy.utils.project import data_path
from scrapy.utils.python import to_bytes, to_unicode
from scrapy.utils.request import RequestFingerprinter
if TYPE_CHECKING:
# typing.Concatenate requires Python 3.10
from typing_extensions import Concatenate
logger = logging.getLogger(__name__)
class DummyPolicy:
def __init__(self, settings):
self.ignore_schemes = settings.getlist("HTTPCACHE_IGNORE_SCHEMES")
self.ignore_http_codes = [
def __init__(self, settings: BaseSettings):
self.ignore_schemes: List[str] = settings.getlist("HTTPCACHE_IGNORE_SCHEMES")
self.ignore_http_codes: List[int] = [
int(x) for x in settings.getlist("HTTPCACHE_IGNORE_HTTP_CODES")
]
def should_cache_request(self, request):
def should_cache_request(self, request: Request) -> bool:
return urlparse_cached(request).scheme not in self.ignore_schemes
def should_cache_response(self, response, request):
def should_cache_response(self, response: Response, request: Request) -> bool:
return response.status not in self.ignore_http_codes
def is_cached_response_fresh(self, cachedresponse, request):
def is_cached_response_fresh(
self, cachedresponse: Response, request: Request
) -> bool:
return True
def is_cached_response_valid(self, cachedresponse, response, request):
def is_cached_response_valid(
self, cachedresponse: Response, response: Response, request: Request
) -> bool:
return True
class RFC2616Policy:
MAXAGE = 3600 * 24 * 365 # one year
def __init__(self, settings):
self.always_store = settings.getbool("HTTPCACHE_ALWAYS_STORE")
self.ignore_schemes = settings.getlist("HTTPCACHE_IGNORE_SCHEMES")
self._cc_parsed = WeakKeyDictionary()
self.ignore_response_cache_controls = [
def __init__(self, settings: BaseSettings):
self.always_store: bool = settings.getbool("HTTPCACHE_ALWAYS_STORE")
self.ignore_schemes: List[str] = settings.getlist("HTTPCACHE_IGNORE_SCHEMES")
self._cc_parsed: WeakKeyDictionary[
Union[Request, Response], Dict[bytes, Optional[bytes]]
] = WeakKeyDictionary()
self.ignore_response_cache_controls: List[bytes] = [
to_bytes(cc)
for cc in settings.getlist("HTTPCACHE_IGNORE_RESPONSE_CACHE_CONTROLS")
]
def _parse_cachecontrol(self, r):
def _parse_cachecontrol(
self, r: Union[Request, Response]
) -> Dict[bytes, Optional[bytes]]:
if r not in self._cc_parsed:
cch = r.headers.get(b"Cache-Control", b"")
assert cch is not None
parsed = parse_cachecontrol(cch)
if isinstance(r, Response):
for key in self.ignore_response_cache_controls:
@ -62,7 +81,7 @@ class RFC2616Policy:
self._cc_parsed[r] = parsed
return self._cc_parsed[r]
def should_cache_request(self, request):
def should_cache_request(self, request: Request) -> bool:
if urlparse_cached(request).scheme in self.ignore_schemes:
return False
cc = self._parse_cachecontrol(request)
@ -72,7 +91,7 @@ class RFC2616Policy:
# Any other is eligible for caching
return True
def should_cache_response(self, response, request):
def should_cache_response(self, response: Response, request: Request) -> bool:
# What is cacheable - https://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.9.1
# Response cacheability - https://www.w3.org/Protocols/rfc2616/rfc2616-sec13.html#sec13.4
# Status code 206 is not included because cache can not deal with partial contents
@ -100,7 +119,9 @@ class RFC2616Policy:
# info and can not be revalidated
return False
def is_cached_response_fresh(self, cachedresponse, request):
def is_cached_response_fresh(
self, cachedresponse: Response, request: Request
) -> bool:
cc = self._parse_cachecontrol(cachedresponse)
ccreq = self._parse_cachecontrol(request)
if b"no-cache" in cc or b"no-cache" in ccreq:
@ -141,7 +162,9 @@ class RFC2616Policy:
self._set_conditional_validators(request, cachedresponse)
return False
def is_cached_response_valid(self, cachedresponse, response, request):
def is_cached_response_valid(
self, cachedresponse: Response, response: Response, request: Request
) -> bool:
# Use the cached response if the new response is a server error,
# as long as the old response didn't specify must-revalidate.
if response.status >= 500:
@ -152,7 +175,9 @@ class RFC2616Policy:
# Use the cached response if the server says it hasn't changed.
return response.status == 304
def _set_conditional_validators(self, request, cachedresponse):
def _set_conditional_validators(
self, request: Request, cachedresponse: Response
) -> None:
if b"Last-Modified" in cachedresponse.headers:
request.headers[b"If-Modified-Since"] = cachedresponse.headers[
b"Last-Modified"
@ -161,13 +186,15 @@ class RFC2616Policy:
if b"ETag" in cachedresponse.headers:
request.headers[b"If-None-Match"] = cachedresponse.headers[b"ETag"]
def _get_max_age(self, cc):
def _get_max_age(self, cc: Dict[bytes, Optional[bytes]]) -> Optional[int]:
try:
return max(0, int(cc[b"max-age"]))
return max(0, int(cc[b"max-age"])) # type: ignore[arg-type]
except (KeyError, ValueError):
return None
def _compute_freshness_lifetime(self, response, request, now):
def _compute_freshness_lifetime(
self, response: Response, request: Request, now: float
) -> float:
# Reference nsHttpResponseHead::ComputeFreshnessLifetime
# https://dxr.mozilla.org/mozilla-central/source/netwerk/protocol/http/nsHttpResponseHead.cpp#706
cc = self._parse_cachecontrol(response)
@ -198,10 +225,12 @@ class RFC2616Policy:
# Insufficient information to compute freshness lifetime
return 0
def _compute_current_age(self, response, request, now):
def _compute_current_age(
self, response: Response, request: Request, now: float
) -> float:
# Reference nsHttpResponseHead::ComputeCurrentAge
# https://dxr.mozilla.org/mozilla-central/source/netwerk/protocol/http/nsHttpResponseHead.cpp#658
currentage = 0
currentage: float = 0
# If Date header is not set we assume it is a fast connection, and
# clock is in sync with the server
date = rfc1123_to_epoch(response.headers.get(b"Date")) or now
@ -210,7 +239,7 @@ class RFC2616Policy:
if b"Age" in response.headers:
try:
age = int(response.headers[b"Age"])
age = int(response.headers[b"Age"]) # type: ignore[arg-type]
currentage = max(currentage, age)
except ValueError:
pass
@ -219,13 +248,13 @@ class RFC2616Policy:
class DbmCacheStorage:
def __init__(self, settings):
self.cachedir = data_path(settings["HTTPCACHE_DIR"], createdir=True)
self.expiration_secs = settings.getint("HTTPCACHE_EXPIRATION_SECS")
self.dbmodule = import_module(settings["HTTPCACHE_DBM_MODULE"])
self.db = None
def __init__(self, settings: BaseSettings):
self.cachedir: str = data_path(settings["HTTPCACHE_DIR"], createdir=True)
self.expiration_secs: int = settings.getint("HTTPCACHE_EXPIRATION_SECS")
self.dbmodule: ModuleType = import_module(settings["HTTPCACHE_DBM_MODULE"])
self.db: Any = None # the real type is private
def open_spider(self, spider: Spider):
def open_spider(self, spider: Spider) -> None:
dbpath = Path(self.cachedir, f"{spider.name}.db")
self.db = self.dbmodule.open(str(dbpath), "c")
@ -235,15 +264,16 @@ class DbmCacheStorage:
extra={"spider": spider},
)
self._fingerprinter = spider.crawler.request_fingerprinter
assert spider.crawler.request_fingerprinter
self._fingerprinter: RequestFingerprinter = spider.crawler.request_fingerprinter
def close_spider(self, spider):
def close_spider(self, spider: Spider) -> None:
self.db.close()
def retrieve_response(self, spider, request):
def retrieve_response(self, spider: Spider, request: Request) -> Optional[Response]:
data = self._read_data(spider, request)
if data is None:
return # not cached
return None # not cached
url = data["url"]
status = data["status"]
headers = Headers(data["headers"])
@ -252,7 +282,9 @@ class DbmCacheStorage:
response = respcls(url=url, headers=headers, status=status, body=body)
return response
def store_response(self, spider, request, response):
def store_response(
self, spider: Spider, request: Request, response: Response
) -> None:
key = self._fingerprinter.fingerprint(request).hex()
data = {
"status": response.status,
@ -263,28 +295,31 @@ class DbmCacheStorage:
self.db[f"{key}_data"] = pickle.dumps(data, protocol=4)
self.db[f"{key}_time"] = str(time())
def _read_data(self, spider, request):
def _read_data(self, spider: Spider, request: Request) -> Optional[Dict[str, Any]]:
key = self._fingerprinter.fingerprint(request).hex()
db = self.db
tkey = f"{key}_time"
if tkey not in db:
return # not found
return None # not found
ts = db[tkey]
if 0 < self.expiration_secs < time() - float(ts):
return # expired
return None # expired
return pickle.loads(db[f"{key}_data"]) # nosec
return cast(Dict[str, Any], pickle.loads(db[f"{key}_data"])) # nosec
class FilesystemCacheStorage:
def __init__(self, settings):
self.cachedir = data_path(settings["HTTPCACHE_DIR"])
self.expiration_secs = settings.getint("HTTPCACHE_EXPIRATION_SECS")
self.use_gzip = settings.getbool("HTTPCACHE_GZIP")
self._open = gzip.open if self.use_gzip else open
def __init__(self, settings: BaseSettings):
self.cachedir: str = data_path(settings["HTTPCACHE_DIR"])
self.expiration_secs: int = settings.getint("HTTPCACHE_EXPIRATION_SECS")
self.use_gzip: bool = settings.getbool("HTTPCACHE_GZIP")
# https://github.com/python/mypy/issues/10740
self._open: Callable[Concatenate[Union[str, os.PathLike], str, ...], IO] = (
gzip.open if self.use_gzip else open # type: ignore[assignment]
)
def open_spider(self, spider: Spider):
def open_spider(self, spider: Spider) -> None:
logger.debug(
"Using filesystem cache storage in %(cachedir)s",
{"cachedir": self.cachedir},
@ -294,27 +329,29 @@ class FilesystemCacheStorage:
assert spider.crawler.request_fingerprinter
self._fingerprinter = spider.crawler.request_fingerprinter
def close_spider(self, spider):
def close_spider(self, spider: Spider) -> None:
pass
def retrieve_response(self, spider: Spider, request: Request):
def retrieve_response(self, spider: Spider, request: Request) -> Optional[Response]:
"""Return response if present in cache, or None otherwise."""
metadata = self._read_meta(spider, request)
if metadata is None:
return # not cached
return None # not cached
rpath = Path(self._get_request_path(spider, request))
with self._open(rpath / "response_body", "rb") as f:
body = f.read()
with self._open(rpath / "response_headers", "rb") as f:
rawheaders = f.read()
url = metadata.get("response_url")
url = metadata["response_url"]
status = metadata["status"]
headers = Headers(headers_raw_to_dict(rawheaders))
respcls = responsetypes.from_args(headers=headers, url=url, body=body)
response = respcls(url=url, headers=headers, status=status, body=body)
return response
def store_response(self, spider: Spider, request: Request, response):
def store_response(
self, spider: Spider, request: Request, response: Response
) -> None:
"""Store the given response in the cache."""
rpath = Path(self._get_request_path(spider, request))
if not rpath.exists():
@ -343,19 +380,19 @@ class FilesystemCacheStorage:
key = self._fingerprinter.fingerprint(request).hex()
return str(Path(self.cachedir, spider.name, key[0:2], key))
def _read_meta(self, spider: Spider, request: Request):
def _read_meta(self, spider: Spider, request: Request) -> Optional[Dict[str, Any]]:
rpath = Path(self._get_request_path(spider, request))
metapath = rpath / "pickled_meta"
if not metapath.exists():
return # not found
return None # not found
mtime = metapath.stat().st_mtime
if 0 < self.expiration_secs < time() - mtime:
return # expired
return None # expired
with self._open(metapath, "rb") as f:
return pickle.load(f) # nosec
return cast(Dict[str, Any], pickle.load(f)) # nosec
def parse_cachecontrol(header):
def parse_cachecontrol(header: bytes) -> Dict[bytes, Optional[bytes]]:
"""Parse Cache-Control header
https://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.9
@ -375,9 +412,9 @@ def parse_cachecontrol(header):
return directives
def rfc1123_to_epoch(date_str):
def rfc1123_to_epoch(date_str: Union[str, bytes, None]) -> Optional[int]:
try:
date_str = to_unicode(date_str, encoding="ascii")
return mktime_tz(parsedate_tz(date_str))
date_str = to_unicode(date_str, encoding="ascii") # type: ignore[arg-type]
return mktime_tz(parsedate_tz(date_str)) # type: ignore[arg-type]
except Exception:
return None

4
scrapy/extensions/postprocessing.py
View File

@ -6,7 +6,7 @@ from bz2 import BZ2File
from gzip import GzipFile
from io import IOBase
from lzma import LZMAFile
from typing import Any, BinaryIO, Dict, List, cast
from typing import IO, Any, BinaryIO, Dict, List, cast
from scrapy.utils.misc import load_object
@ -126,7 +126,7 @@ class PostProcessingManager(IOBase):
"""
def __init__(
self, plugins: List[Any], file: BinaryIO, feed_options: Dict[str, Any]
self, plugins: List[Any], file: IO[bytes], feed_options: Dict[str, Any]
) -> None:
self.plugins = self._load_plugins(plugins)
self.file = file

6
scrapy/utils/iterators.py
View File

@ -3,7 +3,6 @@ import logging
import re
from io import StringIO
from typing import (
TYPE_CHECKING,
Any,
Callable,
Dict,
@ -25,9 +24,6 @@ from scrapy.http import Response, TextResponse
from scrapy.selector import Selector
from scrapy.utils.python import re_rsearch, to_unicode
if TYPE_CHECKING:
from lxml._types import SupportsReadClose # nosec
logger = logging.getLogger(__name__)
@ -98,7 +94,7 @@ def xmliter_lxml(
reader = _StreamReader(obj)
tag = f"{{{namespace}}}{nodename}" if namespace else nodename
iterable = etree.iterparse(
cast("SupportsReadClose[bytes]", reader),
reader,
encoding=reader.encoding,
events=("end", "start-ns"),
resolve_entities=False,

14
tox.ini
View File

@ -43,14 +43,14 @@ install_command =
[testenv:typing]
basepython = python3
deps =
mypy==1.8.0
typing-extensions==4.10.0
mypy==1.10.0
typing-extensions==4.11.0
types-attrs==19.1.0
types-lxml==2024.2.9
types-Pillow==10.2.0.20240213
types-Pygments==2.17.0.20240106
types-pyOpenSSL==24.0.0.20240130
types-setuptools==69.1.0.20240223
types-lxml==2024.4.14
types-Pillow==10.2.0.20240423
types-Pygments==2.17.0.20240310
types-pyOpenSSL==24.0.0.20240417
types-setuptools==69.5.0.20240423
# 2.1.2 fixes a typing bug: https://github.com/scrapy/w3lib/pull/211
w3lib >= 2.1.2
commands =