On 07/27/2016 05:29 PM, T Johnson wrote:
I am generating select queries that include alias tables, literal columns,
renamed columns via label, multiple joins etc.
The queries can be somewhat complex and as a debugging feature, I'd like be
able to know which columns came from which tables...but I don't want to store
this information static and would like it to be derived from the query itself.
If the column is a literal column, constructed, or similar, then there is no
original table. If the table is used as part of an == join key, then the column
should be associated to each of the tables. So if doing three joins, a column
could have three original parent tables, etc.
So I'd like a mapping from columns to a list of tables. Example:
select a as aa, b, c, now() as d
from t1 inner join (select a, b, c from t3) t2
on t1.a == t2.a
This should say:
{aa : [t1, t3]
b : [t3]
c : [t3]
d : []}
But with proper sqlalchemy Column, Label, Table objects etc. Seems like this
should be possible, but I'm not familiar enough with traversing the trees.
the join thing plus being able to reduce below the subquery is awkward,
but perhaps this will illustrate some ideas:
from sqlalchemy import func, table, column, select
t1 = table('t1', column('a'))
t3 = table('t3', column('a'), column('b'), column('c'))
t2 = select([t3]).alias('t2')
stmt = select([t1.c.a.label('aa'), t2.c.b, t3.c.c, func.now().label('d')]).\
select_from(
t1.join(t2, t1.c.a == t2.c.a)
)
print stmt
from sqlalchemy.sql import expression
from sqlalchemy.sql import visitors
import collections
import operator
result = collections.defaultdict(list)
stack = []
all_columns = set()
all_comparisons = []
def search_for_comparisons(binary):
if binary.operator is operator.eq and \
isinstance(binary.left, expression.ColumnClause) and \
isinstance(binary.right, expression.ColumnClause):
all_comparisons.append((binary.left, binary.right))
visitors.traverse(
stmt, {}, {
"column": lambda col: all_columns.add(col),
"binary": search_for_comparisons
})
def _root_col_table(col):
return list(col.base_columns)[0].table
for col in stmt.inner_columns:
list_ = result[col.name]
if isinstance(col, expression.Label):
col = col.element
if hasattr(col, 'table'):
list_.append(_root_col_table(col).name)
for pair in all_comparisons:
if col is pair[0]:
list_.append(_root_col_table(pair[1]).name)
elif col is pair[1]:
list_.append(_root_col_table(pair[2]).name)
print result
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
