Infinidat/infi.clickhouse_orm
1
1
Fork 0
mirror of https://github.com/Infinidat/infi.clickhouse_orm.git synced 2024-11-10 19:36:33 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #183 from mangototango/small_issues

Browse Source 
Small fixes
This commit is contained in:
Kobi Tal 2021-10-21 09:36:30 +03:00 committed by GitHub
commit ed33c09dd7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 15 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/infi/clickhouse_orm/database.py
View File

@ -413,7 +413,7 @@ class Database(object):
except ServerError as e:
logger.exception('Cannot determine server version (%s), assuming 1.1.0', e)
ver = '1.1.0'
return tuple(int(n) for n in ver.split('.')) if as_tuple else ver
return tuple(int(n) for n in ver.split('.') if n.isdigit()) if as_tuple else ver
def _is_existing_database(self):
r = self._send("SELECT count() FROM system.databases WHERE name = '%s'" % self.db_name)

4
src/infi/clickhouse_orm/query.py
View File

@ -209,10 +209,10 @@ class Q(object):
@classmethod
def _construct_from(cls, l_child, r_child, mode):
if mode == l_child._mode:
if mode == l_child._mode and not l_child._negate:
q = deepcopy(l_child)
q._children.append(deepcopy(r_child))
elif mode == r_child._mode:
elif mode == r_child._mode and not r_child._negate:
q = deepcopy(r_child)
q._children.append(deepcopy(l_child))
else:

12
tests/test_querysets.py
View File

@ -302,6 +302,18 @@ class QuerySetTestCase(TestCaseWithData):
self.assertEqual(qs.conditions_as_sql(),
"(first_name = 'a') AND (greater(`height`, 1.7)) AND (last_name = 'b')")
def test_precedence_of_negation(self):
p = ~Q(first_name='a')
q = Q(last_name='b')
r = p & q
self.assertEqual(r.to_sql(Person), "(last_name = 'b') AND (NOT (first_name = 'a'))")
r = q & p
self.assertEqual(r.to_sql(Person), "(last_name = 'b') AND (NOT (first_name = 'a'))")
r = q | p
self.assertEqual(r.to_sql(Person), "(last_name = 'b') OR (NOT (first_name = 'a'))")
r = ~q & p
self.assertEqual(r.to_sql(Person), "(NOT (last_name = 'b')) AND (NOT (first_name = 'a'))")
def test_invalid_filter(self):
qs = Person.objects_in(self.database)
with self.assertRaises(TypeError):