Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __sqlrepr__(self, db):
fieldName = self.fieldName
if isinstance(fieldName, SQLExpression):
fieldName = sqlrepr(fieldName, db)
return self.alias + "." + fieldName
def PostgresBinaryConverter(value, db):
assert db == 'postgres'
return sqlrepr(escape_bytea(value), db)
select += " %s" % _str_or_sqlrepr(self.ops['items'][0], db)
join = []
join_str = ''
if self.ops['join'] is not NoDefault and self.ops['join'] is not None:
_join = self.ops['join']
if isinstance(_join, str):
join_str = " " + _join
elif isinstance(_join, SQLJoin):
join.append(_join)
else:
join.extend(_join)
tables = set()
for x in self.ops['staticTables']:
if isinstance(x, SQLExpression):
x = sqlrepr(x, db)
tables.add(x)
things = list(self.ops['items']) + join
if self.ops['clause'] is not NoDefault:
things.append(self.ops['clause'])
for thing in things:
if isinstance(thing, SQLExpression):
tables.update(tablesUsedSet(thing, db))
for j in join:
t1 = _str_or_sqlrepr(j.table1, db)
if t1 in tables:
tables.remove(t1)
t2 = _str_or_sqlrepr(j.table2, db)
if t2 in tables:
tables.remove(t2)
if tables:
select += " FROM %s" % ", ".join(sorted(tables))
def __sqlrepr__(self, db):
if self.proxy.soClass is None:
return '_DELAYED_' + self.attr
val = self._resolve()
if isinstance(val, SQLExpression):
val = sqlrepr(val, db)
return val
join = "%s %s ON %s" % (self.op, sqlrepr(self.table2, db),
on_condition)
if self.table1:
join = "%s %s" % (sqlrepr(self.table1, db), join)
return join
elif self.using_columns:
using_columns = []
for col in self.using_columns:
if hasattr(col, "__sqlrepr__"):
col = sqlrepr(col, db)
using_columns.append(col)
using_columns = ", ".join(using_columns)
join = "%s %s USING (%s)" % (self.op, sqlrepr(self.table2, db),
using_columns)
if self.table1:
join = "%s %s" % (sqlrepr(self.table1, db), join)
return join
else:
RuntimeError, "Impossible error"
def __sqlrepr__(self, db):
return "(%s %s (%s))" % (
sqlrepr(self.expr, db), self._get_op(db), sqlrepr(self.string, db)
)
def __sqlrepr__(self, db):
return "%s (%s)" % (self.op, sqlrepr(self.subquery, db))
def __sqlrepr__(self, db):
return " UNION ".join([str(sqlrepr(t, db)) for t in self.tables])
def __sqlrepr__(self, db):
update = "%s %s" % (self.sqlName(), self.table)
update += " SET"
first = True
if self.template is not NoDefault:
for i in range(len(self.template)):
if first:
first = False
else:
update += ","
update += " %s=%s" % (self.template[i],
sqlrepr(self.values[i], db))
else:
for key, value in sorted(self.values.items()):
if first:
first = False
else:
update += ","
update += " %s=%s" % (key, sqlrepr(value, db))
if self.whereClause is not NoDefault:
update += " WHERE %s" % _str_or_sqlrepr(self.whereClause, db)
return update
def tablesUsedSet(self, db):
tables = set()
for table in self.tablesUsedImmediate():
if hasattr(table, '__sqlrepr__'):
table = sqlrepr(table, db)
tables.add(table)
for component in self.components():
tables.update(tablesUsedSet(component, db))
return tables