making sure we send the consistency_level explicitly

cqlengine/connection.py

@@ -229,9 +229,11 @@ class ConnectionPool(object):
                 raise ex
 
 
-def execute(query, params=None):
+def execute(query, params=None, consistency_level=None):
     params = params or {}
-    return connection_pool.execute(query, params)
+    if consistency_level is None:
+        consistency_level = connection_pool._consistency
+    return connection_pool.execute(query, params, consistency_level)
 
 @contextmanager
 def connection_manager():

cqlengine/models.py

@@ -106,8 +106,8 @@ class ConsistencyDescriptor(object):
 
         qs = model.__queryset__(model)
 
-        def consistency_setter(ts):
-            qs._consistency = ts
+        def consistency_setter(consistency):
+            qs._consistency = consistency
             return qs
 
         return consistency_setter
@@ -227,6 +227,7 @@ class BaseModel(object):
     __dmlquery__ = DMLQuery
 
     __ttl__ = None
+    __consistency__ = None # can be set per query
 
     __read_repair_chance__ = 0.1
 
@@ -446,7 +447,10 @@ class BaseModel(object):
                 setattr(self, self._polymorphic_column_name, self.__polymorphic_key__)
 
         self.validate()
-        self.__dmlquery__(self.__class__, self, batch=self._batch, ttl=self._ttl).update()
+        self.__dmlquery__(self.__class__, self,
+                          batch=self._batch,
+                          ttl=self._ttl,
+                          consistency=self.consistency).update()
 
         #reset the value managers
         for v in self._values.values():

cqlengine/query.py

@@ -363,7 +363,7 @@ class AbstractQuerySet(object):
         if self._batch:
             raise CQLEngineException("Only inserts, updates, and deletes are available in batch mode")
         if self._result_cache is None:
-            columns, self._result_cache = execute(self._select_query(), self._where_values())
+            columns, self._result_cache = execute(self._select_query(), self._where_values(), self._consistency)
             self._construct_result = self._get_result_constructor(columns)
 
     def _fill_result_cache_to_idx(self, idx):
@@ -777,13 +777,15 @@ class DMLQuery(object):
     unlike the read query object, this is mutable
     """
     _ttl = None
+    _consistency = None
 
-    def __init__(self, model, instance=None, batch=None, ttl=None):
+    def __init__(self, model, instance=None, batch=None, ttl=None, consistency=None):
         self.model = model
         self.column_family_name = self.model.column_family_name()
         self.instance = instance
         self._batch = batch
         self._ttl = ttl
+        self._consistency = consistency
 
     def batch(self, batch_obj):
         if batch_obj is not None and not isinstance(batch_obj, BatchQuery):
@@ -894,7 +896,7 @@ class DMLQuery(object):
             if self._batch:
                 self._batch.add_query(qs, query_values)
             else:
-                execute(qs, query_values)
+                execute(qs, query_values, consistency_level=self._consistency)
 
         self._delete_null_columns()
 
@@ -954,7 +956,7 @@ class DMLQuery(object):
             if self._batch:
                 self._batch.add_query(qs, query_values)
             else:
-                execute(qs, query_values)
+                execute(qs, query_values, self._consistency)
 
         # delete any nulled columns
         self._delete_null_columns()
@@ -978,6 +980,6 @@ class DMLQuery(object):
         if self._batch:
             self._batch.add_query(qs, field_values)
         else:
-            execute(qs, field_values)
+            execute(qs, field_values, self._consistency)
 
 

cqlengine/tests/test_consistency.py

@@ -32,4 +32,10 @@ class TestConsistency(BaseConsistencyTest):
             TestConsistencyModel.consistency(ALL).create(text="i am not fault tolerant this way")
 
         args = m.call_args
-        self.assertEqual(ALL, args[2])
+        self.assertEqual(ALL, args[0][2])
+
+    def test_queryset_is_returned_on_create(self):
+        qs = TestConsistencyModel.consistency(ALL)
+        self.assertTrue(isinstance(qs, TestConsistencyModel.__queryset__), type(qs))
+
+