mirror of
https://github.com/Kozea/Radicale.git
synced 2025-06-29 16:55:32 +00:00
More type hints
This commit is contained in:
Unrud
parent
11f7559ef3
commit
c93d7b8715
51 changed files with 1374 additions and 957 deletions
|
|
@ -29,25 +29,27 @@ import contextlib
|
|||
import math
|
||||
import os
|
||||
import string
|
||||
import sys
|
||||
from collections import OrderedDict
|
||||
from configparser import RawConfigParser
|
||||
from typing import Any, ClassVar
|
||||
from typing import (Any, Callable, ClassVar, Iterable, List, Optional,
|
||||
Sequence, Tuple, TypeVar, Union)
|
||||
|
||||
from radicale import auth, rights, storage, web
|
||||
from radicale import auth, rights, storage, types, web
|
||||
|
||||
DEFAULT_CONFIG_PATH = os.pathsep.join([
|
||||
DEFAULT_CONFIG_PATH: str = os.pathsep.join([
|
||||
"?/etc/radicale/config",
|
||||
"?~/.config/radicale/config"])
|
||||
|
||||
|
||||
def positive_int(value):
|
||||
def positive_int(value: Any) -> int:
|
||||
value = int(value)
|
||||
if value < 0:
|
||||
raise ValueError("value is negative: %d" % value)
|
||||
return value
|
||||
|
||||
|
||||
def positive_float(value):
|
||||
def positive_float(value: Any) -> float:
|
||||
value = float(value)
|
||||
if not math.isfinite(value):
|
||||
raise ValueError("value is infinite")
|
||||
|
|
@ -58,22 +60,22 @@ def positive_float(value):
|
|||
return value
|
||||
|
||||
|
||||
def logging_level(value):
|
||||
def logging_level(value: Any) -> str:
|
||||
if value not in ("debug", "info", "warning", "error", "critical"):
|
||||
raise ValueError("unsupported level: %r" % value)
|
||||
return value
|
||||
|
||||
|
||||
def filepath(value):
|
||||
def filepath(value: Any) -> str:
|
||||
if not value:
|
||||
return ""
|
||||
value = os.path.expanduser(value)
|
||||
if os.name == "nt":
|
||||
if sys.platform == "win32":
|
||||
value = os.path.expandvars(value)
|
||||
return os.path.abspath(value)
|
||||
|
||||
|
||||
def list_of_ip_address(value):
|
||||
def list_of_ip_address(value: Any) -> List[Tuple[str, int]]:
|
||||
def ip_address(value):
|
||||
try:
|
||||
address, port = value.rsplit(":", 1)
|
||||
|
|
@ -83,25 +85,25 @@ def list_of_ip_address(value):
|
|||
return [ip_address(s) for s in value.split(",")]
|
||||
|
||||
|
||||
def str_or_callable(value):
|
||||
def str_or_callable(value: Any) -> Union[str, Callable]:
|
||||
if callable(value):
|
||||
return value
|
||||
return str(value)
|
||||
|
||||
|
||||
def unspecified_type(value):
|
||||
def unspecified_type(value: Any) -> Any:
|
||||
return value
|
||||
|
||||
|
||||
def _convert_to_bool(value):
|
||||
def _convert_to_bool(value: Any) -> bool:
|
||||
if value.lower() not in RawConfigParser.BOOLEAN_STATES:
|
||||
raise ValueError("not a boolean: %r" % value)
|
||||
return RawConfigParser.BOOLEAN_STATES[value.lower()]
|
||||
|
||||
|
||||
INTERNAL_OPTIONS = ("_allow_extra",)
|
||||
INTERNAL_OPTIONS: Sequence[str] = ("_allow_extra",)
|
||||
# Default configuration
|
||||
DEFAULT_CONFIG_SCHEMA = OrderedDict([
|
||||
DEFAULT_CONFIG_SCHEMA: types.CONFIG_SCHEMA = OrderedDict([
|
||||
("server", OrderedDict([
|
||||
("hosts", {
|
||||
"value": "localhost:5232",
|
||||
|
|
@ -227,7 +229,8 @@ DEFAULT_CONFIG_SCHEMA = OrderedDict([
|
|||
("_allow_extra", str)]))])
|
||||
|
||||
|
||||
def parse_compound_paths(*compound_paths):
|
||||
def parse_compound_paths(*compound_paths: Optional[str]
|
||||
) -> List[Tuple[str, bool]]:
|
||||
"""Parse a compound path and return the individual paths.
|
||||
Paths in a compound path are joined by ``os.pathsep``. If a path starts
|
||||
with ``?`` the return value ``IGNORE_IF_MISSING`` is set.
|
||||
|
|
@ -253,7 +256,8 @@ def parse_compound_paths(*compound_paths):
|
|||
return paths
|
||||
|
||||
|
||||
def load(paths=()):
|
||||
def load(paths: Optional[Iterable[Tuple[str, bool]]] = None
|
||||
) -> "Configuration":
|
||||
"""
|
||||
Create instance of ``Configuration`` for use with
|
||||
``radicale.app.Application``.
|
||||
|
|
@ -266,6 +270,8 @@ def load(paths=()):
|
|||
The configuration can later be changed with ``Configuration.update()``.
|
||||
|
||||
"""
|
||||
if paths is None:
|
||||
paths = []
|
||||
configuration = Configuration(DEFAULT_CONFIG_SCHEMA)
|
||||
for path, ignore_if_missing in paths:
|
||||
parser = RawConfigParser()
|
||||
|
|
@ -279,16 +285,24 @@ def load(paths=()):
|
|||
config = {s: {o: parser[s][o] for o in parser.options(s)}
|
||||
for s in parser.sections()}
|
||||
except Exception as e:
|
||||
raise RuntimeError(
|
||||
"Failed to load %s: %s" % (config_source, e)) from e
|
||||
raise RuntimeError("Failed to load %s: %s" % (config_source, e)
|
||||
) from e
|
||||
configuration.update(config, config_source)
|
||||
return configuration
|
||||
|
||||
|
||||
class Configuration:
|
||||
SOURCE_MISSING: ClassVar[Any] = {}
|
||||
_Self = TypeVar("_Self", bound="Configuration")
|
||||
|
||||
def __init__(self, schema):
|
||||
|
||||
class Configuration:
|
||||
|
||||
SOURCE_MISSING: ClassVar[types.CONFIG] = {}
|
||||
|
||||
_schema: types.CONFIG_SCHEMA
|
||||
_values: types.MUTABLE_CONFIG
|
||||
_configs: List[Tuple[types.CONFIG, str, bool]]
|
||||
|
||||
def __init__(self, schema: types.CONFIG_SCHEMA) -> None:
|
||||
"""Initialize configuration.
|
||||
|
||||
``schema`` a dict that describes the configuration format.
|
||||
|
|
@ -309,7 +323,8 @@ class Configuration:
|
|||
for section in self._schema}
|
||||
self.update(default, "default config", privileged=True)
|
||||
|
||||
def update(self, config, source=None, privileged=False):
|
||||
def update(self, config: types.CONFIG, source: Optional[str] = None,
|
||||
privileged: bool = False) -> None:
|
||||
"""Update the configuration.
|
||||
|
||||
``config`` a dict of the format {SECTION: {OPTION: VALUE, ...}, ...}.
|
||||
|
|
@ -323,8 +338,9 @@ class Configuration:
|
|||
``privileged`` allows updating sections and options starting with "_".
|
||||
|
||||
"""
|
||||
source = source or "unspecified config"
|
||||
new_values = {}
|
||||
if source is None:
|
||||
source = "unspecified config"
|
||||
new_values: types.MUTABLE_CONFIG = {}
|
||||
for section in config:
|
||||
if (section not in self._schema or
|
||||
section.startswith("_") and not privileged):
|
||||
|
|
@ -363,40 +379,41 @@ class Configuration:
|
|||
self._values[section] = self._values.get(section, {})
|
||||
self._values[section].update(new_values[section])
|
||||
|
||||
def get(self, section, option):
|
||||
def get(self, section: str, option: str) -> Any:
|
||||
"""Get the value of ``option`` in ``section``."""
|
||||
with contextlib.suppress(KeyError):
|
||||
return self._values[section][option]
|
||||
raise KeyError(section, option)
|
||||
|
||||
def get_raw(self, section, option):
|
||||
def get_raw(self, section: str, option: str) -> Any:
|
||||
"""Get the raw value of ``option`` in ``section``."""
|
||||
for config, _, _ in reversed(self._configs):
|
||||
if option in config.get(section, {}):
|
||||
return config[section][option]
|
||||
raise KeyError(section, option)
|
||||
|
||||
def get_source(self, section, option):
|
||||
def get_source(self, section: str, option: str) -> str:
|
||||
"""Get the source that provides ``option`` in ``section``."""
|
||||
for config, source, _ in reversed(self._configs):
|
||||
if option in config.get(section, {}):
|
||||
return source
|
||||
raise KeyError(section, option)
|
||||
|
||||
def sections(self):
|
||||
def sections(self) -> List[str]:
|
||||
"""List all sections."""
|
||||
return self._values.keys()
|
||||
return list(self._values.keys())
|
||||
|
||||
def options(self, section):
|
||||
def options(self, section: str) -> List[str]:
|
||||
"""List all options in ``section``"""
|
||||
return self._values[section].keys()
|
||||
return list(self._values[section].keys())
|
||||
|
||||
def sources(self):
|
||||
def sources(self) -> List[Tuple[str, bool]]:
|
||||
"""List all config sources."""
|
||||
return [(source, config is self.SOURCE_MISSING) for
|
||||
config, source, _ in self._configs]
|
||||
|
||||
def copy(self, plugin_schema=None):
|
||||
def copy(self: _Self, plugin_schema: Optional[types.CONFIG_SCHEMA] = None
|
||||
) -> _Self:
|
||||
"""Create a copy of the configuration
|
||||
|
||||
``plugin_schema`` is a optional dict that contains additional options
|
||||
|
|
@ -406,20 +423,23 @@ class Configuration:
|
|||
if plugin_schema is None:
|
||||
schema = self._schema
|
||||
else:
|
||||
schema = self._schema.copy()
|
||||
new_schema = dict(self._schema)
|
||||
for section, options in plugin_schema.items():
|
||||
if (section not in schema or "type" not in schema[section] or
|
||||
"internal" not in schema[section]["type"]):
|
||||
if (section not in new_schema or
|
||||
"type" not in new_schema[section] or
|
||||
"internal" not in new_schema[section]["type"]):
|
||||
raise ValueError("not a plugin section: %r" % section)
|
||||
schema[section] = schema[section].copy()
|
||||
schema[section]["type"] = schema[section]["type"].copy()
|
||||
schema[section]["type"]["internal"] = [
|
||||
self.get(section, "type")]
|
||||
new_section = dict(new_schema[section])
|
||||
new_type = dict(new_section["type"])
|
||||
new_type["internal"] = (self.get(section, "type"),)
|
||||
new_section["type"] = new_type
|
||||
for option, value in options.items():
|
||||
if option in schema[section]:
|
||||
raise ValueError("option already exists in %r: %r" % (
|
||||
section, option))
|
||||
schema[section][option] = value
|
||||
if option in new_section:
|
||||
raise ValueError("option already exists in %r: %r" %
|
||||
(section, option))
|
||||
new_section[option] = value
|
||||
new_schema[section] = new_section
|
||||
schema = new_schema
|
||||
copy = type(self)(schema)
|
||||
for config, source, privileged in self._configs:
|
||||
copy.update(config, source, privileged)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue