mirror of https://github.com/scrapy/scrapy.git
Typing for scrapy/settings/__init__.py.
This commit is contained in:
parent
93962ebefc
commit
187e8f9a2d
|
|
@ -55,7 +55,7 @@ class Crawler:
|
|||
def __init__(
|
||||
self,
|
||||
spidercls: Type[Spider],
|
||||
settings: Union[None, dict, Settings] = None,
|
||||
settings: Union[None, Dict[str, Any], Settings] = None,
|
||||
init_reactor: bool = False,
|
||||
):
|
||||
if isinstance(spidercls, Spider):
|
||||
|
|
|
|||
|
|
@ -1,12 +1,41 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
from collections.abc import MutableMapping
|
||||
from importlib import import_module
|
||||
from pprint import pformat
|
||||
from types import ModuleType
|
||||
from typing import (
|
||||
TYPE_CHECKING,
|
||||
Any,
|
||||
Dict,
|
||||
Iterable,
|
||||
Iterator,
|
||||
List,
|
||||
Mapping,
|
||||
MutableMapping,
|
||||
Optional,
|
||||
Tuple,
|
||||
Union,
|
||||
cast,
|
||||
)
|
||||
|
||||
from scrapy.settings import default_settings
|
||||
|
||||
SETTINGS_PRIORITIES = {
|
||||
# The key types are restricted in BaseSettings._get_key() to ones supported by JSON,
|
||||
# see https://github.com/scrapy/scrapy/issues/5383.
|
||||
_SettingsKeyT = Union[bool, float, int, str, None]
|
||||
|
||||
if TYPE_CHECKING:
|
||||
# typing.Self requires Python 3.11
|
||||
# https://github.com/python/typing/issues/445#issuecomment-1131458824
|
||||
from _typeshed import SupportsItems
|
||||
from typing_extensions import Self
|
||||
|
||||
_SettingsInputT = Union[SupportsItems[_SettingsKeyT, Any], str, None]
|
||||
|
||||
|
||||
SETTINGS_PRIORITIES: Dict[str, int] = {
|
||||
"default": 0,
|
||||
"command": 10,
|
||||
"project": 20,
|
||||
|
|
@ -15,7 +44,7 @@ SETTINGS_PRIORITIES = {
|
|||
}
|
||||
|
||||
|
||||
def get_settings_priority(priority):
|
||||
def get_settings_priority(priority: Union[int, str]) -> int:
|
||||
"""
|
||||
Small helper function that looks up a given string priority in the
|
||||
:attr:`~scrapy.settings.SETTINGS_PRIORITIES` dictionary and returns its
|
||||
|
|
@ -34,14 +63,15 @@ class SettingsAttribute:
|
|||
for settings configuration, not this one.
|
||||
"""
|
||||
|
||||
def __init__(self, value, priority):
|
||||
self.value = value
|
||||
def __init__(self, value: Any, priority: int):
|
||||
self.value: Any = value
|
||||
self.priority: int
|
||||
if isinstance(self.value, BaseSettings):
|
||||
self.priority = max(self.value.maxpriority(), priority)
|
||||
else:
|
||||
self.priority = priority
|
||||
|
||||
def set(self, value, priority):
|
||||
def set(self, value: Any, priority: int) -> None:
|
||||
"""Sets value if priority is higher or equal than current priority."""
|
||||
if priority >= self.priority:
|
||||
if isinstance(self.value, BaseSettings):
|
||||
|
|
@ -49,11 +79,11 @@ class SettingsAttribute:
|
|||
self.value = value
|
||||
self.priority = priority
|
||||
|
||||
def __repr__(self):
|
||||
def __repr__(self) -> str:
|
||||
return f"<SettingsAttribute value={self.value!r} priority={self.priority}>"
|
||||
|
||||
|
||||
class BaseSettings(MutableMapping):
|
||||
class BaseSettings(MutableMapping[_SettingsKeyT, Any]):
|
||||
"""
|
||||
Instances of this class behave like dictionaries, but store priorities
|
||||
along with their ``(key, value)`` pairs, and can be frozen (i.e. marked
|
||||
|
|
@ -77,21 +107,23 @@ class BaseSettings(MutableMapping):
|
|||
|
||||
__default = object()
|
||||
|
||||
def __init__(self, values=None, priority="project"):
|
||||
self.frozen = False
|
||||
self.attributes = {}
|
||||
def __init__(
|
||||
self, values: _SettingsInputT = None, priority: Union[int, str] = "project"
|
||||
):
|
||||
self.frozen: bool = False
|
||||
self.attributes: dict[_SettingsKeyT, SettingsAttribute] = {}
|
||||
if values:
|
||||
self.update(values, priority)
|
||||
|
||||
def __getitem__(self, opt_name):
|
||||
def __getitem__(self, opt_name: _SettingsKeyT) -> Any:
|
||||
if opt_name not in self:
|
||||
return None
|
||||
return self.attributes[opt_name].value
|
||||
|
||||
def __contains__(self, name):
|
||||
def __contains__(self, name: Any) -> bool:
|
||||
return name in self.attributes
|
||||
|
||||
def get(self, name, default=None):
|
||||
def get(self, name: _SettingsKeyT, default: Any = None) -> Any:
|
||||
"""
|
||||
Get a setting value without affecting its original type.
|
||||
|
||||
|
|
@ -103,7 +135,7 @@ class BaseSettings(MutableMapping):
|
|||
"""
|
||||
return self[name] if self[name] is not None else default
|
||||
|
||||
def getbool(self, name, default=False):
|
||||
def getbool(self, name: _SettingsKeyT, default: bool = False) -> bool:
|
||||
"""
|
||||
Get a setting value as a boolean.
|
||||
|
||||
|
|
@ -133,7 +165,7 @@ class BaseSettings(MutableMapping):
|
|||
"'True'/'False' and 'true'/'false'"
|
||||
)
|
||||
|
||||
def getint(self, name, default=0):
|
||||
def getint(self, name: _SettingsKeyT, default: int = 0) -> int:
|
||||
"""
|
||||
Get a setting value as an int.
|
||||
|
||||
|
|
@ -145,7 +177,7 @@ class BaseSettings(MutableMapping):
|
|||
"""
|
||||
return int(self.get(name, default))
|
||||
|
||||
def getfloat(self, name, default=0.0):
|
||||
def getfloat(self, name: _SettingsKeyT, default: float = 0.0) -> float:
|
||||
"""
|
||||
Get a setting value as a float.
|
||||
|
||||
|
|
@ -157,7 +189,9 @@ class BaseSettings(MutableMapping):
|
|||
"""
|
||||
return float(self.get(name, default))
|
||||
|
||||
def getlist(self, name, default=None):
|
||||
def getlist(
|
||||
self, name: _SettingsKeyT, default: Optional[List[Any]] = None
|
||||
) -> List[Any]:
|
||||
"""
|
||||
Get a setting value as a list. If the setting original type is a list, a
|
||||
copy of it will be returned. If it's a string it will be split by ",".
|
||||
|
|
@ -176,7 +210,9 @@ class BaseSettings(MutableMapping):
|
|||
value = value.split(",")
|
||||
return list(value)
|
||||
|
||||
def getdict(self, name, default=None):
|
||||
def getdict(
|
||||
self, name: _SettingsKeyT, default: Optional[Dict[Any, Any]] = None
|
||||
) -> Dict[Any, Any]:
|
||||
"""
|
||||
Get a setting value as a dictionary. If the setting original type is a
|
||||
dictionary, a copy of it will be returned. If it is a string it will be
|
||||
|
|
@ -197,7 +233,11 @@ class BaseSettings(MutableMapping):
|
|||
value = json.loads(value)
|
||||
return dict(value)
|
||||
|
||||
def getdictorlist(self, name, default=None):
|
||||
def getdictorlist(
|
||||
self,
|
||||
name: _SettingsKeyT,
|
||||
default: Union[Dict[Any, Any], List[Any], None] = None,
|
||||
) -> Union[Dict[Any, Any], List[Any]]:
|
||||
"""Get a setting value as either a :class:`dict` or a :class:`list`.
|
||||
|
||||
If the setting is already a dict or a list, a copy of it will be
|
||||
|
|
@ -224,24 +264,29 @@ class BaseSettings(MutableMapping):
|
|||
return {}
|
||||
if isinstance(value, str):
|
||||
try:
|
||||
return json.loads(value)
|
||||
value_loaded = json.loads(value)
|
||||
assert isinstance(value_loaded, (dict, list))
|
||||
return value_loaded
|
||||
except ValueError:
|
||||
return value.split(",")
|
||||
assert isinstance(value, (dict, list))
|
||||
return copy.deepcopy(value)
|
||||
|
||||
def getwithbase(self, name):
|
||||
def getwithbase(self, name: _SettingsKeyT) -> "BaseSettings":
|
||||
"""Get a composition of a dictionary-like setting and its `_BASE`
|
||||
counterpart.
|
||||
|
||||
:param name: name of the dictionary-like setting
|
||||
:type name: str
|
||||
"""
|
||||
if not isinstance(name, str):
|
||||
raise ValueError(f"Base setting key must be a string, got {name}")
|
||||
compbs = BaseSettings()
|
||||
compbs.update(self[name + "_BASE"])
|
||||
compbs.update(self[name])
|
||||
return compbs
|
||||
|
||||
def getpriority(self, name):
|
||||
def getpriority(self, name: _SettingsKeyT) -> Optional[int]:
|
||||
"""
|
||||
Return the current numerical priority value of a setting, or ``None`` if
|
||||
the given ``name`` does not exist.
|
||||
|
|
@ -253,7 +298,7 @@ class BaseSettings(MutableMapping):
|
|||
return None
|
||||
return self.attributes[name].priority
|
||||
|
||||
def maxpriority(self):
|
||||
def maxpriority(self) -> int:
|
||||
"""
|
||||
Return the numerical value of the highest priority present throughout
|
||||
all settings, or the numerical value for ``default`` from
|
||||
|
|
@ -261,13 +306,15 @@ class BaseSettings(MutableMapping):
|
|||
stored.
|
||||
"""
|
||||
if len(self) > 0:
|
||||
return max(self.getpriority(name) for name in self)
|
||||
return max(cast(int, self.getpriority(name)) for name in self)
|
||||
return get_settings_priority("default")
|
||||
|
||||
def __setitem__(self, name, value):
|
||||
def __setitem__(self, name: _SettingsKeyT, value: Any) -> None:
|
||||
self.set(name, value)
|
||||
|
||||
def set(self, name, value, priority="project"):
|
||||
def set(
|
||||
self, name: _SettingsKeyT, value: Any, priority: Union[int, str] = "project"
|
||||
) -> None:
|
||||
"""
|
||||
Store a key/value attribute with a given priority.
|
||||
|
||||
|
|
@ -295,17 +342,26 @@ class BaseSettings(MutableMapping):
|
|||
else:
|
||||
self.attributes[name].set(value, priority)
|
||||
|
||||
def setdefault(self, name, default=None, priority="project"):
|
||||
def setdefault(
|
||||
self,
|
||||
name: _SettingsKeyT,
|
||||
default: Any = None,
|
||||
priority: Union[int, str] = "project",
|
||||
) -> Any:
|
||||
if name not in self:
|
||||
self.set(name, default, priority)
|
||||
return default
|
||||
|
||||
return self.attributes[name].value
|
||||
|
||||
def setdict(self, values, priority="project"):
|
||||
def setdict(
|
||||
self, values: _SettingsInputT, priority: Union[int, str] = "project"
|
||||
) -> None:
|
||||
self.update(values, priority)
|
||||
|
||||
def setmodule(self, module, priority="project"):
|
||||
def setmodule(
|
||||
self, module: Union[ModuleType, str], priority: Union[int, str] = "project"
|
||||
) -> None:
|
||||
"""
|
||||
Store settings from a module with a given priority.
|
||||
|
||||
|
|
@ -327,7 +383,8 @@ class BaseSettings(MutableMapping):
|
|||
if key.isupper():
|
||||
self.set(key, getattr(module, key), priority)
|
||||
|
||||
def update(self, values, priority="project"):
|
||||
# BaseSettings.update() doesn't support all inputs that MutableMapping.update() supports
|
||||
def update(self, values: _SettingsInputT, priority: Union[int, str] = "project") -> None: # type: ignore[override]
|
||||
"""
|
||||
Store key/value pairs with a given priority.
|
||||
|
||||
|
|
@ -351,30 +408,34 @@ class BaseSettings(MutableMapping):
|
|||
"""
|
||||
self._assert_mutability()
|
||||
if isinstance(values, str):
|
||||
values = json.loads(values)
|
||||
values = cast(dict, json.loads(values))
|
||||
if values is not None:
|
||||
if isinstance(values, BaseSettings):
|
||||
for name, value in values.items():
|
||||
self.set(name, value, values.getpriority(name))
|
||||
self.set(name, value, cast(int, values.getpriority(name)))
|
||||
else:
|
||||
for name, value in values.items():
|
||||
self.set(name, value, priority)
|
||||
|
||||
def delete(self, name, priority="project"):
|
||||
def delete(
|
||||
self, name: _SettingsKeyT, priority: Union[int, str] = "project"
|
||||
) -> None:
|
||||
if name not in self:
|
||||
raise KeyError(name)
|
||||
self._assert_mutability()
|
||||
priority = get_settings_priority(priority)
|
||||
if priority >= self.getpriority(name):
|
||||
if priority >= cast(int, self.getpriority(name)):
|
||||
del self.attributes[name]
|
||||
|
||||
def __delitem__(self, name):
|
||||
def __delitem__(self, name: _SettingsKeyT) -> None:
|
||||
self._assert_mutability()
|
||||
del self.attributes[name]
|
||||
|
||||
def _assert_mutability(self):
|
||||
def _assert_mutability(self) -> None:
|
||||
if self.frozen:
|
||||
raise TypeError("Trying to modify an immutable Settings object")
|
||||
|
||||
def copy(self):
|
||||
def copy(self) -> "Self":
|
||||
"""
|
||||
Make a deep copy of current settings.
|
||||
|
||||
|
|
@ -386,7 +447,7 @@ class BaseSettings(MutableMapping):
|
|||
"""
|
||||
return copy.deepcopy(self)
|
||||
|
||||
def freeze(self):
|
||||
def freeze(self) -> None:
|
||||
"""
|
||||
Disable further changes to the current settings.
|
||||
|
||||
|
|
@ -396,7 +457,7 @@ class BaseSettings(MutableMapping):
|
|||
"""
|
||||
self.frozen = True
|
||||
|
||||
def frozencopy(self):
|
||||
def frozencopy(self) -> "Self":
|
||||
"""
|
||||
Return an immutable copy of the current settings.
|
||||
|
||||
|
|
@ -406,26 +467,26 @@ class BaseSettings(MutableMapping):
|
|||
copy.freeze()
|
||||
return copy
|
||||
|
||||
def __iter__(self):
|
||||
def __iter__(self) -> Iterator[_SettingsKeyT]:
|
||||
return iter(self.attributes)
|
||||
|
||||
def __len__(self):
|
||||
def __len__(self) -> int:
|
||||
return len(self.attributes)
|
||||
|
||||
def _to_dict(self):
|
||||
def _to_dict(self) -> Dict[_SettingsKeyT, Any]:
|
||||
return {
|
||||
self._get_key(k): (v._to_dict() if isinstance(v, BaseSettings) else v)
|
||||
for k, v in self.items()
|
||||
}
|
||||
|
||||
def _get_key(self, key_value):
|
||||
def _get_key(self, key_value: Any) -> _SettingsKeyT:
|
||||
return (
|
||||
key_value
|
||||
if isinstance(key_value, (bool, float, int, str, type(None)))
|
||||
else str(key_value)
|
||||
)
|
||||
|
||||
def copy_to_dict(self):
|
||||
def copy_to_dict(self) -> Dict[_SettingsKeyT, Any]:
|
||||
"""
|
||||
Make a copy of current settings and convert to a dict.
|
||||
|
||||
|
|
@ -441,13 +502,14 @@ class BaseSettings(MutableMapping):
|
|||
settings = self.copy()
|
||||
return settings._to_dict()
|
||||
|
||||
def _repr_pretty_(self, p, cycle):
|
||||
# https://ipython.readthedocs.io/en/stable/config/integrating.html#pretty-printing
|
||||
def _repr_pretty_(self, p: Any, cycle: bool) -> None:
|
||||
if cycle:
|
||||
p.text(repr(self))
|
||||
else:
|
||||
p.text(pformat(self.copy_to_dict()))
|
||||
|
||||
def pop(self, name, default=__default):
|
||||
def pop(self, name: _SettingsKeyT, default: Any = __default) -> Any:
|
||||
try:
|
||||
value = self.attributes[name].value
|
||||
except KeyError:
|
||||
|
|
@ -471,7 +533,9 @@ class Settings(BaseSettings):
|
|||
described on :ref:`topics-settings-ref` already populated.
|
||||
"""
|
||||
|
||||
def __init__(self, values=None, priority="project"):
|
||||
def __init__(
|
||||
self, values: _SettingsInputT = None, priority: Union[int, str] = "project"
|
||||
):
|
||||
# Do not pass kwarg values here. We don't want to promote user-defined
|
||||
# dicts, and we want to update, not replace, default dicts with the
|
||||
# values given by the user
|
||||
|
|
@ -485,14 +549,16 @@ class Settings(BaseSettings):
|
|||
self.update(values, priority)
|
||||
|
||||
|
||||
def iter_default_settings():
|
||||
def iter_default_settings() -> Iterable[Tuple[str, Any]]:
|
||||
"""Return the default settings as an iterator of (name, value) tuples"""
|
||||
for name in dir(default_settings):
|
||||
if name.isupper():
|
||||
yield name, getattr(default_settings, name)
|
||||
|
||||
|
||||
def overridden_settings(settings):
|
||||
def overridden_settings(
|
||||
settings: Mapping[_SettingsKeyT, Any]
|
||||
) -> Iterable[Tuple[str, Any]]:
|
||||
"""Return a dict of the settings that have been overridden"""
|
||||
for name, defvalue in iter_default_settings():
|
||||
value = settings[name]
|
||||
|
|
|
|||
|
|
@ -28,6 +28,7 @@ def build_component_list(compdict, custom=None, convert=update_classpath):
|
|||
compbs = BaseSettings()
|
||||
for k, v in compdict.items():
|
||||
prio = compdict.getpriority(k)
|
||||
assert prio is not None
|
||||
if compbs.getpriority(convert(k)) == prio:
|
||||
raise ValueError(
|
||||
f"Some paths in {list(compdict.keys())!r} "
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from scrapy.settings import (
|
||||
SETTINGS_PRIORITIES,
|
||||
BaseSettings,
|
||||
|
|
@ -199,6 +201,21 @@ class BaseSettingsTest(unittest.TestCase):
|
|||
settings.update({"key_lowprio": 3}, priority=20)
|
||||
self.assertEqual(settings["key_lowprio"], 1)
|
||||
|
||||
@pytest.mark.xfail(
|
||||
raises=TypeError, reason="BaseSettings.update doesn't support kwargs input"
|
||||
)
|
||||
def test_update_kwargs(self):
|
||||
settings = BaseSettings({"key": 0})
|
||||
settings.update(key=1)
|
||||
|
||||
@pytest.mark.xfail(
|
||||
raises=AttributeError,
|
||||
reason="BaseSettings.update doesn't support iterable input",
|
||||
)
|
||||
def test_update_iterable(self):
|
||||
settings = BaseSettings({"key": 0})
|
||||
settings.update([("key", 1)])
|
||||
|
||||
def test_update_jsonstring(self):
|
||||
settings = BaseSettings({"number": 0, "dict": BaseSettings({"key": "val"})})
|
||||
settings.update('{"number": 1, "newnumber": 2}')
|
||||
|
|
@ -217,6 +234,10 @@ class BaseSettingsTest(unittest.TestCase):
|
|||
self.assertIn("key_highprio", settings)
|
||||
del settings["key_highprio"]
|
||||
self.assertNotIn("key_highprio", settings)
|
||||
with self.assertRaises(KeyError):
|
||||
settings.delete("notkey")
|
||||
with self.assertRaises(KeyError):
|
||||
del settings["notkey"]
|
||||
|
||||
def test_get(self):
|
||||
test_configuration = {
|
||||
|
|
|
|||
Loading…
Reference in New Issue