Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@builder
def do_nothing(self):
if len(self._on_conflict_updates) > 0:
raise QueryException("Can not have two conflict handlers")
self._on_conflict_do_nothing = True
@builder
def join(self, item, how=JoinType.inner):
if isinstance(item, Table):
return Joiner(self, item, how, type_label="table")
elif isinstance(item, QueryBuilder):
return Joiner(self, item, how, type_label="subquery")
elif isinstance(item, AliasedQuery):
return Joiner(self, item, how, type_label="table")
raise ValueError("Cannot join on type '%s'" % type(item))
@builder
def on_duplicate_key_update(self, field, value):
field = Field(field) if not isinstance(field, Field) else field
self._duplicate_updates.append((field, ValueWrapper(value)))
@builder
def range(self, bound, and_bound=None):
self._set_frame_and_bounds("RANGE", bound, and_bound)
@builder
def create_table(self, table):
if self._create_table:
raise AttributeError("'Query' object already has attribute create_table")
self._create_table = table if isinstance(table, Table) else Table(table)
@builder
def on_conflict(self, target_field):
if not self._insert_table:
raise QueryException("On conflict only applies to insert query")
if isinstance(target_field, str):
self._on_conflict_field = self._conflict_field_str(target_field)
elif isinstance(target_field, Field):
self._on_conflict_field = target_field
@builder
def union(self, other):
return _UnionQuery(
self, other, UnionType.distinct, wrapper_cls=self._wrapper_cls
)
@builder
def returning(self, *terms):
for term in terms:
if isinstance(term, Field):
self._return_field(term)
elif isinstance(term, str):
self._return_field_str(term)
elif isinstance(term, ArithmeticExpression):
self._return_other(term)
elif isinstance(term, Function):
raise QueryException("Aggregate functions are not allowed in returning")
else:
self._return_other(self.wrap_constant(term, self._wrapper_cls))
@builder
def union_all(self, other):
self._unions.append((UnionType.all, other))
@builder
def as_(self, alias):
self.alias = alias