diff options
| author | Colin Watson <cjwatson@debian.org> | 2025-02-24 13:08:37 +0000 |
|---|---|---|
| committer | git-ubuntu importer <ubuntu-devel-discuss@lists.ubuntu.com> | 2025-02-24 16:43:26 +0000 |
| commit | 2ba50cddcc72ba83e95358f1a5fdfc06efd1654d (patch) | |
| tree | 91dd65a240e8c7acf6c4c87a901e694d2e252500 /box | |
| parent | 225d8213a7da9cbc1681b74853238a27c4838ecd (diff) | |
7.3.2-1 (patches unapplied)import/7.3.2-1ubuntu/questing-develubuntu/questingubuntu/plucky-proposedubuntu/plucky-develubuntu/pluckydebian/sid
Imported using git-ubuntu import.
Notes
Notes:
* Team upload.
[ Debian Janitor ]
* Update standards version to 4.6.1, no changes needed.
[ Colin Watson ]
* d/watch: Adjust for PyPI normalization.
* New upstream release. Now Architecture: any due to using Cython.
* Use pybuild-plugin-pyproject.
* Mark autopkgtest as superficial (closes: #974492).
Diffstat (limited to 'box')
| -rw-r--r-- | box/__init__.py | 24 | ||||
| -rw-r--r-- | box/box.py | 1195 | ||||
| -rw-r--r-- | box/box.pyi | 121 | ||||
| -rw-r--r-- | box/box_list.py | 476 | ||||
| -rw-r--r-- | box/box_list.pyi | 83 | ||||
| -rw-r--r-- | box/config_box.py | 133 | ||||
| -rw-r--r-- | box/config_box.pyi | 15 | ||||
| -rw-r--r-- | box/converters.py | 363 | ||||
| -rw-r--r-- | box/converters.pyi | 60 | ||||
| -rw-r--r-- | box/exceptions.py | 22 | ||||
| -rw-r--r-- | box/exceptions.pyi | 5 | ||||
| -rw-r--r-- | box/from_file.py | 149 | ||||
| -rw-r--r-- | box/from_file.pyi | 16 | ||||
| -rw-r--r-- | box/py.typed | 0 | ||||
| -rw-r--r-- | box/shorthand_box.py | 69 | ||||
| -rw-r--r-- | box/shorthand_box.pyi | 17 |
16 files changed, 2748 insertions, 0 deletions
diff --git a/box/__init__.py b/box/__init__.py new file mode 100644 index 0000000..7cd918f --- /dev/null +++ b/box/__init__.py @@ -0,0 +1,24 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +__author__ = "Chris Griffith" +__version__ = "7.3.2" + +from box.box import Box +from box.box_list import BoxList +from box.config_box import ConfigBox +from box.exceptions import BoxError, BoxKeyError +from box.from_file import box_from_file, box_from_string +from box.shorthand_box import SBox, DDBox +import box.converters + +__all__ = [ + "Box", + "BoxList", + "ConfigBox", + "BoxError", + "BoxKeyError", + "box_from_file", + "SBox", + "DDBox", +] diff --git a/box/box.py b/box/box.py new file mode 100644 index 0000000..8be6738 --- /dev/null +++ b/box/box.py @@ -0,0 +1,1195 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# +# Copyright (c) 2017-2023 - Chris Griffith - MIT License +""" +Improved dictionary access through dot notation with additional tools. +""" +import copy +import re +import warnings +from keyword import iskeyword +from os import PathLike +from typing import Any, Dict, Generator, List, Optional, Tuple, Type, Union, Literal +from inspect import signature + +try: + from typing import Callable, Iterable, Mapping +except ImportError: + from collections.abc import Callable, Iterable, Mapping + + +import box +from box.converters import ( + BOX_PARAMETERS, + _from_json, + _from_msgpack, + _from_toml, + _from_yaml, + _to_json, + _to_msgpack, + _to_toml, + _to_yaml, + msgpack_available, + toml_read_library, + toml_write_library, + yaml_available, +) +from box.exceptions import BoxError, BoxKeyError, BoxTypeError, BoxValueError, BoxWarning + +__all__ = ["Box"] + +_first_cap_re = re.compile("(.)([A-Z][a-z]+)") +_all_cap_re = re.compile("([a-z0-9])([A-Z])") +_list_pos_re = re.compile(r"\[(\d+)\]") + +# a sentinel object for indicating no default, in order to allow users +# to pass `None` as a valid default value +NO_DEFAULT = object() +# a sentinel object for indicating when to skip adding a new namespace, allowing `None` keys +NO_NAMESPACE = object() + + +def _is_ipython(): + try: + from IPython import get_ipython + except ImportError: + ipython = False + else: + ipython = True if get_ipython() else False + + return ipython + + +def _exception_cause(e): + """ + Unwrap BoxKeyError and BoxValueError errors to their cause. + + Use with `raise ... from _exception_cause(err)` to avoid deeply nested stacktraces, but keep the + context. + """ + return e.__cause__ if isinstance(e, (BoxKeyError, BoxValueError)) else e + + +def _camel_killer(attr): + """ + CamelKiller, qu'est-ce que c'est? + + Taken from http://stackoverflow.com/a/1176023/3244542 + """ + attr = str(attr) + + s1 = _first_cap_re.sub(r"\1_\2", attr) + s2 = _all_cap_re.sub(r"\1_\2", s1) + return re.sub(" *_+", "_", s2.lower()) + + +def _recursive_tuples(iterable, box_class, recreate_tuples=False, **kwargs): + out_list = [] + for i in iterable: + if isinstance(i, dict): + out_list.append(box_class(i, **kwargs)) + elif isinstance(i, list) or (recreate_tuples and isinstance(i, tuple)): + out_list.append(_recursive_tuples(i, box_class, recreate_tuples, **kwargs)) + else: + out_list.append(i) + return tuple(out_list) + + +def _parse_box_dots(bx, item, setting=False): + for idx, char in enumerate(item): + if char == "[": + return item[:idx], item[idx:] + elif char == ".": + return item[:idx], item[idx + 1 :] + if setting and "." in item: + return item.split(".", 1) + raise BoxError("Could not split box dots properly") + + +def _get_dot_paths(bx, current=""): + """A generator of all the end node keys in a box in box_dots format""" + + def handle_dicts(sub_bx, paths=""): + for key, value in sub_bx.items(): + yield f"{paths}.{key}" if paths else key + if isinstance(value, dict): + yield from handle_dicts(value, f"{paths}.{key}" if paths else key) + elif isinstance(value, list): + yield from handle_lists(value, f"{paths}.{key}" if paths else key) + + def handle_lists(bx_list, paths=""): + for i, value in enumerate(bx_list): + yield f"{paths}[{i}]" + if isinstance(value, list): + yield from handle_lists(value, f"{paths}[{i}]") + if isinstance(value, dict): + yield from handle_dicts(value, f"{paths}[{i}]") + + yield from handle_dicts(bx, current) + + +def _get_box_config(): + return { + # Internal use only + "__created": False, + "__safe_keys": {}, + } + + +def _get_property_func(obj, key): + """ + Try to get property helper functions of given object and property name. + + :param obj: object to be checked for property + :param key: property name + :return: a tuple for helper functions(fget, fset, fdel). If no such property, a (None, None, None) returns + """ + obj_type = type(obj) + + if not hasattr(obj_type, key): + return None, None, None + attr = getattr(obj_type, key) + return attr.fget, attr.fset, attr.fdel + + +class Box(dict): + """ + Improved dictionary access through dot notation with additional tools. + + :param default_box: Similar to defaultdict, return a default value + :param default_box_attr: Specify the default replacement. + WARNING: If this is not the default 'Box', it will not be recursive + :param default_box_none_transform: When using default_box, treat keys with none values as absent. True by default + :param default_box_create_on_get: On lookup of a key that doesn't exist, create it if missing + :param frozen_box: After creation, the box cannot be modified + :param camel_killer_box: Convert CamelCase to snake_case + :param conversion_box: Check for near matching keys as attributes + :param modify_tuples_box: Recreate incoming tuples with dicts into Boxes + :param box_safe_prefix: Conversion box prefix for unsafe attributes + :param box_duplicates: "ignore", "error" or "warn" when duplicates exists in a conversion_box + :param box_intact_types: tuple of types to ignore converting + :param box_recast: cast certain keys to a specified type + :param box_dots: access nested Boxes by period separated keys in string + :param box_class: change what type of class sub-boxes will be created as + :param box_namespace: the namespace this (possibly nested) Box lives within + """ + + _box_config: Dict[str, Any] + + _protected_keys = [ + "to_dict", + "to_json", + "to_yaml", + "from_yaml", + "from_json", + "from_toml", + "to_toml", + "merge_update", + ] + [attr for attr in dir({}) if not attr.startswith("_")] + + def __new__( + cls, + *args: Any, + default_box: bool = False, + default_box_attr: Any = NO_DEFAULT, + default_box_none_transform: bool = True, + default_box_create_on_get: bool = True, + frozen_box: bool = False, + camel_killer_box: bool = False, + conversion_box: bool = True, + modify_tuples_box: bool = False, + box_safe_prefix: str = "x", + box_duplicates: str = "ignore", + box_intact_types: Union[Tuple, List] = (), + box_recast: Optional[Dict] = None, + box_dots: bool = False, + box_class: Optional[Union[Dict, Type["Box"]]] = None, + box_namespace: Union[Tuple[str, ...], Literal[False]] = (), + **kwargs: Any, + ): + """ + Due to the way pickling works in python 3, we need to make sure + the box config is created as early as possible. + """ + obj = super().__new__(cls, *args, **kwargs) + obj._box_config = _get_box_config() + obj._box_config.update( + { + "default_box": default_box, + "default_box_attr": cls.__class__ if default_box_attr is NO_DEFAULT else default_box_attr, + "default_box_none_transform": default_box_none_transform, + "default_box_create_on_get": default_box_create_on_get, + "conversion_box": conversion_box, + "box_safe_prefix": box_safe_prefix, + "frozen_box": frozen_box, + "camel_killer_box": camel_killer_box, + "modify_tuples_box": modify_tuples_box, + "box_duplicates": box_duplicates, + "box_intact_types": tuple(box_intact_types), + "box_recast": box_recast, + "box_dots": box_dots, + "box_class": box_class if box_class is not None else Box, + "box_namespace": box_namespace, + } + ) + return obj + + def __init__( + self, + *args: Any, + default_box: bool = False, + default_box_attr: Any = NO_DEFAULT, + default_box_none_transform: bool = True, + default_box_create_on_get: bool = True, + frozen_box: bool = False, + camel_killer_box: bool = False, + conversion_box: bool = True, + modify_tuples_box: bool = False, + box_safe_prefix: str = "x", + box_duplicates: str = "ignore", + box_intact_types: Union[Tuple, List] = (), + box_recast: Optional[Dict] = None, + box_dots: bool = False, + box_class: Optional[Union[Dict, Type["Box"]]] = None, + box_namespace: Union[Tuple[str, ...], Literal[False]] = (), + **kwargs: Any, + ): + super().__init__() + self._box_config = _get_box_config() + self._box_config.update( + { + "default_box": default_box, + "default_box_attr": self.__class__ if default_box_attr is NO_DEFAULT else default_box_attr, + "default_box_none_transform": default_box_none_transform, + "default_box_create_on_get": default_box_create_on_get, + "conversion_box": conversion_box, + "box_safe_prefix": box_safe_prefix, + "frozen_box": frozen_box, + "camel_killer_box": camel_killer_box, + "modify_tuples_box": modify_tuples_box, + "box_duplicates": box_duplicates, + "box_intact_types": tuple(box_intact_types), + "box_recast": box_recast, + "box_dots": box_dots, + "box_class": box_class if box_class is not None else self.__class__, + "box_namespace": box_namespace, + } + ) + if not self._box_config["conversion_box"] and self._box_config["box_duplicates"] != "ignore": + raise BoxError("box_duplicates are only for conversion_boxes") + if len(args) == 1: + if isinstance(args[0], str): + raise BoxValueError("Cannot extrapolate Box from string") + if isinstance(args[0], Mapping): + for k, v in args[0].items(): + if v is args[0]: + v = self + if v is None and self._box_config["default_box"] and self._box_config["default_box_none_transform"]: + continue + self.__setitem__(k, v) + elif isinstance(args[0], Iterable): + for k, v in args[0]: + self.__setitem__(k, v) + else: + raise BoxValueError("First argument must be mapping or iterable") + elif args: + raise BoxTypeError(f"Box expected at most 1 argument, got {len(args)}") + + for k, v in kwargs.items(): + if args and isinstance(args[0], Mapping) and v is args[0]: + v = self + self.__setitem__(k, v) + + self._box_config["__created"] = True + + def __add__(self, other: Mapping[Any, Any]): + if not isinstance(other, dict): + raise BoxTypeError("Box can only merge two boxes or a box and a dictionary.") + new_box = self.copy() + new_box._box_config["frozen_box"] = False + new_box.merge_update(other) # type: ignore[attr-defined] + new_box._box_config["frozen_box"] = self._box_config["frozen_box"] + return new_box + + def __radd__(self, other: Mapping[Any, Any]): + if not isinstance(other, dict): + raise BoxTypeError("Box can only merge two boxes or a box and a dictionary.") + + new_box = other.copy() + if not isinstance(other, Box): + new_box = self._box_config["box_class"](new_box) + new_box._box_config["frozen_box"] = False # type: ignore[attr-defined] + new_box.merge_update(self) # type: ignore[attr-defined] + new_box._box_config["frozen_box"] = self._box_config["frozen_box"] # type: ignore[attr-defined] + return new_box + + def __iadd__(self, other: Mapping[Any, Any]): + if not isinstance(other, dict): + raise BoxTypeError("Box can only merge two boxes or a box and a dictionary.") + self.merge_update(other) + return self + + def __or__(self, other: Mapping[Any, Any]): + if not isinstance(other, dict): + raise BoxTypeError("Box can only merge two boxes or a box and a dictionary.") + new_box = self.copy() + new_box._box_config["frozen_box"] = False + new_box.update(other) # type: ignore[attr-defined] + new_box._box_config["frozen_box"] = self._box_config["frozen_box"] + return new_box + + def __ror__(self, other: Mapping[Any, Any]): + if not isinstance(other, dict): + raise BoxTypeError("Box can only merge two boxes or a box and a dictionary.") + new_box = other.copy() + if not isinstance(other, Box): + new_box = self._box_config["box_class"](new_box) + new_box._box_config["frozen_box"] = False # type: ignore[attr-defined] + new_box.update(self) # type: ignore[attr-defined] + new_box._box_config["frozen_box"] = self._box_config["frozen_box"] # type: ignore[attr-defined] + return new_box + + def __ior__(self, other: Mapping[Any, Any]): # type: ignore[override] + if not isinstance(other, dict): + raise BoxTypeError("Box can only merge two boxes or a box and a dictionary.") + self.update(other) + return self + + def __sub__(self, other: Mapping[Any, Any]): + frozen = self._box_config["frozen_box"] + config = self.__box_config() + config["frozen_box"] = False + config.pop("box_namespace") # Detach namespace; it will be reassigned if we nest again + output = self._box_config["box_class"](**config) + if not isinstance(other, dict): + raise BoxError("Box can only compare two boxes or a box and a dictionary.") + if not isinstance(other, Box): + other = self._box_config["box_class"](other, **config) + for item in self: + if item not in other: + output[item] = self[item] + elif isinstance(self.get(item), Box) and isinstance(other.get(item), Box): + output[item] = self[item] - other[item] + if not output[item]: + del output[item] + output._box_config["frozen_box"] = frozen + return output + + def __hash__(self): + if self._box_config["frozen_box"]: + hashing = 54321 + for item in self.items(): + hashing ^= hash(item) + return hashing + raise BoxTypeError('unhashable type: "Box"') + + def __dir__(self) -> List[str]: + items = set(super().__dir__()) + # Only show items accessible by dot notation + for key in self.keys(): + key = str(key) + if key.isidentifier() and not iskeyword(key): + items.add(key) + + for key in self.keys(): + if key not in items: + if self._box_config["conversion_box"]: + key = self._safe_attr(key) + if key: + items.add(key) + + return list(items) + + def __contains__(self, item): + in_me = super().__contains__(item) + if not self._box_config["box_dots"] or not isinstance(item, str): + return in_me + if in_me: + return True + if "." not in item: + return False + try: + first_item, children = _parse_box_dots(self, item) + except BoxError: + return False + else: + if not super().__contains__(first_item): + return False + it = self[first_item] + return isinstance(it, Iterable) and children in it + + def keys(self, dotted: Union[bool] = False): + if not dotted: + return super().keys() + + if not self._box_config["box_dots"]: + raise BoxError("Cannot return dotted keys as this Box does not have `box_dots` enabled") + + keys = set() + for key, value in self.items(): + added = False + if isinstance(key, str): + if isinstance(value, Box): + for sub_key in value.keys(dotted=True): + keys.add(f"{key}.{sub_key}") + added = True + elif isinstance(value, box.BoxList): + for pos in value._dotted_helper(): + keys.add(f"{key}{pos}") + added = True + if not added: + keys.add(key) + return sorted(keys, key=lambda x: str(x)) + + def items(self, dotted: Union[bool] = False): + if not dotted: + return super().items() + + if not self._box_config["box_dots"]: + raise BoxError("Cannot return dotted keys as this Box does not have `box_dots` enabled") + + return [(k, self[k]) for k in self.keys(dotted=True)] + + def get(self, key, default=NO_DEFAULT): + if key not in self: + if default is NO_DEFAULT: + if self._box_config["default_box"] and self._box_config["default_box_none_transform"]: + return self.__get_default(key) + else: + return None + if isinstance(default, dict) and not isinstance(default, Box): + return Box(default) + if isinstance(default, list) and not isinstance(default, box.BoxList): + return box.BoxList(default) + return default + return self[key] + + def copy(self) -> "Box": + config = self.__box_config() + config.pop("box_namespace") # Detach namespace; it will be reassigned if we nest again + return Box(super().copy(), **config) + + def __copy__(self) -> "Box": + return self.copy() + + def __deepcopy__(self, memodict=None) -> "Box": + frozen = self._box_config["frozen_box"] + config = self.__box_config() + config["frozen_box"] = False + out = self._box_config["box_class"](**config) + memodict = memodict or {} + memodict[id(self)] = out + for k, v in self.items(): + out[copy.deepcopy(k, memodict)] = copy.deepcopy(v, memodict) + out._box_config["frozen_box"] = frozen + return out + + def __setstate__(self, state): + self._box_config = state["_box_config"] + self.__dict__.update(state) + + def __get_default(self, item, attr=False): + if item in ("getdoc", "shape") and _is_ipython(): + return None + default_value = self._box_config["default_box_attr"] + if default_value in (self._box_config["box_class"], dict): + value = self._box_config["box_class"](**self.__box_config(extra_namespace=item)) + elif isinstance(default_value, dict): + value = self._box_config["box_class"](**self.__box_config(extra_namespace=item), **default_value) + elif isinstance(default_value, list): + value = box.BoxList(**self.__box_config(extra_namespace=item)) + elif isinstance(default_value, Callable): + args = [] + kwargs = {} + p_sigs = [ + p.name + for p in signature(default_value).parameters.values() + if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD) + ] + k_sigs = [p.name for p in signature(default_value).parameters.values() if p.kind is p.KEYWORD_ONLY] + for name in p_sigs: + if name not in ("key", "box_instance"): + raise BoxError("default_box_attr can only have the arguments 'key' and 'box_instance'") + if "key" in p_sigs: + args.append(item) + if "box_instance" in p_sigs: + args.insert(p_sigs.index("box_instance"), self) + if "key" in k_sigs: + kwargs["key"] = item + if "box_instance" in k_sigs: + kwargs["box_instance"] = self + value = default_value(*args, **kwargs) + elif hasattr(default_value, "copy"): + value = default_value.copy() + else: + value = default_value + if self._box_config["default_box_create_on_get"]: + if not attr or not (item.startswith("_") and item.endswith("_")): + if self._box_config["box_dots"] and isinstance(item, str) and ("." in item or "[" in item): + first_item, children = _parse_box_dots(self, item, setting=True) + if first_item in self.keys(): + if hasattr(self[first_item], "__setitem__"): + self[first_item].__setitem__(children, value) + else: + super().__setitem__( + first_item, self._box_config["box_class"](**self.__box_config(extra_namespace=first_item)) + ) + self[first_item].__setitem__(children, value) + else: + super().__setitem__(item, value) + return value + + def __box_config(self, extra_namespace: Any = NO_NAMESPACE) -> Dict: + out = {} + for k, v in self._box_config.copy().items(): + if not k.startswith("__"): + out[k] = v + if extra_namespace is not NO_NAMESPACE and self._box_config["box_namespace"] is not False: + out["box_namespace"] = (*out["box_namespace"], extra_namespace) + return out + + def __recast(self, item, value): + if self._box_config["box_recast"] and item in self._box_config["box_recast"]: + recast = self._box_config["box_recast"][item] + try: + if isinstance(recast, type) and issubclass(recast, (Box, box.BoxList)): + return recast(value, **self.__box_config()) + else: + return recast(value) + except ValueError as err: + raise BoxValueError(f"Cannot convert {value} to {recast}") from _exception_cause(err) + return value + + def __convert_and_store(self, item, value): + if self._box_config["conversion_box"]: + safe_key = self._safe_attr(item) + self._box_config["__safe_keys"][safe_key] = item + if isinstance(value, (int, float, str, bytes, bytearray, bool, complex, set, frozenset)): + return super().__setitem__(item, value) + # If the value has already been converted or should not be converted, return it as-is + if self._box_config["box_intact_types"] and isinstance(value, self._box_config["box_intact_types"]): + return super().__setitem__(item, value) + # This is the magic sauce that makes sub dictionaries into new box objects + if isinstance(value, dict): + # We always re-create even if it was already a Box object to pass down configurations correctly + value = self._box_config["box_class"](value, **self.__box_config(extra_namespace=item)) + elif isinstance(value, list) and not isinstance(value, box.BoxList): + if self._box_config["frozen_box"]: + value = _recursive_tuples( + value, + recreate_tuples=self._box_config["modify_tuples_box"], + **self.__box_config(extra_namespace=item), + ) + else: + value = box.BoxList(value, **self.__box_config(extra_namespace=item)) + elif isinstance(value, box.BoxList): + value.box_options.update(self.__box_config(extra_namespace=item)) + elif self._box_config["modify_tuples_box"] and isinstance(value, tuple): + value = _recursive_tuples(value, recreate_tuples=True, **self.__box_config(extra_namespace=item)) + super().__setitem__(item, value) + + def __getitem__(self, item, _ignore_default=False): + try: + return super().__getitem__(item) + except KeyError as err: + if item == "_box_config": + cause = _exception_cause(err) + raise BoxKeyError("_box_config should only exist as an attribute and is never defaulted") from cause + if isinstance(item, slice): + # In Python 3.12 this changes to a KeyError instead of TypeError + new_box = self._box_config["box_class"](**self.__box_config()) + for x in list(super().keys())[item.start : item.stop : item.step]: + new_box[x] = self[x] + return new_box + if self._box_config["box_dots"] and isinstance(item, str) and ("." in item or "[" in item): + try: + first_item, children = _parse_box_dots(self, item) + except BoxError: + if self._box_config["default_box"] and not _ignore_default: + return self.__get_default(item) + raise BoxKeyError(str(item)) from _exception_cause(err) + if first_item in self.keys(): + if hasattr(self[first_item], "__getitem__"): + return self[first_item][children] + if self._box_config["camel_killer_box"] and isinstance(item, str): + converted = _camel_killer(item) + if converted in self.keys(): + return super().__getitem__(converted) + if self._box_config["default_box"] and not _ignore_default: + return self.__get_default(item) + raise BoxKeyError(str(err)) from _exception_cause(err) + except TypeError as err: + if isinstance(item, slice): + new_box = self._box_config["box_class"](**self.__box_config()) + for x in list(super().keys())[item.start : item.stop : item.step]: + new_box[x] = self[x] + return new_box + raise BoxTypeError(str(err)) from _exception_cause(err) + + def __getattr__(self, item): + try: + try: + value = self.__getitem__(item, _ignore_default=True) + except KeyError: + value = object.__getattribute__(self, item) + except AttributeError as err: + if item == "__getstate__": + raise BoxKeyError(item) from _exception_cause(err) + if item == "_box_config": + raise BoxError("_box_config key must exist") from _exception_cause(err) + if self._box_config["conversion_box"]: + safe_key = self._safe_attr(item) + if safe_key in self._box_config["__safe_keys"]: + return self.__getitem__(self._box_config["__safe_keys"][safe_key]) + if self._box_config["default_box"]: + if item.startswith("_") and item.endswith("_"): + raise BoxKeyError(f"{item}: Does not exist and internal methods are never defaulted") + return self.__get_default(item, attr=True) + raise BoxKeyError(str(err)) from _exception_cause(err) + return value + + def __setitem__(self, key, value): + if key != "_box_config" and self._box_config["frozen_box"] and self._box_config["__created"]: + raise BoxError("Box is frozen") + if self._box_config["box_dots"] and isinstance(key, str) and ("." in key or "[" in key): + first_item, children = _parse_box_dots(self, key, setting=True) + if first_item in self.keys(): + if hasattr(self[first_item], "__setitem__"): + return self[first_item].__setitem__(children, value) + elif self._box_config["default_box"]: + if children[0] == "[": + super().__setitem__(first_item, box.BoxList(**self.__box_config(extra_namespace=first_item))) + else: + super().__setitem__( + first_item, self._box_config["box_class"](**self.__box_config(extra_namespace=first_item)) + ) + return self[first_item].__setitem__(children, value) + else: + raise BoxKeyError(f"'{self.__class__}' object has no attribute {first_item}") + value = self.__recast(key, value) + if key not in self.keys() and self._box_config["camel_killer_box"]: + if self._box_config["camel_killer_box"] and isinstance(key, str): + key = _camel_killer(key) + if self._box_config["conversion_box"] and self._box_config["box_duplicates"] != "ignore": + self._conversion_checks(key) + self.__convert_and_store(key, value) + + def __setattr__(self, key, value): + if key == "_box_config": + return object.__setattr__(self, key, value) + if self._box_config["frozen_box"] and self._box_config["__created"]: + raise BoxError("Box is frozen") + if key in self._protected_keys: + raise BoxKeyError(f'Key name "{key}" is protected') + + safe_key = self._safe_attr(key) + if safe_key in self._box_config["__safe_keys"]: + key = self._box_config["__safe_keys"][safe_key] + + # if user has customized property setter, fall back to default implementation + if _get_property_func(self, key)[1] is not None: + super().__setattr__(key, value) + else: + self.__setitem__(key, value) + + def __delitem__(self, key): + if self._box_config["frozen_box"]: + raise BoxError("Box is frozen") + if ( + key not in self.keys() + and self._box_config["box_dots"] + and isinstance(key, str) + and ("." in key or "[" in key) + ): + try: + first_item, children = _parse_box_dots(self, key) + except BoxError: + raise BoxKeyError(str(key)) from None + if hasattr(self[first_item], "__delitem__"): + return self[first_item].__delitem__(children) + if key not in self.keys() and self._box_config["camel_killer_box"]: + if self._box_config["camel_killer_box"] and isinstance(key, str): + for each_key in self: + if _camel_killer(key) == each_key: + key = each_key + break + try: + super().__delitem__(key) + except KeyError as err: + raise BoxKeyError(str(err)) from _exception_cause(err) + + def __delattr__(self, item): + if self._box_config["frozen_box"]: + raise BoxError("Box is frozen") + if item == "_box_config": + raise BoxError('"_box_config" is protected') + if item in self._protected_keys: + raise BoxKeyError(f'Key name "{item}" is protected') + + property_fdel = _get_property_func(self, item)[2] + + # if user has customized property deleter, route to it + if property_fdel is not None: + property_fdel(self) + return + try: + self.__delitem__(item) + except KeyError as err: + if self._box_config["conversion_box"]: + safe_key = self._safe_attr(item) + if safe_key in self._box_config["__safe_keys"]: + self.__delitem__(self._box_config["__safe_keys"][safe_key]) + del self._box_config["__safe_keys"][safe_key] + return + raise BoxKeyError(str(err)) from _exception_cause(err) + + def pop(self, key, *args): + if self._box_config["frozen_box"]: + raise BoxError("Box is frozen") + + if args: + if len(args) != 1: + raise BoxError('pop() takes only one optional argument "default"') + try: + item = self[key] + except KeyError: + return args[0] + else: + del self[key] + return item + try: + item = self[key] + except KeyError: + raise BoxKeyError(f"{key}") from None + else: + del self[key] + return item + + def clear(self): + if self._box_config["frozen_box"]: + raise BoxError("Box is frozen") + super().clear() + self._box_config["__safe_keys"].clear() + + def popitem(self): + if self._box_config["frozen_box"]: + raise BoxError("Box is frozen") + try: + key = next(self.__iter__()) + except StopIteration: + raise BoxKeyError("Empty box") from None + return key, self.pop(key) + + def __repr__(self) -> str: + return f"{self.__class__.__name__}({self})" + + def __str__(self) -> str: + return str(self.to_dict()) + + def __iter__(self) -> Generator: + for key in self.keys(): + yield key + + def __reversed__(self) -> Generator: + for key in reversed(list(self.keys())): + yield key + + def to_dict(self) -> Dict: + """ + Turn the Box and sub Boxes back into a native python dictionary. + + :return: python dictionary of this Box + """ + out_dict = dict(self) + for k, v in out_dict.items(): + if v is self: + out_dict[k] = out_dict + elif isinstance(v, Box): + out_dict[k] = v.to_dict() + elif isinstance(v, box.BoxList): + out_dict[k] = v.to_list() + return out_dict + + def update(self, *args, **kwargs): + if self._box_config["frozen_box"]: + raise BoxError("Box is frozen") + if (len(args) + int(bool(kwargs))) > 1: + raise BoxTypeError(f"update expected at most 1 argument, got {len(args) + int(bool(kwargs))}") + single_arg = next(iter(args), None) + if single_arg: + if hasattr(single_arg, "keys"): + for k in single_arg: + self.__convert_and_store(k, single_arg[k]) + else: + for k, v in single_arg: + self.__convert_and_store(k, v) + for k in kwargs: + self.__convert_and_store(k, kwargs[k]) + + def merge_update(self, *args, **kwargs): + merge_type = None + if "box_merge_lists" in kwargs: + merge_type = kwargs.pop("box_merge_lists") + + def convert_and_set(k, v): + intact_type = self._box_config["box_intact_types"] and isinstance(v, self._box_config["box_intact_types"]) + if isinstance(v, dict) and not intact_type: + # Box objects must be created in case they are already + # in the `converted` box_config set + v = self._box_config["box_class"](v, **self.__box_config(extra_namespace=k)) + if k in self and isinstance(self[k], dict): + self[k].merge_update(v, box_merge_lists=merge_type) + return + if isinstance(v, list) and not intact_type: + v = box.BoxList(v, **self.__box_config(extra_namespace=k)) + if merge_type == "extend" and k in self and isinstance(self[k], list): + self[k].extend(v) + return + if merge_type == "unique" and k in self and isinstance(self[k], list): + for item in v: + if item not in self[k]: + self[k].append(item) + return + self.__setitem__(k, v) + + if (len(args) + int(bool(kwargs))) > 1: + raise BoxTypeError(f"merge_update expected at most 1 argument, got {len(args) + int(bool(kwargs))}") + single_arg = next(iter(args), None) + if single_arg: + if hasattr(single_arg, "keys"): + for k in single_arg: + convert_and_set(k, single_arg[k]) + else: + for k, v in single_arg: + convert_and_set(k, v) + + for key in kwargs: + convert_and_set(key, kwargs[key]) + + def setdefault(self, item, default=None): + if item in self: + return self[item] + + if self._box_config["box_dots"]: + if item in _get_dot_paths(self): + return self[item] + + if isinstance(default, dict): + default = self._box_config["box_class"](default, **self.__box_config(extra_namespace=item)) + if isinstance(default, list): + default = box.BoxList(default, **self.__box_config(extra_namespace=item)) + self[item] = default + return self[item] + + def _safe_attr(self, attr): + """Convert a key into something that is accessible as an attribute""" + if isinstance(attr, str): + # By assuming most people are using string first we get substantial speed ups + if attr.isidentifier() and not iskeyword(attr): + return attr + + if isinstance(attr, tuple): + attr = "_".join([str(x) for x in attr]) + + attr = attr.decode("utf-8", "ignore") if isinstance(attr, bytes) else str(attr) + if self.__box_config()["camel_killer_box"]: + attr = _camel_killer(attr) + + if attr.isidentifier() and not iskeyword(attr): + return attr + + if sum(1 for character in attr if character.isidentifier() and not iskeyword(character)) == 0: + attr = f'{self.__box_config()["box_safe_prefix"]}{attr}' + if attr.isidentifier() and not iskeyword(attr): + return attr + + out = [] + last_safe = 0 + for i, character in enumerate(attr): + if f"x{character}".isidentifier(): + last_safe = i + out.append(character) + elif not out: + continue + else: + if last_safe == i - 1: + out.append("_") + + out = "".join(out)[: last_safe + 1] + + try: + int(out[0]) + except (ValueError, IndexError): + pass + else: + out = f'{self.__box_config()["box_safe_prefix"]}{out}' + + if iskeyword(out): + out = f'{self.__box_config()["box_safe_prefix"]}{out}' + + return out + + def _conversion_checks(self, item): + """ + Internal use for checking if a duplicate safe attribute already exists + + :param item: Item to see if a dup exists + """ + safe_item = self._safe_attr(item) + + if safe_item in self._box_config["__safe_keys"]: + dups = [f"{item}({safe_item})", f'{self._box_config["__safe_keys"][safe_item]}({safe_item})'] + if self._box_config["box_duplicates"].startswith("warn"): + warnings.warn(f"Duplicate conversion attributes exist: {dups}", BoxWarning) + else: + raise BoxError(f"Duplicate conversion attributes exist: {dups}") + + def to_json( + self, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + **json_kwargs, + ): + """ + Transform the Box object into a JSON string. + + :param filename: If provided will save to file + :param encoding: File encoding + :param errors: How to handle encoding errors + :param json_kwargs: additional arguments to pass to json.dump(s) + :return: string of JSON (if no filename provided) + """ + return _to_json(self.to_dict(), filename=filename, encoding=encoding, errors=errors, **json_kwargs) + + @classmethod + def from_json( + cls, + json_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + **kwargs, + ) -> "Box": + """ + Transform a json object string into a Box object. If the incoming + json is a list, you must use BoxList.from_json. + + :param json_string: string to pass to `json.loads` + :param filename: filename to open and pass to `json.load` + :param encoding: File encoding + :param errors: How to handle encoding errors + :param kwargs: parameters to pass to `Box()` or `json.loads` + :return: Box object from json data + """ + box_args = {} + for arg in kwargs.copy(): + if arg in BOX_PARAMETERS: + box_args[arg] = kwargs.pop(arg) + + data = _from_json(json_string, filename=filename, encoding=encoding, errors=errors, **kwargs) + + if not isinstance(data, dict): + raise BoxError(f"json data not returned as a dictionary, but rather a {type(data).__name__}") + return cls(data, **box_args) + + if yaml_available: + + def to_yaml( + self, + filename: Optional[Union[str, PathLike]] = None, + default_flow_style: bool = False, + encoding: str = "utf-8", + errors: str = "strict", + **yaml_kwargs, + ): + """ + Transform the Box object into a YAML string. + + :param filename: If provided will save to file + :param default_flow_style: False will recursively dump dicts + :param encoding: File encoding + :param errors: How to handle encoding errors + :param yaml_kwargs: additional arguments to pass to yaml.dump + :return: string of YAML (if no filename provided) + """ + return _to_yaml( + self.to_dict(), + filename=filename, + default_flow_style=default_flow_style, + encoding=encoding, + errors=errors, + **yaml_kwargs, + ) + + @classmethod + def from_yaml( + cls, + yaml_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + **kwargs, + ) -> "Box": + """ + Transform a yaml object string into a Box object. By default will use SafeLoader. + + :param yaml_string: string to pass to `yaml.load` + :param filename: filename to open and pass to `yaml.load` + :param encoding: File encoding + :param errors: How to handle encoding errors + :param kwargs: parameters to pass to `Box()` or `yaml.load` + :return: Box object from yaml data + """ + box_args = {} + for arg in kwargs.copy(): + if arg in BOX_PARAMETERS: + box_args[arg] = kwargs.pop(arg) + + data = _from_yaml(yaml_string=yaml_string, filename=filename, encoding=encoding, errors=errors, **kwargs) + if not data: + return cls(**box_args) + if not isinstance(data, dict): + raise BoxError(f"yaml data not returned as a dictionary but rather a {type(data).__name__}") + return cls(data, **box_args) + + else: + + def to_yaml( + self, + filename: Optional[Union[str, PathLike]] = None, + default_flow_style: bool = False, + encoding: str = "utf-8", + errors: str = "strict", + **yaml_kwargs, + ): + raise BoxError('yaml is unavailable on this system, please install the "ruamel.yaml" or "PyYAML" package') + + @classmethod + def from_yaml( + cls, + yaml_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + **kwargs, + ) -> "Box": + raise BoxError('yaml is unavailable on this system, please install the "ruamel.yaml" or "PyYAML" package') + + if toml_write_library is not None: + + def to_toml( + self, filename: Optional[Union[str, PathLike]] = None, encoding: str = "utf-8", errors: str = "strict" + ): + """ + Transform the Box object into a toml string. + + :param filename: File to write toml object too + :param encoding: File encoding + :param errors: How to handle encoding errors + :return: string of TOML (if no filename provided) + """ + return _to_toml(self.to_dict(), filename=filename, encoding=encoding, errors=errors) + + else: + + def to_toml( + self, filename: Optional[Union[str, PathLike]] = None, encoding: str = "utf-8", errors: str = "strict" + ): + raise BoxError('toml is unavailable on this system, please install the "tomli-w" package') + + if toml_read_library is not None: + + @classmethod + def from_toml( + cls, + toml_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + **kwargs, + ) -> "Box": + """ + Transforms a toml string or file into a Box object + + :param toml_string: string to pass to `toml.load` + :param filename: filename to open and pass to `toml.load` + :param encoding: File encoding + :param errors: How to handle encoding errors + :param kwargs: parameters to pass to `Box()` + :return: Box object + """ + box_args = {} + for arg in kwargs.copy(): + if arg in BOX_PARAMETERS: + box_args[arg] = kwargs.pop(arg) + + data = _from_toml(toml_string=toml_string, filename=filename, encoding=encoding, errors=errors) + return cls(data, **box_args) + + else: + + @classmethod + def from_toml( + cls, + toml_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + **kwargs, + ) -> "Box": + raise BoxError('toml is unavailable on this system, please install the "tomli" package') + + if msgpack_available: + + def to_msgpack(self, filename: Optional[Union[str, PathLike]] = None, **kwargs): + """ + Transform the Box object into a msgpack string. + + :param filename: File to write msgpack object too + :param kwargs: parameters to pass to `msgpack.pack` + :return: bytes of msgpack (if no filename provided) + """ + return _to_msgpack(self.to_dict(), filename=filename, **kwargs) + + @classmethod + def from_msgpack( + cls, + msgpack_bytes: Optional[bytes] = None, + filename: Optional[Union[str, PathLike]] = None, + **kwargs, + ) -> "Box": + """ + Transforms msgpack bytes or file into a Box object + + :param msgpack_bytes: string to pass to `msgpack.unpackb` + :param filename: filename to open and pass to `msgpack.unpack` + :param kwargs: parameters to pass to `Box()` + :return: Box object + """ + box_args = {} + for arg in kwargs.copy(): + if arg in BOX_PARAMETERS: + box_args[arg] = kwargs.pop(arg) + + data = _from_msgpack(msgpack_bytes=msgpack_bytes, filename=filename, **kwargs) + if not isinstance(data, dict): + raise BoxError(f"msgpack data not returned as a dictionary but rather a {type(data).__name__}") + return cls(data, **box_args) + + else: + + def to_msgpack(self, filename: Optional[Union[str, PathLike]] = None, **kwargs): + raise BoxError('msgpack is unavailable on this system, please install the "msgpack" package') + + @classmethod + def from_msgpack( + cls, + msgpack_bytes: Optional[bytes] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + **kwargs, + ) -> "Box": + raise BoxError('msgpack is unavailable on this system, please install the "msgpack" package') diff --git a/box/box.pyi b/box/box.pyi new file mode 100644 index 0000000..9d56ab3 --- /dev/null +++ b/box/box.pyi @@ -0,0 +1,121 @@ +from _typeshed import Incomplete +from collections.abc import Mapping +from os import PathLike +from typing import Any, Dict, Generator, List, Optional, Tuple, Type, Union, Literal + +class Box(dict): + def __new__( + cls, + *args: Any, + default_box: bool = ..., + default_box_attr: Any = ..., + default_box_none_transform: bool = ..., + default_box_create_on_get: bool = ..., + frozen_box: bool = ..., + camel_killer_box: bool = ..., + conversion_box: bool = ..., + modify_tuples_box: bool = ..., + box_safe_prefix: str = ..., + box_duplicates: str = ..., + box_intact_types: Union[Tuple, List] = ..., + box_recast: Optional[Dict] = ..., + box_dots: bool = ..., + box_class: Optional[Union[Dict, Type["Box"]]] = ..., + box_namespace: Union[Tuple[str, ...], Literal[False]] = ..., + **kwargs: Any, + ): ... + def __init__( + self, + *args: Any, + default_box: bool = ..., + default_box_attr: Any = ..., + default_box_none_transform: bool = ..., + default_box_create_on_get: bool = ..., + frozen_box: bool = ..., + camel_killer_box: bool = ..., + conversion_box: bool = ..., + modify_tuples_box: bool = ..., + box_safe_prefix: str = ..., + box_duplicates: str = ..., + box_intact_types: Union[Tuple, List] = ..., + box_recast: Optional[Dict] = ..., + box_dots: bool = ..., + box_class: Optional[Union[Dict, Type["Box"]]] = ..., + box_namespace: Union[Tuple[str, ...], Literal[False]] = ..., + **kwargs: Any, + ) -> None: ... + def __add__(self, other: Mapping[Any, Any]): ... + def __radd__(self, other: Mapping[Any, Any]): ... + def __iadd__(self, other: Mapping[Any, Any]): ... + def __or__(self, other: Mapping[Any, Any]): ... + def __ror__(self, other: Mapping[Any, Any]): ... + def __ior__(self, other: Mapping[Any, Any]): ... # type: ignore[override] + def __sub__(self, other: Mapping[Any, Any]): ... + def __hash__(self): ... + def __dir__(self) -> List[str]: ... + def __contains__(self, item) -> bool: ... + def keys(self, dotted: Union[bool] = ...): ... + def items(self, dotted: Union[bool] = ...): ... + def get(self, key, default=...): ... + def copy(self) -> Box: ... + def __copy__(self) -> Box: ... + def __deepcopy__(self, memodict: Incomplete | None = ...) -> Box: ... + def __getitem__(self, item, _ignore_default: bool = ...): ... + def __getattr__(self, item): ... + def __setitem__(self, key, value): ... + def __setattr__(self, key, value): ... + def __delitem__(self, key): ... + def __delattr__(self, item) -> None: ... + def pop(self, key, *args): ... + def clear(self) -> None: ... + def popitem(self): ... + def __iter__(self) -> Generator: ... + def __reversed__(self) -> Generator: ... + def to_dict(self) -> Dict: ... + def update(self, *args, **kwargs) -> None: ... + def merge_update(self, *args, **kwargs) -> None: ... + def setdefault(self, item, default: Incomplete | None = ...): ... + def to_json( + self, filename: Optional[Union[str, PathLike]] = ..., encoding: str = ..., errors: str = ..., **json_kwargs + ): ... + @classmethod + def from_json( + cls, + json_string: Optional[str] = ..., + filename: Optional[Union[str, PathLike]] = ..., + encoding: str = ..., + errors: str = ..., + **kwargs, + ) -> Box: ... + def to_yaml( + self, + filename: Optional[Union[str, PathLike]] = ..., + default_flow_style: bool = ..., + encoding: str = ..., + errors: str = ..., + **yaml_kwargs, + ): ... + @classmethod + def from_yaml( + cls, + yaml_string: Optional[str] = ..., + filename: Optional[Union[str, PathLike]] = ..., + encoding: str = ..., + errors: str = ..., + **kwargs, + ) -> Box: ... + def to_toml(self, filename: Optional[Union[str, PathLike]] = ..., encoding: str = ..., errors: str = ...): ... + @classmethod + def from_toml( + cls, + toml_string: Optional[str] = ..., + filename: Optional[Union[str, PathLike]] = ..., + encoding: str = ..., + errors: str = ..., + **kwargs, + ) -> Box: ... + def to_msgpack(self, filename: Optional[Union[str, PathLike]] = ..., **kwargs): ... + @classmethod + def from_msgpack( + cls, msgpack_bytes: Optional[bytes] = ..., filename: Optional[Union[str, PathLike]] = ..., **kwargs + ) -> Box: ... diff --git a/box/box_list.py b/box/box_list.py new file mode 100644 index 0000000..fb00864 --- /dev/null +++ b/box/box_list.py @@ -0,0 +1,476 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# +# Copyright (c) 2017-2023 - Chris Griffith - MIT License +import copy +import re +from os import PathLike +from typing import Optional, Iterable, Type, Union, List, Any + +import box +from box.converters import ( + BOX_PARAMETERS, + _from_csv, + _from_json, + _from_msgpack, + _from_toml, + _from_yaml, + _to_csv, + _to_json, + _to_msgpack, + _to_toml, + _to_yaml, + msgpack_available, + toml_read_library, + yaml_available, +) +from box.exceptions import BoxError, BoxTypeError + +_list_pos_re = re.compile(r"\[(\d+)\]") + + +class BoxList(list): + """ + Drop in replacement of list, that converts added objects to Box or BoxList + objects as necessary. + """ + + def __new__(cls, *args, **kwargs): + obj = super().__new__(cls, *args, **kwargs) + # This is required for pickling to work correctly + obj.box_options = {"box_class": box.Box} + obj.box_options.update(kwargs) + obj.box_org_ref = None + return obj + + def __init__(self, iterable: Optional[Iterable] = None, box_class: Type[box.Box] = box.Box, **box_options): + self.box_options = box_options + self.box_options["box_class"] = box_class + self.box_org_ref = iterable + if iterable: + for x in iterable: + self.append(x) + self.box_org_ref = None + if box_options.get("frozen_box"): + + def frozen(*args, **kwargs): + raise BoxError("BoxList is frozen") + + for method in ["append", "extend", "insert", "pop", "remove", "reverse", "sort"]: + self.__setattr__(method, frozen) + + def __getitem__(self, item): + if self.box_options.get("box_dots") and isinstance(item, str) and item.startswith("["): + list_pos = _list_pos_re.search(item) + value = super().__getitem__(int(list_pos.groups()[0])) + if len(list_pos.group()) == len(item): + return value + return value.__getitem__(item[len(list_pos.group()) :].lstrip(".")) + if isinstance(item, tuple): + result = self + for idx in item: + if isinstance(result, list): + result = result[idx] + else: + raise BoxTypeError(f"Cannot numpy-style indexing on {type(result).__name__}.") + return result + return super().__getitem__(item) + + def __delitem__(self, key): + if self.box_options.get("frozen_box"): + raise BoxError("BoxList is frozen") + if self.box_options.get("box_dots") and isinstance(key, str) and key.startswith("["): + list_pos = _list_pos_re.search(key) + pos = int(list_pos.groups()[0]) + if len(list_pos.group()) == len(key): + return super().__delitem__(pos) + if hasattr(self[pos], "__delitem__"): + return self[pos].__delitem__(key[len(list_pos.group()) :].lstrip(".")) # type: ignore + super().__delitem__(key) + + def __setitem__(self, key, value): + if self.box_options.get("frozen_box"): + raise BoxError("BoxList is frozen") + if self.box_options.get("box_dots") and isinstance(key, str) and key.startswith("["): + list_pos = _list_pos_re.search(key) + pos = int(list_pos.groups()[0]) + if pos >= len(self) and self.box_options.get("default_box"): + self.extend([None] * (pos - len(self) + 1)) + if len(list_pos.group()) == len(key): + return super().__setitem__(pos, value) + children = key[len(list_pos.group()):].lstrip(".") + if self.box_options.get("default_box"): + if children[0] == "[": + super().__setitem__(pos, box.BoxList(**self.box_options)) + else: + super().__setitem__(pos, self.box_options.get("box_class")(**self.box_options)) + return super().__getitem__(pos).__setitem__(children, value) + super().__setitem__(key, value) + + def _is_intact_type(self, obj): + if self.box_options.get("box_intact_types") and isinstance(obj, self.box_options["box_intact_types"]): + return True + return False + + def _convert(self, p_object): + if isinstance(p_object, dict) and not self._is_intact_type(p_object): + p_object = self.box_options["box_class"](p_object, **self.box_options) + elif isinstance(p_object, box.Box): + p_object._box_config.update(self.box_options) + if isinstance(p_object, list) and not self._is_intact_type(p_object): + p_object = ( + self + if p_object is self or p_object is self.box_org_ref + else self.__class__(p_object, **self.box_options) + ) + elif isinstance(p_object, BoxList): + p_object.box_options.update(self.box_options) + return p_object + + def append(self, p_object): + super().append(self._convert(p_object)) + + def extend(self, iterable): + for item in iterable: + self.append(item) + + def insert(self, index, p_object): + super().insert(index, self._convert(p_object)) + + def _dotted_helper(self) -> List[str]: + keys = [] + for idx, item in enumerate(self): + added = False + if isinstance(item, box.Box): + for key in item.keys(dotted=True): + keys.append(f"[{idx}].{key}") + added = True + elif isinstance(item, BoxList): + for key in item._dotted_helper(): + keys.append(f"[{idx}]{key}") + added = True + if not added: + keys.append(f"[{idx}]") + return keys + + def __repr__(self): + return f"{self.__class__.__name__}({self.to_list()})" + + def __str__(self): + return str(self.to_list()) + + def __copy__(self): + return self.__class__((x for x in self), **self.box_options) + + def __deepcopy__(self, memo=None): + out = self.__class__() + memo = memo or {} + memo[id(self)] = out + for k in self: + out.append(copy.deepcopy(k, memo=memo)) + return out + + def __hash__(self) -> int: # type: ignore[override] + if self.box_options.get("frozen_box"): + hashing = 98765 + hashing ^= hash(tuple(self)) + return hashing + raise BoxTypeError("unhashable type: 'BoxList'") + + def to_list(self) -> List: + new_list: List[Any] = [] + for x in self: + if x is self: + new_list.append(new_list) + elif isinstance(x, box.Box): + new_list.append(x.to_dict()) + elif isinstance(x, BoxList): + new_list.append(x.to_list()) + else: + new_list.append(x) + return new_list + + def to_json( + self, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + multiline: bool = False, + **json_kwargs, + ): + """ + Transform the BoxList object into a JSON string. + + :param filename: If provided will save to file + :param encoding: File encoding + :param errors: How to handle encoding errors + :param multiline: Put each item in list onto it's own line + :param json_kwargs: additional arguments to pass to json.dump(s) + :return: string of JSON or return of `json.dump` + """ + if filename and multiline: + lines = [_to_json(item, filename=None, encoding=encoding, errors=errors, **json_kwargs) for item in self] + with open(filename, "w", encoding=encoding, errors=errors) as f: + f.write("\n".join(lines)) + else: + return _to_json(self.to_list(), filename=filename, encoding=encoding, errors=errors, **json_kwargs) + + @classmethod + def from_json( + cls, + json_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + multiline: bool = False, + **kwargs, + ): + """ + Transform a json object string into a BoxList object. If the incoming + json is a dict, you must use Box.from_json. + + :param json_string: string to pass to `json.loads` + :param filename: filename to open and pass to `json.load` + :param encoding: File encoding + :param errors: How to handle encoding errors + :param multiline: One object per line + :param kwargs: parameters to pass to `Box()` or `json.loads` + :return: BoxList object from json data + """ + box_args = {} + for arg in list(kwargs.keys()): + if arg in BOX_PARAMETERS: + box_args[arg] = kwargs.pop(arg) + + data = _from_json( + json_string, filename=filename, encoding=encoding, errors=errors, multiline=multiline, **kwargs + ) + + if not isinstance(data, list): + raise BoxError(f"json data not returned as a list, but rather a {type(data).__name__}") + return cls(data, **box_args) + + if yaml_available: + + def to_yaml( + self, + filename: Optional[Union[str, PathLike]] = None, + default_flow_style: bool = False, + encoding: str = "utf-8", + errors: str = "strict", + **yaml_kwargs, + ): + """ + Transform the BoxList object into a YAML string. + + :param filename: If provided will save to file + :param default_flow_style: False will recursively dump dicts + :param encoding: File encoding + :param errors: How to handle encoding errors + :param yaml_kwargs: additional arguments to pass to yaml.dump + :return: string of YAML or return of `yaml.dump` + """ + return _to_yaml( + self.to_list(), + filename=filename, + default_flow_style=default_flow_style, + encoding=encoding, + errors=errors, + **yaml_kwargs, + ) + + @classmethod + def from_yaml( + cls, + yaml_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + **kwargs, + ): + """ + Transform a yaml object string into a BoxList object. + + :param yaml_string: string to pass to `yaml.load` + :param filename: filename to open and pass to `yaml.load` + :param encoding: File encoding + :param errors: How to handle encoding errors + :param kwargs: parameters to pass to `BoxList()` or `yaml.load` + :return: BoxList object from yaml data + """ + box_args = {} + for arg in list(kwargs.keys()): + if arg in BOX_PARAMETERS: + box_args[arg] = kwargs.pop(arg) + + data = _from_yaml(yaml_string=yaml_string, filename=filename, encoding=encoding, errors=errors, **kwargs) + if not data: + return cls(**box_args) + if not isinstance(data, list): + raise BoxError(f"yaml data not returned as a list but rather a {type(data).__name__}") + return cls(data, **box_args) + + else: + + def to_yaml( + self, + filename: Optional[Union[str, PathLike]] = None, + default_flow_style: bool = False, + encoding: str = "utf-8", + errors: str = "strict", + **yaml_kwargs, + ): + raise BoxError('yaml is unavailable on this system, please install the "ruamel.yaml" or "PyYAML" package') + + @classmethod + def from_yaml( + cls, + yaml_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + **kwargs, + ): + raise BoxError('yaml is unavailable on this system, please install the "ruamel.yaml" or "PyYAML" package') + + if toml_read_library is not None: + + def to_toml( + self, + filename: Optional[Union[str, PathLike]] = None, + key_name: str = "toml", + encoding: str = "utf-8", + errors: str = "strict", + ): + """ + Transform the BoxList object into a toml string. + + :param filename: File to write toml object too + :param key_name: Specify the name of the key to store the string under + (cannot directly convert to toml) + :param encoding: File encoding + :param errors: How to handle encoding errors + :return: string of TOML (if no filename provided) + """ + return _to_toml({key_name: self.to_list()}, filename=filename, encoding=encoding, errors=errors) + + else: + + def to_toml( + self, + filename: Optional[Union[str, PathLike]] = None, + key_name: str = "toml", + encoding: str = "utf-8", + errors: str = "strict", + ): + raise BoxError('toml is unavailable on this system, please install the "tomli-w" package') + + if toml_read_library is not None: + + @classmethod + def from_toml( + cls, + toml_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + key_name: str = "toml", + encoding: str = "utf-8", + errors: str = "strict", + **kwargs, + ): + """ + Transforms a toml string or file into a BoxList object + + :param toml_string: string to pass to `toml.load` + :param filename: filename to open and pass to `toml.load` + :param key_name: Specify the name of the key to pull the list from + (cannot directly convert from toml) + :param encoding: File encoding + :param errors: How to handle encoding errors + :param kwargs: parameters to pass to `Box()` + :return: + """ + box_args = {} + for arg in list(kwargs.keys()): + if arg in BOX_PARAMETERS: + box_args[arg] = kwargs.pop(arg) + + data = _from_toml(toml_string=toml_string, filename=filename, encoding=encoding, errors=errors) + if key_name not in data: + raise BoxError(f"{key_name} was not found.") + return cls(data[key_name], **box_args) + + else: + + @classmethod + def from_toml( + cls, + toml_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + key_name: str = "toml", + encoding: str = "utf-8", + errors: str = "strict", + **kwargs, + ): + raise BoxError('toml is unavailable on this system, please install the "toml" package') + + if msgpack_available: + + def to_msgpack(self, filename: Optional[Union[str, PathLike]] = None, **kwargs): + """ + Transform the BoxList object into a toml string. + + :param filename: File to write toml object too + :return: string of TOML (if no filename provided) + """ + return _to_msgpack(self.to_list(), filename=filename, **kwargs) + + @classmethod + def from_msgpack( + cls, msgpack_bytes: Optional[bytes] = None, filename: Optional[Union[str, PathLike]] = None, **kwargs + ): + """ + Transforms a toml string or file into a BoxList object + + :param msgpack_bytes: string to pass to `msgpack.packb` + :param filename: filename to open and pass to `msgpack.pack` + :param kwargs: parameters to pass to `Box()` + :return: + """ + box_args = {} + for arg in list(kwargs.keys()): + if arg in BOX_PARAMETERS: + box_args[arg] = kwargs.pop(arg) + + data = _from_msgpack(msgpack_bytes=msgpack_bytes, filename=filename, **kwargs) + if not isinstance(data, list): + raise BoxError(f"msgpack data not returned as a list but rather a {type(data).__name__}") + return cls(data, **box_args) + + else: + + def to_msgpack(self, filename: Optional[Union[str, PathLike]] = None, **kwargs): + raise BoxError('msgpack is unavailable on this system, please install the "msgpack" package') + + @classmethod + def from_msgpack( + cls, + msgpack_bytes: Optional[bytes] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + **kwargs, + ): + raise BoxError('msgpack is unavailable on this system, please install the "msgpack" package') + + def to_csv(self, filename: Optional[Union[str, PathLike]] = None, encoding: str = "utf-8", errors: str = "strict"): + return _to_csv(self, filename=filename, encoding=encoding, errors=errors) + + @classmethod + def from_csv( + cls, + csv_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + ): + return cls(_from_csv(csv_string=csv_string, filename=filename, encoding=encoding, errors=errors)) diff --git a/box/box_list.pyi b/box/box_list.pyi new file mode 100644 index 0000000..982093f --- /dev/null +++ b/box/box_list.pyi @@ -0,0 +1,83 @@ +import box +from box.converters import ( + BOX_PARAMETERS as BOX_PARAMETERS, + msgpack_available as msgpack_available, + toml_read_library as toml_read_library, + toml_write_library as toml_write_library, + yaml_available as yaml_available, +) +from os import PathLike as PathLike +from typing import Any, Iterable, Optional, Type, Union, List + +class BoxList(list): + def __new__(cls, *args: Any, **kwargs: Any): ... + box_options: Any + box_org_ref: Any + def __init__(self, iterable: Iterable = ..., box_class: Type[box.Box] = ..., **box_options: Any) -> None: ... + def __getitem__(self, item: Any): ... + def __delitem__(self, key: Any): ... + def __setitem__(self, key: Any, value: Any): ... + def append(self, p_object: Any) -> None: ... + def extend(self, iterable: Any) -> None: ... + def insert(self, index: Any, p_object: Any) -> None: ... + def __copy__(self) -> "BoxList": ... + def __deepcopy__(self, memo: Optional[Any] = ...) -> "BoxList": ... + def __hash__(self) -> int: ... # type: ignore[override] + def to_list(self) -> List: ... + def _dotted_helper(self) -> List[str]: ... + def to_json( + self, + filename: Union[str, PathLike] = ..., + encoding: str = ..., + errors: str = ..., + multiline: bool = ..., + **json_kwargs: Any, + ) -> Any: ... + @classmethod + def from_json( + cls, + json_string: str = ..., + filename: Union[str, PathLike] = ..., + encoding: str = ..., + errors: str = ..., + multiline: bool = ..., + **kwargs: Any, + ) -> Any: ... + def to_yaml( + self, + filename: Union[str, PathLike] = ..., + default_flow_style: bool = ..., + encoding: str = ..., + errors: str = ..., + **yaml_kwargs: Any, + ) -> Any: ... + @classmethod + def from_yaml( + cls, + yaml_string: str = ..., + filename: Union[str, PathLike] = ..., + encoding: str = ..., + errors: str = ..., + **kwargs: Any, + ) -> Any: ... + def to_toml( + self, filename: Union[str, PathLike] = ..., key_name: str = ..., encoding: str = ..., errors: str = ... + ) -> Any: ... + @classmethod + def from_toml( + cls, + toml_string: str = ..., + filename: Union[str, PathLike] = ..., + key_name: str = ..., + encoding: str = ..., + errors: str = ..., + **kwargs: Any, + ) -> Any: ... + def to_msgpack(self, filename: Union[str, PathLike] = ..., **kwargs: Any) -> Any: ... + @classmethod + def from_msgpack(cls, msgpack_bytes: bytes = ..., filename: Union[str, PathLike] = ..., **kwargs: Any) -> Any: ... + def to_csv(self, filename: Union[str, PathLike] = ..., encoding: str = ..., errors: str = ...) -> Any: ... + @classmethod + def from_csv( + cls, csv_string: str = ..., filename: Union[str, PathLike] = ..., encoding: str = ..., errors: str = ... + ) -> Any: ... diff --git a/box/config_box.py b/box/config_box.py new file mode 100644 index 0000000..4c48877 --- /dev/null +++ b/box/config_box.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +from typing import List + +from box.box import Box + + +class ConfigBox(Box): + """ + Modified box object to add object transforms. + + Allows for build in transforms like: + + cns = ConfigBox(my_bool='yes', my_int='5', my_list='5,4,3,3,2') + + cns.bool('my_bool') # True + cns.int('my_int') # 5 + cns.list('my_list', mod=lambda x: int(x)) # [5, 4, 3, 3, 2] + """ + + _protected_keys = dir(Box) + ["bool", "int", "float", "list", "getboolean", "getfloat", "getint"] + + def __getattr__(self, item): + """ + Config file keys are stored in lower case, be a little more + loosey goosey + """ + try: + return super().__getattr__(item) + except AttributeError: + return super().__getattr__(item.lower()) + + def __dir__(self) -> List[str]: + return super().__dir__() + ["bool", "int", "float", "list", "getboolean", "getfloat", "getint"] + + def bool(self, item, default=None): + """ + Return value of key as a boolean + + :param item: key of value to transform + :param default: value to return if item does not exist + :return: approximated bool of value + """ + try: + item = self.__getattr__(item) + except AttributeError as err: + if default is not None: + return default + raise err + + if isinstance(item, (bool, int)): + return bool(item) + + if isinstance(item, str) and item.lower() in ("n", "no", "false", "f", "0"): + return False + + return True if item else False + + def int(self, item, default=None): + """ + Return value of key as an int + + :param item: key of value to transform + :param default: value to return if item does not exist + :return: int of value + """ + try: + item = self.__getattr__(item) + except AttributeError as err: + if default is not None: + return default + raise err + return int(item) + + def float(self, item, default=None): + """ + Return value of key as a float + + :param item: key of value to transform + :param default: value to return if item does not exist + :return: float of value + """ + try: + item = self.__getattr__(item) + except AttributeError as err: + if default is not None: + return default + raise err + return float(item) + + def list(self, item, default=None, spliter: str = ",", strip=True, mod=None): + """ + Return value of key as a list + + :param item: key of value to transform + :param mod: function to map against list + :param default: value to return if item does not exist + :param spliter: character to split str on + :param strip: clean the list with the `strip` + :return: list of items + """ + try: + item = self.__getattr__(item) + except AttributeError as err: + if default is not None: + return default + raise err + if strip: + item = item.lstrip("[").rstrip("]") + out = [x.strip() if strip else x for x in item.split(spliter)] + if mod: + return list(map(mod, out)) + return out + + # loose configparser compatibility + + def getboolean(self, item, default=None): + return self.bool(item, default) + + def getint(self, item, default=None): + return self.int(item, default) + + def getfloat(self, item, default=None): + return self.float(item, default) + + def __repr__(self): + return f"{self.__class__.__name__}({str(self.to_dict())})" + + def copy(self): + return ConfigBox(super().copy()) + + def __copy__(self): + return ConfigBox(super().copy()) diff --git a/box/config_box.pyi b/box/config_box.pyi new file mode 100644 index 0000000..1022f1b --- /dev/null +++ b/box/config_box.pyi @@ -0,0 +1,15 @@ +from box.box import Box as Box +from typing import Any, Optional, List + +class ConfigBox(Box): + def __getattr__(self, item: Any): ... + def __dir__(self) -> List[str]: ... + def bool(self, item: Any, default: Optional[Any] = ...): ... + def int(self, item: Any, default: Optional[Any] = ...): ... + def float(self, item: Any, default: Optional[Any] = ...): ... + def list(self, item: Any, default: Optional[Any] = ..., spliter: str = ..., strip: bool = ..., mod: Optional[Any] = ...): ... # type: ignore + def getboolean(self, item: Any, default: Optional[Any] = ...): ... + def getint(self, item: Any, default: Optional[Any] = ...): ... + def getfloat(self, item: Any, default: Optional[Any] = ...): ... + def copy(self) -> "ConfigBox": ... + def __copy__(self) -> "ConfigBox": ... diff --git a/box/converters.py b/box/converters.py new file mode 100644 index 0000000..80c1ced --- /dev/null +++ b/box/converters.py @@ -0,0 +1,363 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +# Abstract converter functions for use in any Box class + +import csv +import json +from io import StringIO +from os import PathLike +from pathlib import Path +from typing import Union, Optional, Dict, Any, Callable + +from box.exceptions import BoxError + +pyyaml_available = True +ruamel_available = True +msgpack_available = True + +try: + from ruamel.yaml import version_info, YAML +except ImportError: + ruamel_available = False +else: + if version_info[1] < 17: + ruamel_available = False + +try: + import yaml +except ImportError: + pyyaml_available = False + +MISSING_PARSER_ERROR = "No YAML Parser available, please install ruamel.yaml>=0.17 or PyYAML" + +toml_read_library: Optional[Any] = None +toml_write_library: Optional[Any] = None +toml_decode_error: Optional[Callable] = None + +__all__ = [ + "_to_json", + "_to_yaml", + "_to_toml", + "_to_csv", + "_to_msgpack", + "_from_json", + "_from_yaml", + "_from_toml", + "_from_csv", + "_from_msgpack", +] + + +class BoxTomlDecodeError(BoxError): + """Toml Decode Error""" + + +try: + import toml +except ImportError: + pass +else: + toml_read_library = toml + toml_write_library = toml + toml_decode_error = toml.TomlDecodeError + + class BoxTomlDecodeError(BoxError, toml.TomlDecodeError): # type: ignore + """Toml Decode Error""" + + +try: + import tomllib +except ImportError: + pass +else: + toml_read_library = tomllib + toml_decode_error = tomllib.TOMLDecodeError + + class BoxTomlDecodeError(BoxError, tomllib.TOMLDecodeError): # type: ignore + """Toml Decode Error""" + + +try: + import tomli +except ImportError: + pass +else: + toml_read_library = tomli + toml_decode_error = tomli.TOMLDecodeError + + class BoxTomlDecodeError(BoxError, tomli.TOMLDecodeError): # type: ignore + """Toml Decode Error""" + + +try: + import tomli_w +except ImportError: + pass +else: + toml_write_library = tomli_w + + +try: + import msgpack # type: ignore +except ImportError: + msgpack = None # type: ignore + msgpack_available = False + +yaml_available = pyyaml_available or ruamel_available + +BOX_PARAMETERS = ( + "default_box", + "default_box_attr", + "default_box_none_transform", + "default_box_create_on_get", + "frozen_box", + "camel_killer_box", + "conversion_box", + "modify_tuples_box", + "box_safe_prefix", + "box_duplicates", + "box_intact_types", + "box_dots", + "box_recast", + "box_class", + "box_namespace", +) + + +def _exists(filename: Union[str, PathLike], create: bool = False) -> Path: + path = Path(filename) + if create: + try: + path.touch(exist_ok=True) + except OSError as err: + raise BoxError(f"Could not create file {filename} - {err}") + else: + return path + if not path.exists(): + raise BoxError(f'File "{filename}" does not exist') + if not path.is_file(): + raise BoxError(f"{filename} is not a file") + return path + + +def _to_json( + obj, filename: Optional[Union[str, PathLike]] = None, encoding: str = "utf-8", errors: str = "strict", **json_kwargs +): + if filename: + _exists(filename, create=True) + with open(filename, "w", encoding=encoding, errors=errors) as f: + json.dump(obj, f, ensure_ascii=False, **json_kwargs) + else: + return json.dumps(obj, ensure_ascii=False, **json_kwargs) + + +def _from_json( + json_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + multiline: bool = False, + **kwargs, +): + if filename: + with open(filename, "r", encoding=encoding, errors=errors) as f: + if multiline: + data = [ + json.loads(line.strip(), **kwargs) + for line in f + if line.strip() and not line.strip().startswith("#") + ] + else: + data = json.load(f, **kwargs) + elif json_string: + data = json.loads(json_string, **kwargs) + else: + raise BoxError("from_json requires a string or filename") + return data + + +def _to_yaml( + obj, + filename: Optional[Union[str, PathLike]] = None, + default_flow_style: bool = False, + encoding: str = "utf-8", + errors: str = "strict", + ruamel_typ: str = "rt", + ruamel_attrs: Optional[Dict] = None, + **yaml_kwargs, +): + if not ruamel_attrs: + ruamel_attrs = {} + if filename: + _exists(filename, create=True) + with open(filename, "w", encoding=encoding, errors=errors) as f: + if ruamel_available: + yaml_dumper = YAML(typ=ruamel_typ) + yaml_dumper.default_flow_style = default_flow_style + for attr, value in ruamel_attrs.items(): + setattr(yaml_dumper, attr, value) + return yaml_dumper.dump(obj, stream=f, **yaml_kwargs) + elif pyyaml_available: + return yaml.dump(obj, stream=f, default_flow_style=default_flow_style, **yaml_kwargs) + else: + raise BoxError(MISSING_PARSER_ERROR) + + else: + if ruamel_available: + yaml_dumper = YAML(typ=ruamel_typ) + yaml_dumper.default_flow_style = default_flow_style + for attr, value in ruamel_attrs.items(): + setattr(yaml_dumper, attr, value) + with StringIO() as string_stream: + yaml_dumper.dump(obj, stream=string_stream, **yaml_kwargs) + return string_stream.getvalue() + elif pyyaml_available: + return yaml.dump(obj, default_flow_style=default_flow_style, **yaml_kwargs) + else: + raise BoxError(MISSING_PARSER_ERROR) + + +def _from_yaml( + yaml_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + ruamel_typ: str = "rt", + ruamel_attrs: Optional[Dict] = None, + **kwargs, +): + if not ruamel_attrs: + ruamel_attrs = {} + if filename: + _exists(filename) + with open(filename, "r", encoding=encoding, errors=errors) as f: + if ruamel_available: + yaml_loader = YAML(typ=ruamel_typ) + for attr, value in ruamel_attrs.items(): + setattr(yaml_loader, attr, value) + data = yaml_loader.load(stream=f) + elif pyyaml_available: + if "Loader" not in kwargs: + kwargs["Loader"] = yaml.SafeLoader + data = yaml.load(f, **kwargs) + else: + raise BoxError(MISSING_PARSER_ERROR) + elif yaml_string: + if ruamel_available: + yaml_loader = YAML(typ=ruamel_typ) + for attr, value in ruamel_attrs.items(): + setattr(yaml_loader, attr, value) + data = yaml_loader.load(stream=yaml_string) + elif pyyaml_available: + if "Loader" not in kwargs: + kwargs["Loader"] = yaml.SafeLoader + data = yaml.load(yaml_string, **kwargs) + else: + raise BoxError(MISSING_PARSER_ERROR) + else: + raise BoxError("from_yaml requires a string or filename") + return data + + +def _to_toml(obj, filename: Optional[Union[str, PathLike]] = None, encoding: str = "utf-8", errors: str = "strict"): + if filename: + _exists(filename, create=True) + if toml_write_library.__name__ == "toml": # type: ignore + with open(filename, "w", encoding=encoding, errors=errors) as f: + try: + toml_write_library.dump(obj, f) # type: ignore + except toml_decode_error as err: # type: ignore + raise BoxTomlDecodeError(err) from err + else: + with open(filename, "wb") as f: + try: + toml_write_library.dump(obj, f) # type: ignore + except toml_decode_error as err: # type: ignore + raise BoxTomlDecodeError(err) from err + else: + try: + return toml_write_library.dumps(obj) # type: ignore + except toml_decode_error as err: # type: ignore + raise BoxTomlDecodeError(err) from err + + +def _from_toml( + toml_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", +): + if filename: + _exists(filename) + if toml_read_library.__name__ == "toml": # type: ignore + with open(filename, "r", encoding=encoding, errors=errors) as f: + data = toml_read_library.load(f) # type: ignore + else: + with open(filename, "rb") as f: + data = toml_read_library.load(f) # type: ignore + elif toml_string: + data = toml_read_library.loads(toml_string) # type: ignore + else: + raise BoxError("from_toml requires a string or filename") + return data + + +def _to_msgpack(obj, filename: Optional[Union[str, PathLike]] = None, **kwargs): + if filename: + _exists(filename, create=True) + with open(filename, "wb") as f: + msgpack.pack(obj, f, **kwargs) + else: + return msgpack.packb(obj, **kwargs) + + +def _from_msgpack(msgpack_bytes: Optional[bytes] = None, filename: Optional[Union[str, PathLike]] = None, **kwargs): + if filename: + _exists(filename) + with open(filename, "rb") as f: + data = msgpack.unpack(f, **kwargs) + elif msgpack_bytes: + data = msgpack.unpackb(msgpack_bytes, **kwargs) + else: + raise BoxError("from_msgpack requires a string or filename") + return data + + +def _to_csv( + box_list, filename: Optional[Union[str, PathLike]] = None, encoding: str = "utf-8", errors: str = "strict", **kwargs +): + csv_column_names = list(box_list[0].keys()) + for row in box_list: + if list(row.keys()) != csv_column_names: + raise BoxError("BoxList must contain the same dictionary structure for every item to convert to csv") + + if filename: + _exists(filename, create=True) + out_data = open(filename, "w", encoding=encoding, errors=errors, newline="") + else: + out_data = StringIO("") + writer = csv.DictWriter(out_data, fieldnames=csv_column_names, **kwargs) + writer.writeheader() + for data in box_list: + writer.writerow(data) + if not filename: + return out_data.getvalue() # type: ignore + out_data.close() + + +def _from_csv( + csv_string: Optional[str] = None, + filename: Optional[Union[str, PathLike]] = None, + encoding: str = "utf-8", + errors: str = "strict", + **kwargs, +): + if csv_string: + with StringIO(csv_string) as cs: + reader = csv.DictReader(cs) + return [row for row in reader] + _exists(filename) # type: ignore + with open(filename, "r", encoding=encoding, errors=errors, newline="") as f: # type: ignore + reader = csv.DictReader(f, **kwargs) + return [row for row in reader] diff --git a/box/converters.pyi b/box/converters.pyi new file mode 100644 index 0000000..43d2020 --- /dev/null +++ b/box/converters.pyi @@ -0,0 +1,60 @@ +from typing import Any, Callable, Optional, Union, Dict +from os import PathLike + +yaml_available: bool +toml_available: bool +msgpack_available: bool +BOX_PARAMETERS: Any +toml_read_library: Optional[Any] +toml_write_library: Optional[Any] +toml_decode_error: Optional[Callable] + +def _to_json( + obj, filename: Optional[Union[str, PathLike]] = ..., encoding: str = ..., errors: str = ..., **json_kwargs +): ... +def _from_json( + json_string: Optional[str] = ..., + filename: Optional[Union[str, PathLike]] = ..., + encoding: str = ..., + errors: str = ..., + multiline: bool = ..., + **kwargs, +): ... +def _to_yaml( + obj, + filename: Optional[Union[str, PathLike]] = ..., + default_flow_style: bool = ..., + encoding: str = ..., + errors: str = ..., + ruamel_typ: str = ..., + ruamel_attrs: Optional[Dict] = ..., + **yaml_kwargs, +): ... +def _from_yaml( + yaml_string: Optional[str] = ..., + filename: Optional[Union[str, PathLike]] = ..., + encoding: str = ..., + errors: str = ..., + ruamel_typ: str = ..., + ruamel_attrs: Optional[Dict] = ..., + **kwargs, +): ... +def _to_toml(obj, filename: Optional[Union[str, PathLike]] = ..., encoding: str = ..., errors: str = ...): ... +def _from_toml( + toml_string: Optional[str] = ..., + filename: Optional[Union[str, PathLike]] = ..., + encoding: str = ..., + errors: str = ..., +): ... +def _to_msgpack(obj, filename: Optional[Union[str, PathLike]] = ..., **kwargs): ... +def _from_msgpack(msgpack_bytes: Optional[bytes] = ..., filename: Optional[Union[str, PathLike]] = ..., **kwargs): ... +def _to_csv( + box_list, filename: Optional[Union[str, PathLike]] = ..., encoding: str = ..., errors: str = ..., **kwargs +): ... +def _from_csv( + csv_string: Optional[str] = ..., + filename: Optional[Union[str, PathLike]] = ..., + encoding: str = ..., + errors: str = ..., + **kwargs, +): ... diff --git a/box/exceptions.py b/box/exceptions.py new file mode 100644 index 0000000..18c82d0 --- /dev/null +++ b/box/exceptions.py @@ -0,0 +1,22 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + + +class BoxError(Exception): + """Non standard dictionary exceptions""" + + +class BoxKeyError(BoxError, KeyError, AttributeError): + """Key does not exist""" + + +class BoxTypeError(BoxError, TypeError): + """Cannot handle that instance's type""" + + +class BoxValueError(BoxError, ValueError): + """Issue doing something with that value""" + + +class BoxWarning(UserWarning): + """Here be dragons""" diff --git a/box/exceptions.pyi b/box/exceptions.pyi new file mode 100644 index 0000000..2be6b54 --- /dev/null +++ b/box/exceptions.pyi @@ -0,0 +1,5 @@ +class BoxError(Exception): ... +class BoxKeyError(BoxError, KeyError, AttributeError): ... +class BoxTypeError(BoxError, TypeError): ... +class BoxValueError(BoxError, ValueError): ... +class BoxWarning(UserWarning): ... diff --git a/box/from_file.py b/box/from_file.py new file mode 100644 index 0000000..8f4ce68 --- /dev/null +++ b/box/from_file.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +from json import JSONDecodeError +from os import PathLike +from pathlib import Path +from typing import Optional, Callable, Dict, Union +import sys + +from box.box import Box +from box.box_list import BoxList +from box.converters import msgpack_available, toml_read_library, yaml_available, toml_decode_error +from box.exceptions import BoxError + +try: + from ruamel.yaml import YAMLError +except ImportError: + try: + from yaml import YAMLError # type: ignore + except ImportError: + YAMLError = False # type: ignore + +try: + from msgpack import UnpackException # type: ignore +except ImportError: + UnpackException = False # type: ignore + + +__all__ = ["box_from_file", "box_from_string"] + + +def _to_json(file, encoding, errors, **kwargs): + try: + return Box.from_json(filename=file, encoding=encoding, errors=errors, **kwargs) + except JSONDecodeError: + raise BoxError("File is not JSON as expected") + except BoxError: + return BoxList.from_json(filename=file, encoding=encoding, errors=errors, **kwargs) + + +def _to_csv(file, encoding, errors, **kwargs): + return BoxList.from_csv(filename=file, encoding=encoding, errors=errors, **kwargs) + + +def _to_yaml(file, encoding, errors, **kwargs): + if not yaml_available: + raise BoxError( + f'File "{file}" is yaml but no package is available to open it. Please install "ruamel.yaml" or "PyYAML"' + ) + try: + return Box.from_yaml(filename=file, encoding=encoding, errors=errors, **kwargs) + except YAMLError: + raise BoxError("File is not YAML as expected") + except BoxError: + return BoxList.from_yaml(filename=file, encoding=encoding, errors=errors, **kwargs) + + +def _to_toml(file, encoding, errors, **kwargs): + if not toml_read_library: + raise BoxError(f'File "{file}" is toml but no package is available to open it. Please install "tomli"') + try: + return Box.from_toml(filename=file, encoding=encoding, errors=errors, **kwargs) + except toml_decode_error: + raise BoxError("File is not TOML as expected") + + +def _to_msgpack(file, _, __, **kwargs): + if not msgpack_available: + raise BoxError(f'File "{file}" is msgpack but no package is available to open it. Please install "msgpack"') + try: + return Box.from_msgpack(filename=file, **kwargs) + except (UnpackException, ValueError): + raise BoxError("File is not msgpack as expected") + except BoxError: + return BoxList.from_msgpack(filename=file, **kwargs) + + +converters = { + "json": _to_json, + "jsn": _to_json, + "yaml": _to_yaml, + "yml": _to_yaml, + "toml": _to_toml, + "tml": _to_toml, + "msgpack": _to_msgpack, + "pack": _to_msgpack, + "csv": _to_csv, +} # type: Dict[str, Callable] + + +def box_from_file( + file: Union[str, PathLike], + file_type: Optional[str] = None, + encoding: str = "utf-8", + errors: str = "strict", + **kwargs, +) -> Union[Box, BoxList]: + """ + Loads the provided file and tries to parse it into a Box or BoxList object as appropriate. + + :param file: Location of file + :param encoding: File encoding + :param errors: How to handle encoding errors + :param file_type: manually specify file type: json, toml or yaml + :return: Box or BoxList + """ + + if not isinstance(file, Path): + file = Path(file) + if not file.exists(): + raise BoxError(f'file "{file}" does not exist') + file_type = file_type or file.suffix + file_type = file_type.lower().lstrip(".") + if file_type.lower() in converters: + return converters[file_type.lower()](file, encoding, errors, **kwargs) # type: ignore + raise BoxError(f'"{file_type}" is an unknown type. Please use either csv, toml, msgpack, yaml or json') + + +def box_from_string(content: str, string_type: str = "json") -> Union[Box, BoxList]: + """ + Parse the provided string into a Box or BoxList object as appropriate. + + :param content: String to parse + :param string_type: manually specify file type: json, toml or yaml + :return: Box or BoxList + """ + + if string_type == "json": + try: + return Box.from_json(json_string=content) + except JSONDecodeError: + raise BoxError("File is not JSON as expected") + except BoxError: + return BoxList.from_json(json_string=content) + elif string_type == "toml": + try: + return Box.from_toml(toml_string=content) + except toml_decode_error: # type: ignore + raise BoxError("File is not TOML as expected") + except BoxError: + return BoxList.from_toml(toml_string=content) + elif string_type == "yaml": + try: + return Box.from_yaml(yaml_string=content) + except YAMLError: + raise BoxError("File is not YAML as expected") + except BoxError: + return BoxList.from_yaml(yaml_string=content) + else: + raise BoxError(f"Unsupported string_string of {string_type}") diff --git a/box/from_file.pyi b/box/from_file.pyi new file mode 100644 index 0000000..9e8be8a --- /dev/null +++ b/box/from_file.pyi @@ -0,0 +1,16 @@ +from box.box import Box as Box +from box.box_list import BoxList as BoxList +from os import PathLike +from typing import Any, Union + +def box_from_file( + file: Union[str, PathLike], + file_type: str = ..., + encoding: str = ..., + errors: str = ..., + **kwargs: Any, +) -> Union[Box, BoxList]: ... +def box_from_string( + content: str, + string_type: str = ..., +) -> Union[Box, BoxList]: ... diff --git a/box/py.typed b/box/py.typed new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/box/py.typed diff --git a/box/shorthand_box.py b/box/shorthand_box.py new file mode 100644 index 0000000..a82edbd --- /dev/null +++ b/box/shorthand_box.py @@ -0,0 +1,69 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +from typing import Dict + +from box.box import Box + +__all__ = ["SBox", "DDBox"] + + +class SBox(Box): + """ + ShorthandBox (SBox) allows for + property access of `dict` `json` and `yaml` + """ + + _protected_keys = dir({}) + [ + "to_dict", + "to_json", + "to_yaml", + "json", + "yaml", + "from_yaml", + "from_json", + "dict", + "toml", + "from_toml", + "to_toml", + ] + + @property + def dict(self) -> Dict: + return self.to_dict() + + @property + def json(self) -> str: + return self.to_json() + + @property + def yaml(self) -> str: + return self.to_yaml() + + @property + def toml(self) -> str: + return self.to_toml() + + def __repr__(self): + return f"{self.__class__.__name__}({self})" + + def copy(self) -> "SBox": + return SBox(super(SBox, self).copy()) + + def __copy__(self) -> "SBox": + return SBox(super(SBox, self).copy()) + + +class DDBox(SBox): + def __init__(self, *args, **kwargs): + kwargs["box_dots"] = True + kwargs["default_box"] = True + super().__init__(*args, **kwargs) + + def __new__(cls, *args, **kwargs): + obj = super().__new__(cls, *args, **kwargs) + obj._box_config["box_dots"] = True + obj._box_config["default_box"] = True + return obj + + def __repr__(self) -> str: + return f"{self.__class__.__name__}({self})" diff --git a/box/shorthand_box.pyi b/box/shorthand_box.pyi new file mode 100644 index 0000000..deef693 --- /dev/null +++ b/box/shorthand_box.pyi @@ -0,0 +1,17 @@ +from typing import Dict + +from box.box import Box as Box + +class SBox(Box): + @property + def dict(self) -> Dict: ... + @property + def json(self) -> str: ... + @property + def yaml(self) -> str: ... + @property + def toml(self) -> str: ... + def copy(self) -> "SBox": ... + def __copy__(self) -> "SBox": ... + +class DDBox(Box): ... |
