Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_select(self):
instance = Select('test')
self.assertEqual(sqlrepr(instance), repr(instance))
ascol = ColumnAS(col.dbName, n)
if isAggregate(col.dbName):
restriction = getattr(col, 'aggregateClause', None)
if restriction:
restrictkey = sqlrepr(restriction, dbName)
aggregates[restrictkey] = \
aggregates.get(restrictkey, [restriction]) + \
[ascol]
else:
aggregates[''].append(ascol)
else:
columns.append(ascol)
metajoin = getattr(cls.sqlmeta, 'join', NoDefault)
clause = getattr(cls.sqlmeta, 'clause', NoDefault)
select = Select(columns,
distinct=True,
# @@ LDO check if this really mattered
# for performance
# @@ Postgres (and MySQL?) extension!
# distinctOn=cls.sqlmeta.idName,
join=metajoin,
clause=clause)
aggregates = aggregates.values()
if aggregates != [[None]]:
join = []
last_alias = "%s_base" % alias
last_id = "id"
last = Alias(select, last_alias)
columns = [
columns = [
ColumnAS(SQLConstant("%s.%s" % (last_alias, x.expr2)),
x.expr2) for x in columns]
for i, agg in enumerate(aggregates):
restriction = agg[0]
if restriction is None:
restriction = clause
else:
restriction = AND(clause, restriction)
agg = agg[1:]
agg_alias = "%s_%s" % (alias, i)
agg_id = '%s_id' % agg_alias
if not last.q.alias.endswith('base'):
last = None
new_alias = Alias(Select(
[ColumnAS(cls.sqlmeta.idName, agg_id)] + agg,
groupBy=cls.sqlmeta.idName,
join=metajoin,
clause=restriction),
agg_alias)
agg_join = LEFTJOINOn(last, new_alias,
"%s.%s = %s.%s" % (
last_alias, last_id,
agg_alias, agg_id))
join.append(agg_join)
for col in agg:
columns.append(
ColumnAS(SQLConstant(
"%s.%s" % (agg_alias, col.expr2)),
col.expr2))
def getValsCol(column):
i = sqlbuilder.table.Image
c = sqlbuilder.table.Image.__getattr__(column)
rid = getRID()
where = (i.enzorun_ID == rid)
a = sqlbuilder.Select(c, where=where, distinct=True)
a = model.Image._connection.queryAll(sqlbuilder.sqlrepr(a))
a = map(str, [r[0] for r in a])
a.sort()
return zip(a,a)
def _SO_selectOneAlt(self, so, columnNames, condition):
if columnNames:
columns = [isinstance(x, string_type)
and sqlbuilder.SQLConstant(x)
or x for x in columnNames]
else:
columns = None
return self.queryOne(self.sqlrepr(sqlbuilder.Select(
columns, staticTables=[so.sqlmeta.table], clause=condition)))
def _query_all_filedata(self):
#return self.fd.select()
return Filedata.select(
# only query things that actually have dependencies
EXISTS(Select(DtNeededList.q.filedata, where=(Outer(Filedata).q.id == DtNeededList.q.filedata))),
# sort by filename
orderBy=Filedata.q.basename
)
def __init__(self, table, alias=None):
if hasattr(table, "sqlmeta"):
tableName = SQLConstant(table.sqlmeta.table)
elif isinstance(table, (Select, Union)):
assert alias is not None, \
"Alias name cannot be constructed from Select instances, " \
"please provide an 'alias' keyword."
tableName = Subquery('', table)
table = None
else:
tableName = SQLConstant(table)
table = None
Table.__init__(self, tableName)
self.table = table
if alias is None:
self._alias_lock.acquire()
try:
AliasTable._alias_counter += 1
alias = "%s_alias%d" % (tableName, AliasTable._alias_counter)
finally: