Rework item operations

LowVoltage/attribute_types.py

@@ -2,9 +2,9 @@
 
 # Copyright 2013-2014 Vincent Jacques <vincent@vincent-jacques.net>
 
-BINARY = "B"
+# @todo BINARY = "B"
 BOOLEAN = "BOOL"
-BINARY_SET = "BS"
+# @todo BINARY_SET = "BS"
 LIST = "L"
 MAP = "M"
 NUMBER = "N"

LowVoltage/operations/batch_operations.py

@@ -5,6 +5,7 @@
 import unittest
 
 from LowVoltage.operations.operation import Operation, ReturnConsumedCapacityMixin, ReturnItemCollectionMetricsMixin
+from LowVoltage.operations.conversion import _convert_dict_to_db, _convert_value_to_db, _convert_db_to_dict, _convert_db_to_value
 
 
 class BatchGetItem(Operation, ReturnConsumedCapacityMixin):
@@ -34,7 +35,7 @@ def keys(self, *keys):
         for key in keys:
             if isinstance(key, dict):
                 key = [key]
-            self.__last_table["Keys"].extend(self._convert_dict(k) for k in key)
+            self.__last_table["Keys"].extend(_convert_dict_to_db(k) for k in key)
         return self
 
     def consistent_read_true(self):
@@ -166,14 +167,14 @@ def delete(self, *keys):
         for key in keys:
             if isinstance(key, dict):
                 key = [key]
-            self.__last_table.extend({"DeleteRequest": {"Key": self._convert_dict(k)}} for k in key)
+            self.__last_table.extend({"DeleteRequest": {"Key": _convert_dict_to_db(k)}} for k in key)
         return self
 
     def put(self, *keys):
         for key in keys:
             if isinstance(key, dict):
                 key = [key]
-            self.__last_table.extend({"PutRequest": {"Item": self._convert_dict(k)}} for k in key)
+            self.__last_table.extend({"PutRequest": {"Item": _convert_dict_to_db(k)}} for k in key)
         return self
 
 

LowVoltage/operations/conversion.py

@@ -0,0 +1,117 @@
+# -*- coding: utf-8 -*-
+
+# Copyright 2013-2014 Vincent Jacques <vincent@vincent-jacques.net>
+
+import numbers
+import unittest
+
+
+def _convert_dict_to_db(attributes):
+    return {
+        key: _convert_value_to_db(val)
+        for key, val in attributes.iteritems()
+    }
+
+
+def _convert_value_to_db(value):
+    if isinstance(value, basestring):
+        return {"S": value}
+    elif isinstance(value, bool):
+        return {"BOOL": value}
+    elif isinstance(value, numbers.Integral):
+        return {"N": str(value)}
+    elif value is None:
+        return {"NULL": True}
+    elif isinstance(value, set):
+        for first in value:
+            break  # http://stackoverflow.com/a/59841/905845
+        else:
+            raise TypeError
+        if isinstance(first, basestring):
+            return {"SS": list(value)}
+        elif isinstance(first, numbers.Integral):
+            return {"NS": [str(n) for n in value]}
+        else:
+            raise TypeError
+    elif isinstance(value, list):
+        return {"L": [_convert_value_to_db(v) for v in value]}
+    elif isinstance(value, dict):
+        return {"M": {n: _convert_value_to_db(v) for n, v in value.iteritems()}}
+    else:
+        raise TypeError
+
+
+def _convert_db_to_dict(attributes):
+    return {
+        key: _convert_db_to_value(val)
+        for key, val in attributes.iteritems()
+    }
+
+
+def _convert_db_to_value(value):
+    if "S" in value:
+        return value["S"]
+    elif "BOOL" in value:
+        return value["BOOL"]
+    elif "N" in value:
+        return int(value["N"])
+    elif "NULL" in value:
+        return None
+    elif "NS" in value:
+        return set(int(v) for v in value["NS"])
+    elif "SS" in value:
+        return set(v for v in value["SS"])
+    elif "L" in value:
+        return [_convert_db_to_value(v) for v in value["L"]]
+    elif "M" in value:
+        return {n: _convert_db_to_value(v) for n, v in value["M"].iteritems()}
+    else:
+        raise TypeError
+
+
+class ConversionUnitTests(unittest.TestCase):
+    def testConvertValueToDb(self):
+        self.assertEqual(_convert_value_to_db("foo"), {"S": "foo"})
+        self.assertEqual(_convert_value_to_db(True), {"BOOL": True})
+        self.assertEqual(_convert_value_to_db(False), {"BOOL": False})
+        self.assertEqual(_convert_value_to_db(42), {"N": "42"})
+        self.assertEqual(_convert_value_to_db(None), {"NULL": True})
+        self.assertEqual(_convert_value_to_db(set([42, 43])), {"NS": ["42", "43"]})
+        self.assertEqual(_convert_value_to_db(set(["foo", "bar"])), {"SS": ["foo", "bar"]})
+        self.assertEqual(_convert_value_to_db([True, 42]), {"L": [{"BOOL": True}, {"N": "42"}]})
+        self.assertEqual(_convert_value_to_db({"a": True, "b": 42}), {"M": {"a": {"BOOL": True}, "b": {"N": "42"}}})
+        with self.assertRaises(TypeError):
+            _convert_value_to_db(set())
+        with self.assertRaises(TypeError):
+            _convert_value_to_db(set([(1, 2)]))
+        with self.assertRaises(TypeError):
+            _convert_value_to_db((1, 2))
+
+    def testConvertTbToValue(self):
+        self.assertEqual(_convert_db_to_value({"S": "foo"}), "foo")
+        self.assertEqual(_convert_db_to_value({"BOOL": True}), True)
+        self.assertEqual(_convert_db_to_value({"BOOL": False}), False)
+        self.assertEqual(_convert_db_to_value({"N": "42"}), 42)
+        self.assertEqual(_convert_db_to_value({"NULL": True}), None)
+        self.assertEqual(_convert_db_to_value({"NS": ["42", "43"]}), set([42, 43]))
+        self.assertEqual(_convert_db_to_value({"SS": ["foo", "bar"]}), set(["foo", "bar"]))
+        self.assertEqual(_convert_db_to_value({"L": [{"BOOL": True}, {"N": "42"}]}), [True, 42])
+        self.assertEqual(_convert_db_to_value({"M": {"a": {"BOOL": True}, "b": {"N": "42"}}}), {"a": True, "b": 42})
+        with self.assertRaises(TypeError):
+            _convert_db_to_value({})
+        with self.assertRaises(TypeError):
+            _convert_db_to_value([])
+        with self.assertRaises(TypeError):
+            _convert_db_to_value("SSS")
+
+    def testConvertDictToDb(self):
+        self.assertEqual(_convert_dict_to_db({}), {})
+        self.assertEqual(_convert_dict_to_db({"a": 42}), {"a": {"N": "42"}})
+
+    def testConvertDbToDict(self):
+        self.assertEqual(_convert_db_to_dict({}), {})
+        self.assertEqual(_convert_db_to_dict({"a": {"N": "42"}}), {"a": 42})
+
+
+if __name__ == "__main__":
+    unittest.main()  # pragma no cover (Test code)