Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions .travis.yml
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ language: python

jobs:
include:
- python: 3.9
- python: 3.8
- python: 3.7
- python: 3.6
Expand All @@ -10,7 +11,7 @@ jobs:
- python: 2.7
- python: pypy3
- python: pypy
- name: "Python: 3.7"
- name: "Python: 2.7"
os: osx
osx_image: xcode11.2 # Python 3.7.4 running on macOS 10.14.4
language: shell # 'language: python' is an error on Travis CI macOS
Expand All @@ -22,5 +23,7 @@ jobs:
- python -m pip install --upgrade pip
env: PATH=/c/Python38:/c/Python38/Scripts:$PATH

install: python -m pip install . pytest
script: python -m pytest
install: python -m pip install --upgrade . pytest==4.6.1 pytest-cov==2.8.1 codecov==2.1.10
script: python -m pytest -vv --cov=jsonstore --cov-append
after_success:
- codecov
6 changes: 5 additions & 1 deletion README.rst
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
|PyPi Package| |Build Status| |Codacy Rating|
|PyPi Package| |Build Status| |Codacy Rating| |Coverage Report|

jsonstore
=========
Expand Down Expand Up @@ -47,6 +47,8 @@ Basics
assert store['a_list', -1] == 3
# you can use slices in lists
assert len(store['a_list', 1:]) == 2
del store['a_list', :2]
assert store.a_list == [3]

# deep copies are made when assigning values
my_list = ['fun']
Expand Down Expand Up @@ -95,3 +97,5 @@ file until all of the transactions have been closed.
:target: https://www.codacy.com/app/evilumbrella-github/python-jsonstore?utm_source=github.com&utm_medium=referral&utm_content=Code0x58/python-jsonstore&utm_campaign=Badge_Grade
.. |PyPi Package| image:: https://badge.fury.io/py/python-jsonstore.svg
:target: https://pypi.org/project/python-jsonstore/
.. |Coverage Report| image:: https://codecov.io/gh/Code0x58/python-jsonstore/branch/master/graph/badge.svg
:target: https://codecov.io/gh/Code0x58/python-jsonstore
125 changes: 74 additions & 51 deletions jsonstore.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,13 @@

__all__ = ["JsonStore"]

STRING_TYPES = (str,)
INT_TYPES = (int,)
if sys.version_info < (3,):
STRING_TYPES += (unicode,)
INT_TYPES += (long,)
VALUE_TYPES = (bool, int, float, type(None)) + INT_TYPES


class JsonStore(object):
"""A class to provide object based access to a JSON file"""
Expand Down Expand Up @@ -81,9 +88,10 @@ def __getattr__(self, key):
raise AttributeError(key)

@classmethod
def _valid_object(cls, obj, parents=None):
def _verify_object(cls, obj, parents=None):
"""
Determine if the object can be encoded into JSON
Raise an exception if the object is not suitable for assignment.

"""
# pylint: disable=unicode-builtin,long-builtin
if isinstance(obj, (dict, list)):
Expand All @@ -94,85 +102,100 @@ def _valid_object(cls, obj, parents=None):
parents.append(obj)

if isinstance(obj, dict):
return all(
cls._valid_string(k) and cls._valid_object(v, parents)
for k, v in obj.items()
)
for k, v in obj.items():
if not cls._valid_string(k):
# this is necessary because of the JSON serialisation
raise TypeError("a dict has non-string keys")
cls._verify_object(v, parents)
elif isinstance(obj, (list, tuple)):
return all(cls._valid_object(o, parents) for o in obj)
for o in obj:
cls._verify_object(o, parents)
else:
return cls._valid_value(obj)

@classmethod
def _valid_value(cls, value):
if isinstance(value, (bool, int, float, type(None))):
return True
elif sys.version_info < (3,) and isinstance(value, long):
if isinstance(value, VALUE_TYPES):
return True
else:
return cls._valid_string(value)

@classmethod
def _valid_string(cls, value):
if isinstance(value, str):
if isinstance(value, STRING_TYPES):
return True
elif sys.version_info < (3,):
return isinstance(value, unicode)
else:
return False

def __setattr__(self, key, value):
if not self._valid_object(value):
raise AttributeError
self._data[key] = deepcopy(value)
@classmethod
def _canonical_key(cls, key):
"""Convert a set/get/del key into the canonical form."""
if cls._valid_string(key):
return tuple(key.split("."))

if isinstance(key, (tuple, list)):
key = tuple(key)
if not key:
raise TypeError("key must be a string or non-empty tuple/list")
return key

raise TypeError("key must be a string or non-empty tuple/list")

def __setattr__(self, attr, value):
self._verify_object(value)
self._data[attr] = deepcopy(value)
self._do_auto_commit()

def __delattr__(self, key):
del self._data[key]
def __delattr__(self, attr):
del self._data[attr]

def __get_obj(self, full_path):
"""
Returns the object which is under the given path
"""
if isinstance(full_path, (tuple, list)):
steps = full_path
else:
steps = full_path.split(".")
def __get_obj(self, steps):
"""Returns the object which is under the given path."""
path = []
obj = self._data
if not full_path:
return obj
for step in steps:
path.append(step)
if isinstance(obj, dict) and not self._valid_string(step):
# this is necessary because of the JSON serialisation
raise TypeError("%s is a dict and %s is not a string" % (path, step))
try:
obj = obj[step]
except KeyError:
raise KeyError(".".join(path))
except (KeyError, IndexError, TypeError) as e:
raise type(e)("unable to get %s from %s: %s" % (step, path, e))
path.append(step)
return obj

def __setitem__(self, name, value):
path, _, key = name.rpartition(".")
if self._valid_object(value):
dictionary = self.__get_obj(path)
dictionary[key] = deepcopy(value)
self._do_auto_commit()
else:
raise AttributeError
def __setitem__(self, key, value):
steps = self._canonical_key(key)
path, step = steps[:-1], steps[-1]
self._verify_object(value)
container = self.__get_obj(path)
if isinstance(container, dict) and not self._valid_string(step):
raise TypeError("%s is a dict and %s is not a string" % (path, step))
try:
container[step] = deepcopy(value)
except (IndexError, TypeError) as e:
raise type(e)("unable to set %s from %s: %s" % (step, path, e))
self._do_auto_commit()

def __getitem__(self, key):
obj = self.__get_obj(key)
if obj is self._data:
raise KeyError
steps = self._canonical_key(key)
obj = self.__get_obj(steps)
return deepcopy(obj)

def __delitem__(self, name):
if isinstance(name, (tuple, list)):
path = name[:-1]
key = name[-1]
else:
path, _, key = name.rpartition(".")
def __delitem__(self, key):
steps = self._canonical_key(key)
path, step = steps[:-1], steps[-1]
obj = self.__get_obj(path)
del obj[key]
try:
del obj[step]
except (KeyError, IndexError, TypeError) as e:
raise type(e)("unable to delete %s from %s: %s" % (step, path, e))

def __contains__(self, key):
return key in self._data
steps = self._canonical_key(key)
try:
self.__get_obj(steps)
return True
except (KeyError, IndexError, TypeError):
# this is rather permissive as the types are dynamic
return False
42 changes: 37 additions & 5 deletions test_jsonstore.py
Original file line number Diff line number Diff line change
Expand Up @@ -105,19 +105,39 @@ def test_assign_valid_types(self):
self.assertRaises(KeyError, self._getitem(name))
self.assertRaises(AttributeError, self._getattr(name))

def test_slices(self):
self.store.list = [1, 2, 3]
self.store["list", :2] = ["a", "b"]
self.assertEqual(self.store["list", 1:], ["b", 3])
del self.store["list", :1]
self.assertEqual(self.store.list, ["b", 3])

def test_assign_invalid_types(self):
for method in (self._setattr, self._setitem):

def assign(value):
return method("key", value)

self.assertRaises(AttributeError, assign(set()))
self.assertRaises(AttributeError, assign(object()))
self.assertRaises(AttributeError, assign(None for i in range(2)))
self.assertRaises(TypeError, assign(set()))
self.assertRaises(TypeError, assign(object()))
self.assertRaises(TypeError, assign(None for i in range(2)))
self.assertRaises(TypeError, assign({1: 1}))

def test_assign_bad_keys(self):
# FIXME: a ValueError would make more sense
self.assertRaises(AttributeError, self._setitem(1, 2))
value = 1
# the root object is a dict, so a string key is needed
self.assertRaises(TypeError, self._setitem(1, value))
self.assertRaises(TypeError, self._setitem((1, "a"), value))

self.store["dict"] = {}
self.store["list"] = []
self.assertRaises(TypeError, self._setitem((), value))
self.assertRaises(TypeError, self._setitem(("dict", 1), value))
self.assertRaises(TypeError, self._setitem(("dict", slice(1)), value))
self.assertRaises(TypeError, self._setitem(("list", "a"), value))
self.assertRaises(TypeError, self._setitem(("list", slice("a")), value))
self.assertRaises(IndexError, self._setitem(("list", 1), value))


def test_retrieve_values(self):
for name, value in self.TEST_DATA:
Expand Down Expand Up @@ -177,6 +197,7 @@ def test_nested_getitem(self):
assert self.store[["list", 0, "key", -1]] == "last"
self.assertRaises(TypeError, self._getitem("list.0.key.1"))
assert len(self.store["list", 0, "key", 1:]) == 2
self.assertRaises(IndexError, self._getitem(("list", 1)))

def test_del(self):
self.store.key = None
Expand All @@ -187,6 +208,17 @@ def test_del(self):
del self.store["key"]
self.assertRaises(KeyError, self._getitem("key"))

with self.assertRaises(KeyError):
del self.store["missing"]

self.store.list = []
with self.assertRaises(IndexError):
del self.store["list", 1]

self.store.dict = {}
with self.assertRaises(TypeError):
del self.store["dict", slice("a")]

def test_context_and_deserialisation(self):
store_file = mktemp()
for name, value in self.TEST_DATA:
Expand Down