0.1.4
-- create_engine() now uses genericized parameters; host/hostname, db/dbname/database, password/passwd, etc. for all engine connections. makes
-engine URIs much more "universal"
+- create_engine() now uses genericized parameters; host/hostname, db/dbname/database, password/passwd, etc. for all engine connections. makes engine URIs much more "universal"
+- added support for SELECT statements embedded into a column clause, using the flag
+"scalar=True"
- another overhaul to EagerLoading when used in conjunction with mappers that
inherit; improvements to eager loads figuring out their aliased queries
correctly, also relations set up against a mapper with inherited mappers will
<&|doclib.myt:item, name="primitives", description="Non-engine primitives: TableClause/ColumnClause" &>
- <p>TableClause and ColumnClause are "primitive" versions of the Table and Column objects which dont use engines at all; applications that just want to generate SQL strings but not directly communicate with a database can use TableClause and ColumnClause objects, which are non-singleton and serve as the "lexical" base class of Table and Column:</p>
+ <p>TableClause and ColumnClause are "primitive" versions of the Table and Column objects which dont use engines at all; applications that just want to generate SQL strings but not directly communicate with a database can use TableClause and ColumnClause objects (accessed via 'table' and 'column'), which are non-singleton and serve as the "lexical" base class of Table and Column:</p>
<&|formatting.myt:code&>
- tab1 = TableClause('table1',
- ColumnClause('id'),
- ColumnClause('name'))
+ tab1 = table('table1',
+ column('id'),
+ column('name'))
- tab2 = TableClause('table2',
- ColumnClause('id'),
- ColumnClause('email'))
+ tab2 = table('table2',
+ column('id'),
+ column('email'))
tab1.select(tab1.c.name == 'foo')
</&>
<P>The sql package supports embedding select statements into other select statements as the criterion in a WHERE condition, or as one of the "selectable" objects in the FROM list of the query. It does not at the moment directly support embedding a SELECT statement as one of the column criterion for a statement, although this can be achieved via direct text insertion, described later.</p>
+ <&|doclib.myt:item, name="scalar", description="Scalar Column Queries"&>
+ <p>Subqueries can be used in the column clause of a select statement by specifying the <span class="codeline">scalar=True</span> flag:</p>
+ <&|formatting.myt:code &>
+<&formatting.myt:poplink&>select([table2.c.col1, table2.c.col2, select([table1.c.col1], table1.c.col2==7, scalar=True)])
+<&|formatting.myt:codepopper, link="sql" &>
+SELECT table2.col1, table2.col2,
+(SELECT table1.col1 AS col1 FROM table1 WHERE col2=:table1_col2)
+FROM table2
+{'table1_col2': 7}
+</&>
+ </&>
+ </&>
<&|doclib.myt:item, name="correlated", description="Correlated Subqueries" &>
- <P>When a select object is embedded inside of another select object, and both objects reference the same table, SQLAlchemy makes the assumption that the table should be correlated from the child query to the parent query.
+ <P>When a select object is embedded inside of another select object, and both objects reference the same table, SQLAlchemy makes the assumption that the table should be correlated from the child query to the parent query. To disable this behavior, specify the flag <span class="codeline">correlate=False</span> to the Select statement.</p>
<&|formatting.myt:code &>
# make an alias of a regular select.
s = select([addresses.c.street], addresses.c.user_id==users.c.user_id).alias('s')
self.select_stack.append(select)
for c in select._raw_columns:
- if c.is_selectable():
+ # TODO: make this polymorphic?
+ if isinstance(c, sql.Select) and c._scalar:
+ c.accept_visitor(self)
+ inner_columns[self.get_str(c)] = c
+ elif c.is_selectable():
for co in c.columns:
if select.use_labels:
l = co.label(co._label)
"""
return Select(columns, whereclause = whereclause, from_obj = from_obj, **kwargs)
+def subquery(alias, *args, **kwargs):
+ return Select(*args, **kwargs).alias(alias)
+
+
def insert(table, values = None, **kwargs):
"""returns an INSERT clause element.
def alias(*args, **params):
return Alias(*args, **params)
-def subquery(alias, *args, **params):
- return Alias(Select(*args, **params), alias)
-
def literal(value, type=None):
"""returns a literal clause, bound to a bind parameter.
"""base class for elements of a programmatically constructed SQL expression."""
def _get_from_objects(self):
"""returns objects represented in this ClauseElement that should be added to the
- FROM list of a query."""
+ FROM list of a query, when this ClauseElement is placed in the column clause of a Select
+ statement."""
raise NotImplementedError(repr(self))
def _process_from_dict(self, data, asfrom):
"""given a dictionary attached to a Select object, places the appropriate
return []
def alias(self, name=None):
- return self.select(use_labels=True).alias(name)
+ """creates a Select out of this Join clause and returns an Alias of it. The Select is not correlating."""
+ return self.select(use_labels=True, correlate=False).alias(name)
def _process_from_dict(self, data, asfrom):
for f in self.onclause._get_from_objects():
data[f.id] = f
def select(self, whereclauses = None, **params):
return select([self], whereclauses, **params)
def _get_from_objects(self):
- if self.is_where:
+ if self.is_where or self._scalar:
return []
else:
return [self]
class Select(SelectBaseMixin, FromClause):
"""represents a SELECT statement, with appendable clauses, as well as
the ability to execute itself and return a result set."""
- def __init__(self, columns=None, whereclause = None, from_obj = [], order_by = None, group_by=None, having=None, use_labels = False, distinct=False, engine = None, limit=None, offset=None, correlate=False):
+ def __init__(self, columns=None, whereclause = None, from_obj = [], order_by = None, group_by=None, having=None, use_labels = False, distinct=False, engine = None, limit=None, offset=None, scalar=False, correlate=True):
self._froms = util.OrderedDict()
self.use_labels = use_labels
self.id = "Select(%d)" % id(self)
self.oid_column = None
self.limit = limit
self.offset = offset
+
+ # indicates that this select statement should not expand its columns
+ # into the column clause of an enclosing select, and should instead
+ # act like a single scalar column
+ self._scalar = scalar
+
+ # indicates if this select statement, as a subquery, should correlate
+ # its FROM clause to that of an enclosing select statement
self.correlate = correlate
# indicates if this select statement is a subquery inside another query
self.issubquery = False
+
# indicates if this select statement is a subquery as a criterion
# inside of a WHERE clause
self.is_where = False
-
+
self.distinct = distinct
self._text = None
self._raw_columns = []
select.is_where = self.is_where
select.issubquery = True
select.parens = True
- if not self.is_where and not select.correlate:
+ if not select.correlate:
return
if getattr(select, '_correlated', None) is None:
select._correlated = self.select._froms
self._raw_columns.append(column)
+ # if the column is a Select statement itself,
+ # accept visitor
+ column.accept_visitor(self._correlator)
+
+ # visit the FROM objects of the column looking for more Selects
for f in column._get_from_objects():
f.accept_visitor(self._correlator)
column._process_from_dict(self._froms, False)
return column._make_proxy(self, name=column._label)
else:
return column._make_proxy(self, name=column.name)
-
def append_whereclause(self, whereclause):
self._append_condition('whereclause', whereclause)
def append_having(self, having):
self.runtest(select([table1, table2]), "SELECT mytable.myid, mytable.name, mytable.description, myothertable.otherid, \
myothertable.othername FROM mytable, myothertable")
- def testsubquery(self):
+ def testselectselect(self):
+ """tests placing select statements in the column clause of another select, for the
+ purposes of selecting from the exported columns of that select."""
s = select([table1], table1.c.name == 'jack')
- print [key for key in s.c.keys()]
+ #print [key for key in s.c.keys()]
self.runtest(
select(
[s],
"SELECT myid, name, description FROM (SELECT mytable.myid AS myid, mytable.name AS name, mytable.description AS description FROM mytable)"
)
- sq = subquery(
- 'sq',
+ sq = select(
[table1],
- )
+ ).alias('sq')
self.runtest(
sq.select(sq.c.myid == 7),
(SELECT mytable.myid AS myid, mytable.name AS name, mytable.description AS description FROM mytable) AS sq WHERE sq.myid = :sq_myid"
)
- sq = subquery(
- 'sq',
+ sq = select(
[table1, table2],
and_(table1.c.myid ==7, table2.c.otherid==table1.c.myid),
use_labels = True
- )
+ ).alias('sq')
sqstring = "SELECT mytable.myid AS mytable_myid, mytable.name AS mytable_name, \
mytable.description AS mytable_description, myothertable.otherid AS myothertable_otherid, \
self.runtest(sq.select(), "SELECT sq.mytable_myid, sq.mytable_name, sq.mytable_description, sq.myothertable_otherid, \
sq.myothertable_othername FROM (" + sqstring + ") AS sq")
- sq2 = subquery(
- 'sq2',
+ sq2 = select(
[sq],
use_labels = True
- )
+ ).alias('sq2')
self.runtest(sq2.select(), "SELECT sq2.sq_mytable_myid, sq2.sq_mytable_name, sq2.sq_mytable_description, \
sq2.sq_myothertable_otherid, sq2.sq_myothertable_othername FROM \
(SELECT sq.mytable_myid AS sq_mytable_myid, sq.mytable_name AS sq_mytable_name, \
sq.mytable_description AS sq_mytable_description, sq.myothertable_otherid AS sq_myothertable_otherid, \
sq.myothertable_othername AS sq_myothertable_othername FROM (" + sqstring + ") AS sq) AS sq2")
+
+ def testwheresubquery(self):
+ self.runtest(
+ table1.select(table1.c.myid == select([table2.c.otherid], table1.c.name == table2.c.othername)),
+ "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid = (SELECT myothertable.otherid AS otherid FROM myothertable WHERE mytable.name = myothertable.othername)"
+ )
+
+ self.runtest(
+ table1.select(exists([1], table2.c.otherid == table1.c.myid)),
+ "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE EXISTS (SELECT 1 FROM myothertable WHERE myothertable.otherid = mytable.myid)"
+ )
+
+ talias = table1.alias('ta')
+ s = subquery('sq2', [talias], exists([1], table2.c.otherid == talias.c.myid))
+ self.runtest(
+ select([s, table1])
+ ,"SELECT sq2.myid, sq2.name, sq2.description, mytable.myid, mytable.name, mytable.description FROM (SELECT ta.myid AS myid, ta.name AS name, ta.description AS description FROM mytable AS ta WHERE EXISTS (SELECT 1 FROM myothertable WHERE myothertable.otherid = ta.myid)) AS sq2, mytable")
+
+ s = select([addresses.c.street], addresses.c.user_id==users.c.user_id, correlate=True).alias('s')
+ self.runtest(
+ select([users, s.c.street], from_obj=[s]),
+ """SELECT users.user_id, users.user_name, users.password, s.street FROM users, (SELECT addresses.street AS street FROM addresses WHERE addresses.user_id = users.user_id) AS s""")
+
+ def testcolumnsubquery(self):
+ s = select([table1.c.myid], scalar=True, correlate=False)
+ self.runtest(select([table1, s]), "SELECT mytable.myid, mytable.name, mytable.description, (SELECT mytable.myid AS myid FROM mytable) FROM mytable")
+
+ s = select([table1.c.myid], scalar=True)
+ self.runtest(select([table2, s]), "SELECT myothertable.otherid, myothertable.othername, (SELECT mytable.myid AS myid FROM mytable) FROM myothertable")
+
+ zips = table('zips',
+ column('zipcode'),
+ column('latitude'),
+ column('longitude'),
+ )
+ places = table('places',
+ column('id'),
+ column('nm')
+ )
+ zip = '12345'
+ qlat = select([zips.c.latitude], zips.c.zipcode == zip, scalar=True, correlate=False)
+ qlng = select([zips.c.longitude], zips.c.zipcode == zip, scalar=True, correlate=False)
+
+ q = select([places.c.id, places.c.nm, zips.c.zipcode, func.latlondist(qlat, qlng).label('dist')],
+ zips.c.zipcode==zip,
+ order_by = ['dist', places.c.nm]
+ )
+
+ self.runtest(q,"SELECT places.id, places.nm, zips.zipcode, latlondist((SELECT zips.latitude AS latitude FROM zips WHERE zips.zipcode = :zips_zipcode_1), (SELECT zips.longitude AS longitude FROM zips WHERE zips.zipcode = :zips_zipcode_2)) AS dist FROM places, zips WHERE zips.zipcode = :zips_zipcode ORDER BY dist, places.nm")
+ zalias = zips.alias('main_zip')
+ qlat = select([zips.c.latitude], zips.c.zipcode == zalias.c.zipcode, scalar=True)
+ qlng = select([zips.c.longitude], zips.c.zipcode == zalias.c.zipcode, scalar=True)
+ q = select([places.c.id, places.c.nm, zalias.c.zipcode, func.latlondist(qlat, qlng).label('dist')],
+ order_by = ['dist', places.c.nm]
+ )
+ self.runtest(q, "SELECT places.id, places.nm, main_zip.zipcode, latlondist((SELECT zips.latitude AS latitude FROM zips WHERE zips.zipcode = main_zip.zipcode), (SELECT zips.longitude AS longitude FROM zips WHERE zips.zipcode = main_zip.zipcode)) AS dist FROM places, zips AS main_zip ORDER BY dist, places.nm")
+
def testand(self):
self.runtest(
select(['*'], and_(table1.c.myid == 12, table1.c.name=='asdf', table2.c.othername == 'foo', "sysdate() = today()")),
c = s.compile(parameters = {'test' : 7}, engine=db)
self.assert_(c.get_params() == {'test' : 7})
- def testcorrelatedsubquery(self):
- self.runtest(
- table1.select(table1.c.myid == select([table2.c.otherid], table1.c.name == table2.c.othername)),
- "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid = (SELECT myothertable.otherid AS otherid FROM myothertable WHERE mytable.name = myothertable.othername)"
- )
-
- self.runtest(
- table1.select(exists([1], table2.c.otherid == table1.c.myid)),
- "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE EXISTS (SELECT 1 FROM myothertable WHERE myothertable.otherid = mytable.myid)"
- )
-
- talias = table1.alias('ta')
- s = subquery('sq2', [talias], exists([1], table2.c.otherid == talias.c.myid))
- self.runtest(
- select([s, table1])
- ,"SELECT sq2.myid, sq2.name, sq2.description, mytable.myid, mytable.name, mytable.description FROM (SELECT ta.myid AS myid, ta.name AS name, ta.description AS description FROM mytable AS ta WHERE EXISTS (SELECT 1 FROM myothertable WHERE myothertable.otherid = ta.myid)) AS sq2, mytable")
-
- s = select([addresses.c.street], addresses.c.user_id==users.c.user_id, correlate=True).alias('s')
- self.runtest(
- select([users, s.c.street], from_obj=[s]),
- """SELECT users.user_id, users.user_name, users.password, s.street FROM users, (SELECT addresses.street AS street FROM addresses WHERE addresses.user_id = users.user_id) AS s""")
def testin(self):
self.runtest(select([table1], table1.c.myid.in_(1, 2, 3)),
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
