openstack/deb-python-cassandra-driver
Code Issues Proposed changes

fix list append and prepend with empty lists

Browse Source
This commit is contained in:
Lucas Chi
2014-11-12 17:03:37 -05:00
parent ccf694c201
commit c388365236
2 changed files with 19 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
cqlengine/statements.py
View File

@@ -248,11 +248,11 @@ class ListUpdateClause(ContainerUpdateClause):
qs += ['"{}" = %({})s'.format(self.field, ctx_id)]
ctx_id += 1
if self._prepend:
if self._prepend is not None:
qs += ['"{0}" = %({1})s + "{0}"'.format(self.field, ctx_id)]
ctx_id += 1
if self._append:
if self._append is not None:
qs += ['"{0}" = "{0}" + %({1})s'.format(self.field, ctx_id)]
return ', '.join(qs)
@@ -267,13 +267,13 @@ class ListUpdateClause(ContainerUpdateClause):
if self._assignments is not None:
ctx[str(ctx_id)] = self._to_database(self._assignments)
ctx_id += 1
if self._prepend:
if self._prepend is not None:
# CQL seems to prepend element at a time, starting
# with the element at idx 0, we can either reverse
# it here, or have it inserted in reverse
ctx[str(ctx_id)] = self._to_database(list(reversed(self._prepend)))
ctx_id += 1
if self._append:
if self._append is not None:
ctx[str(ctx_id)] = self._to_database(self._append)
def _analyze(self):

17
cqlengine/tests/statements/test_update_statement.py
View File

@@ -1,9 +1,12 @@
from unittest import TestCase
from cqlengine.columns import Set
from cqlengine.columns import Set, List
from cqlengine.operators import *
from cqlengine.statements import UpdateStatement, WhereClause, AssignmentClause, SetUpdateClause
from cqlengine.statements import (UpdateStatement, WhereClause,
AssignmentClause, SetUpdateClause,
ListUpdateClause)
import six
class UpdateStatementTests(TestCase):
def test_table_rendering(self):
@@ -50,3 +53,13 @@ class UpdateStatementTests(TestCase):
us = UpdateStatement('table')
us.add_assignment_clause(SetUpdateClause('a', Set.Quoter(set()), operation='add'))
self.assertEqual(six.text_type(us), 'UPDATE table SET "a" = "a" + %(0)s')
def test_update_list_prepend_with_empty_list(self):
us = UpdateStatement('table')
us.add_assignment_clause(ListUpdateClause('a', List.Quoter([]), operation='prepend'))
self.assertEqual(six.text_type(us), 'UPDATE table SET "a" = %(0)s + "a"')
def test_update_list_append_with_empty_list(self):
us = UpdateStatement('table')
us.add_assignment_clause(ListUpdateClause('a', List.Quoter([]), operation='append'))
self.assertEqual(six.text_type(us), 'UPDATE table SET "a" = "a" + %(0)s')