100% coverage

README.md

@@ -12,7 +12,7 @@ DictDataBase is a simple and fast database for handling json or compressed json
 - **No database server** required. Simply import DictDataBase in your project and use it.
 - **Compression**. Configure if the files should be stored as raw json or as json compressed with zlib.
 - **Fast**. A dict can be accessed partially without having to parse the entire file, making the read and writes very efficient.
-- **Tested** with over 400 test cases.
+- **Tested**. 100% coverage, over 1000 test cases.
 
 ### Why use DictDataBase
 - For example, have a webserver dispatch database reads and writes concurrently.

dictdatabase/models.py

@@ -28,9 +28,9 @@ def __init__(self, path, key, where):
 		self.key = key is not None
 
 		if self.key and self.where:
-			raise ValueError("Cannot specify both key and where")
+			raise TypeError("Cannot specify both key and where")
 		if self.key and self.dir:
-			raise ValueError("Cannot specify sub-key when selecting a folder. Specify the key in the path instead.")
+			raise TypeError("Cannot specify sub-key when selecting a folder. Specify the key in the path instead.")
 
 	@property
 	def file_normal(self):

dictdatabase/session.py

@@ -69,7 +69,7 @@ def type_cast(data):
 				self.write_lock._lock()
 				self.original_data = io_unsafe.read(self.db_name)
 				data = {}
-				for k, v in self.original_data:
+				for k, v in self.original_data.items():
 					if self.where(k, v):
 						data[k] = v
 				self.data_handle = data

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "dictdatabase"
-version = "2.1.2"
+version = "2.1.3"
 repository = "https://github.com/mkrd/DictDataBase"
 description = "Easy-to-use database using dicts"
 authors = ["Marcel Kröker <[email protected]>"]

tests/conftest.py

@@ -14,21 +14,25 @@ def env(request):
 @pytest.fixture(params=[True, False])
 def use_compression(request):
 	DDB.config.use_compression = request.param
+	return request.param
 
 
 
 @pytest.fixture(params=[False, True])
 def use_orjson(request):
 	DDB.config.use_orjson = request.param
+	return request.param
 
 
 
 @pytest.fixture(params=[False, True])
 def sort_keys(request):
 	DDB.config.sort_keys = request.param
+	return request.param
 
 
 
 @pytest.fixture(params=[None, 0, 2, "\t"])
 def indent(request):
 	DDB.config.indent = request.param
+	return request.param

tests/test_create.py

@@ -16,6 +16,12 @@ def test_create(env, use_compression, use_orjson, sort_keys, indent):
 		session.write()
 	assert DDB.at("create").read() == {"a": {"b": {"c": "😁"}}}
 
+	with pytest.raises(RuntimeError):
+		DDB.at("create", where=lambda k, v: True).create(force_overwrite=True)
+
+	with pytest.raises(RuntimeError):
+		DDB.at("create", key="any").create(force_overwrite=True)
+
 
 def test_create_edge_cases(env, use_compression, use_orjson, sort_keys, indent):
 	cases = [-2, 0.0, "", "x", [], {}, True]

tests/test_delete.py

@@ -1,13 +1,20 @@
 import dictdatabase as DDB
+import pytest
 
 
 def test_delete(env, use_compression, use_orjson, sort_keys, indent):
-    DDB.at("test_delete").create({"a": 1}, force_overwrite=True)
-    assert DDB.at("test_delete").read() == {"a": 1}
-    DDB.at("test_delete").delete()
-    assert DDB.at("test_delete").read() is None
+	DDB.at("test_delete").create({"a": 1}, force_overwrite=True)
+	assert DDB.at("test_delete").read() == {"a": 1}
+	DDB.at("test_delete").delete()
+	assert DDB.at("test_delete").read() is None
+
+	with pytest.raises(RuntimeError):
+		DDB.at("test_delete", where=lambda k, v: True).delete()
+
+	with pytest.raises(RuntimeError):
+		DDB.at("test_delete", key="any").delete()
 
 
 
 def test_delete_nonexistent(env, use_compression, use_orjson, sort_keys, indent):
-    DDB.at("test_delete_nonexistent").delete()
+	DDB.at("test_delete_nonexistent").delete()

tests/test_excepts.py

@@ -52,7 +52,7 @@ def test_except_on_write_outside_session(env, use_compression, use_orjson, sort_
 
 
 def test_wildcard_and_subkey_except(env, use_compression, use_orjson, sort_keys, indent):
-	with pytest.raises(ValueError):
+	with pytest.raises(TypeError):
 		DDB.at("test_wildcard_and_subkey_except/*", key="key").read()
 
 

tests/test_exists.py

@@ -0,0 +1,12 @@
+import dictdatabase as DDB
+import pytest
+
+
+def test_exists(env, use_compression, use_orjson, sort_keys, indent):
+	DDB.at("test_exists").create({"a": 1}, force_overwrite=True)
+	assert DDB.at("test_exists").exists()
+	assert not DDB.at("test_exists/nonexistent").exists()
+	assert DDB.at("test_exists", key="a").exists()
+	assert not DDB.at("test_exists", key="b").exists()
+	with pytest.raises(RuntimeError):
+		DDB.at("test_exists", where=lambda k, v: True).exists()

tests/test_indentation.py

@@ -0,0 +1,51 @@
+
+import dictdatabase as DDB
+import orjson
+import json
+from dictdatabase import utils, io_unsafe, config
+
+data = {
+    'a': 1,
+    'b': {
+        'c': 2,
+        "cl": [1, "\\"],
+        'd': {
+            'e': 3,
+            "el": [1, "\\"],
+        }
+    },
+    "l": [1, "\\"],
+}
+
+
+def string_dump(db: dict):
+	if not config.use_orjson:
+		return json.dumps(db, indent=config.indent, sort_keys=config.sort_keys)
+	option = orjson.OPT_INDENT_2 if config.indent else 0
+	option |= orjson.OPT_SORT_KEYS if config.sort_keys else 0
+	return orjson.dumps(db, option=option)
+
+
+
+
+def test_indentation(env, use_compression, use_orjson, sort_keys, indent):
+    DDB.at("test_indentation").create(data, force_overwrite=True)
+
+    with DDB.at("test_indentation", key="b").session() as (session, db_b):
+        db_b["c"] = 3
+        session.write()
+    data["b"]["c"] = 3
+    if use_orjson:
+        assert io_unsafe.read_file("test_indentation", as_bytes=True) == string_dump(data)
+    else:
+        assert io_unsafe.read_file("test_indentation") == string_dump(data)
+
+
+    with DDB.at("test_indentation", key="d").session() as (session, db_d):
+        db_d["e"] = 4
+        session.write()
+    data["b"]["d"]["e"] = 4
+    if use_orjson:
+        assert io_unsafe.read_file("test_indentation", as_bytes=True) == string_dump(data)
+    else:
+        assert io_unsafe.read_file("test_indentation") == string_dump(data)

tests/test_partial.py

@@ -15,6 +15,7 @@ def test_subread(env, use_compression, use_orjson, sort_keys, indent):
 	DDB.at(name).create(j, force_overwrite=True)
 
 	assert DDB.at(name, key="a").read() == "Hello{}"
+	assert DDB.at(name, where=lambda k, v: isinstance(v, list)).read() == {"b": [0, 1]}
 
 	with pytest.raises(KeyError):
 		DDB.at(name, key="f").read()
@@ -50,3 +51,44 @@ def test_subwrite(env, use_compression, use_orjson, sort_keys, indent):
 		task["f"] = lambda x: (x or 0) + 2
 		session.write()
 	assert DDB.at(name, key="f").read() == 2
+
+
+def test_write_file_where(env, use_compression, use_orjson, sort_keys, indent):
+	name = "test_write_file_where"
+	j = {
+		"a": 1,
+		"b": 20,
+		"c": 3,
+		"d": 40,
+	}
+
+	DDB.at(name).create(j, force_overwrite=True)
+
+	with DDB.at(name, where=lambda k, v: v > 10).session() as (session, vals):
+		vals |= {"b": 30, "d": 50, "e": 60}
+		session.write()
+	assert DDB.at(name).read() == {
+		"a": 1,
+		"b": 30,
+		"c": 3,
+		"d": 50,
+		"e": 60,
+	}
+
+
+def test_dir_where(env, use_compression, use_orjson, sort_keys, indent):
+	name = "test_dir_where"
+	for i in range(5):
+		DDB.at(name, i).create({"k": i}, force_overwrite=True)
+
+	with DDB.at(name, "*", where=lambda k, v: v["k"] > 2).session() as (session, vals):
+		for k, v in vals.items():
+			v["k"] += 1
+		session.write()
+	assert DDB.at(name, "*").read() == {
+		"0": {"k": 0},
+		"1": {"k": 1},
+		"2": {"k": 2},
+		"3": {"k": 4},
+		"4": {"k": 5},
+	}

tests/test_read.py

@@ -19,12 +19,9 @@ def test_file_exists_error(env, use_compression, use_orjson, sort_keys, indent):
 		DDB.at("test_file_exists_error").read()
 
 
-def test_exists(env, use_compression, use_orjson, sort_keys, indent):
-	DDB.at("test_exists").create({"a": 1}, force_overwrite=True)
-	assert DDB.at("test_exists").exists()
-	assert not DDB.at("test_exists/nonexistent").exists()
-	assert DDB.at("test_exists", key="a").exists()
-	assert not DDB.at("test_exists", key="b").exists()
+def test_invalid_params(env, use_compression, use_orjson, sort_keys, indent):
+	with pytest.raises(TypeError):
+		DDB.at("test_invalid_params", key="any", where=lambda k, v: True).read()
 
 
 

tests/test_write.py

@@ -55,6 +55,6 @@ def test_multi_session(env, use_compression, use_orjson, sort_keys, indent):
 
 
 def test_write_wildcard_key_except(env, use_compression, use_orjson, sort_keys, indent):
-	with pytest.raises(ValueError):
+	with pytest.raises(TypeError):
 		with DDB.at("test/*", key="any").session() as (session, d):
 			pass