Conatum
4229fe8b18
Add `LEQ` condition type. Ensure that batch updates don't fire with nothing to update. Add `cast_row` to `update_many` for handling typed `NULL`s.
75 lines
1.9 KiB
Python
75 lines
1.9 KiB
Python
from .connection import _replace_char
|
|
|
|
|
|
class Condition:
|
|
"""
|
|
ABC representing a selection condition.
|
|
"""
|
|
__slots__ = ()
|
|
|
|
def apply(self, key, values, conditions):
|
|
raise NotImplementedError
|
|
|
|
|
|
class NOT(Condition):
|
|
__slots__ = ('value',)
|
|
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
def apply(self, key, values, conditions):
|
|
item = self.value
|
|
if isinstance(item, (list, tuple)):
|
|
if item:
|
|
conditions.append("{} NOT IN ({})".format(key, ", ".join([_replace_char] * len(item))))
|
|
values.extend(item)
|
|
else:
|
|
raise ValueError("Cannot check an empty iterable!")
|
|
else:
|
|
conditions.append("{}!={}".format(key, _replace_char))
|
|
values.append(item)
|
|
|
|
|
|
class GEQ(Condition):
|
|
__slots__ = ('value',)
|
|
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
def apply(self, key, values, conditions):
|
|
item = self.value
|
|
if isinstance(item, (list, tuple)):
|
|
raise ValueError("Cannot apply GEQ condition to a list!")
|
|
else:
|
|
conditions.append("{} >= {}".format(key, _replace_char))
|
|
values.append(item)
|
|
|
|
|
|
class LEQ(Condition):
|
|
__slots__ = ('value',)
|
|
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
def apply(self, key, values, conditions):
|
|
item = self.value
|
|
if isinstance(item, (list, tuple)):
|
|
raise ValueError("Cannot apply LEQ condition to a list!")
|
|
else:
|
|
conditions.append("{} <= {}".format(key, _replace_char))
|
|
values.append(item)
|
|
|
|
|
|
class Constant(Condition):
|
|
__slots__ = ('value',)
|
|
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
def apply(self, key, values, conditions):
|
|
conditions.append("{} {}".format(key, self.value))
|
|
|
|
|
|
NULL = Constant('IS NULL')
|
|
NOTNULL = Constant('IS NOT NULL')
|