ForeignKeyConstraint, courtesy Ryan Kelly.
[ticket:2502]
- [feature] Added support for DELETE and UPDATE from
an alias of a table, which would assumedly
be related to itself elsewhere in the query,
courtesy Ryan Kelly. [ticket:2507]
- [feature] Added support for the Postgresql ONLY
keyword, which can appear corresponding to a
table in a SELECT, UPDATE, or DELETE statement.
The phrase is established using with_hint().
Courtesy Ryan Kelly [ticket:2506]
ForeignKeyConstraint, courtesy Ryan Kelly.
[ticket:2502]
+ - [feature] Added support for DELETE and UPDATE from
+ an alias of a table, which would assumedly
+ be related to itself elsewhere in the query,
+ courtesy Ryan Kelly. [ticket:2507]
+
- [feature] select() features a correlate_except()
method, auto correlates all selectables except those
passed.
performance of bind/result processing.
[ticket:2441]
+ - [feature] Added support for the Postgresql ONLY
+ keyword, which can appear corresponding to a
+ table in a SELECT, UPDATE, or DELETE statement.
+ The phrase is established using with_hint().
+ Courtesy Ryan Kelly [ticket:2506]
+
+
- mysql
- [bug] Dialect no longer emits expensive server
collations query, as well as server casing,
else:
return None
- def visit_table(self, table, mssql_aliased=False, **kwargs):
- if mssql_aliased is table:
+ def visit_table(self, table, mssql_aliased=False, iscrud=False, **kwargs):
+ if mssql_aliased is table or iscrud:
return super(MSSQLCompiler, self).visit_table(table, **kwargs)
# alias schema-qualified tables
where(table.c.name=='foo')
print result.fetchall()
+FROM ONLY ...
+------------------------
+
+The dialect supports PostgreSQL's ONLY keyword for targeting only a particular
+table in an inheritance hierarchy. This can be used to produce the
+``SELECT ... FROM ONLY``, ``UPDATE ONLY ...``, and ``DELETE FROM ONLY ...``
+syntaxes. It uses SQLAlchemy's hints mechanism:
+
+ # SELECT ... FROM ONLY ...
+ result = table.select().with_hint(table, 'ONLY', 'postgresql')
+ print result.fetchall()
+
+ # UPDATE ONLY ...
+ table.update(values=dict(foo='bar')).with_hint('ONLY',
+ dialect_name='postgresql')
+
+ # DELETE FROM ONLY ...
+ table.delete().with_hint('ONLY', dialect_name='postgresql')
.. _postgresql_indexes:
text += " OFFSET " + self.process(sql.literal(select._offset))
return text
+ def format_from_hint_text(self, sqltext, table, hint, iscrud):
+ if hint.upper() != 'ONLY':
+ raise exc.CompileError("Unrecognized hint: %r" % hint)
+ return "ONLY " + sqltext
+
def get_select_precolumns(self, select):
if select._distinct is not False:
if select._distinct is True:
return text
def visit_alias(self, alias, asfrom=False, ashint=False,
+ iscrud=False,
fromhints=None, **kwargs):
if asfrom or ashint:
if isinstance(alias.name, sql._truncated_label):
self.preparer.format_alias(alias, alias_name)
if fromhints and alias in fromhints:
- hinttext = self.get_from_hint_text(alias, fromhints[alias])
- if hinttext:
- ret += " " + hinttext
+ ret = self.format_from_hint_text(ret, alias,
+ fromhints[alias], iscrud)
return ret
else:
else:
return column
+ def format_from_hint_text(self, sqltext, table, hint, iscrud):
+ hinttext = self.get_from_hint_text(table, hint)
+ if hinttext:
+ sqltext += " " + hinttext
+ return sqltext
+
def get_select_hint_text(self, byfroms):
return None
text += " OFFSET " + self.process(sql.literal(select._offset))
return text
- def visit_table(self, table, asfrom=False, ashint=False,
+ def visit_table(self, table, asfrom=False, iscrud=False, ashint=False,
fromhints=None, **kwargs):
if asfrom or ashint:
if getattr(table, "schema", None):
else:
ret = self.preparer.quote(table.name, table.quote)
if fromhints and table in fromhints:
- hinttext = self.get_from_hint_text(table, fromhints[table])
- if hinttext:
- ret += " " + hinttext
+ ret = self.format_from_hint_text(ret, table,
+ fromhints[table], iscrud)
return ret
else:
return ""
if prefixes:
text += " " + " ".join(prefixes)
- text += " INTO " + preparer.format_table(insert_stmt.table)
+ text += " INTO "
+ table_text = preparer.format_table(insert_stmt.table)
if insert_stmt._hints:
dialect_hints = dict([
if dialect in ('*', self.dialect.name)
])
if insert_stmt.table in dialect_hints:
- text += " " + self.get_crud_hint_text(
+ table_text = self.format_from_hint_text(
+ table_text,
insert_stmt.table,
- dialect_hints[insert_stmt.table]
+ dialect_hints[insert_stmt.table],
+ True
)
+ text += table_text
+
if colparams or not supports_default_values:
text += " (%s)" % ', '.join([preparer.format_column(c[0])
for c in colparams])
MySQL overrides this.
"""
- return self.preparer.format_table(from_table)
+ return from_table._compiler_dispatch(self, asfrom=True,
+ iscrud=True, **kw)
def update_from_clause(self, update_stmt,
from_table, extra_froms,
colparams = self._get_colparams(update_stmt, extra_froms)
- text = "UPDATE " + self.update_tables_clause(
- update_stmt,
- update_stmt.table,
- extra_froms, **kw)
+ text = "UPDATE "
+ table_text = self.update_tables_clause(update_stmt, update_stmt.table,
+ extra_froms, **kw)
if update_stmt._hints:
dialect_hints = dict([
if dialect in ('*', self.dialect.name)
])
if update_stmt.table in dialect_hints:
- text += " " + self.get_crud_hint_text(
+ table_text = self.format_from_hint_text(
+ table_text,
update_stmt.table,
- dialect_hints[update_stmt.table]
+ dialect_hints[update_stmt.table],
+ True
)
else:
dialect_hints = None
+ text += table_text
+
text += ' SET '
if extra_froms and self.render_table_with_column_in_update_from:
text += ', '.join(
self.stack.append({'from': set([delete_stmt.table])})
self.isdelete = True
- text = "DELETE FROM " + self.preparer.format_table(delete_stmt.table)
+ text = "DELETE FROM "
+ table_text = delete_stmt.table._compiler_dispatch(self,
+ asfrom=True, iscrud=True)
if delete_stmt._hints:
dialect_hints = dict([
if dialect in ('*', self.dialect.name)
])
if delete_stmt.table in dialect_hints:
- text += " " + self.get_crud_hint_text(
- delete_stmt.table,
- dialect_hints[delete_stmt.table]
+ table_text = self.format_from_hint_text(
+ table_text,
+ delete_stmt.table,
+ dialect_hints[delete_stmt.table],
+ True
)
+
else:
dialect_hints = None
+ text += table_text
+
if delete_stmt._returning:
self.returning = delete_stmt._returning
if self.returning_precedes_values:
delete_stmt, delete_stmt._returning)
if delete_stmt._whereclause is not None:
- text += " WHERE " + self.process(delete_stmt._whereclause)
+ text += " WHERE "
+ text += delete_stmt._whereclause._compiler_dispatch(self)
if self.returning and not self.returning_precedes_values:
text += " " + self.returning_clause(
self.assert_compile(x,
'''SELECT pg_table.col1, pg_table."variadic" FROM pg_table''')
+ def test_from_only(self):
+ m = MetaData()
+ tbl1 = Table('testtbl1', m, Column('id', Integer))
+ tbl2 = Table('testtbl2', m, Column('id', Integer))
+
+ stmt = tbl1.select().with_hint(tbl1, 'ONLY', 'postgresql')
+ expected = 'SELECT testtbl1.id FROM ONLY testtbl1'
+ self.assert_compile(stmt, expected)
+
+ talias1 = tbl1.alias('foo')
+ stmt = talias1.select().with_hint(talias1, 'ONLY', 'postgresql')
+ expected = 'SELECT foo.id FROM ONLY testtbl1 AS foo'
+ self.assert_compile(stmt, expected)
+
+ stmt = select([tbl1, tbl2]).with_hint(tbl1, 'ONLY', 'postgresql')
+ expected = ('SELECT testtbl1.id, testtbl2.id FROM ONLY testtbl1, '
+ 'testtbl2')
+ self.assert_compile(stmt, expected)
+
+ stmt = select([tbl1, tbl2]).with_hint(tbl2, 'ONLY', 'postgresql')
+ expected = ('SELECT testtbl1.id, testtbl2.id FROM testtbl1, ONLY '
+ 'testtbl2')
+ self.assert_compile(stmt, expected)
+
+ stmt = select([tbl1, tbl2])
+ stmt = stmt.with_hint(tbl1, 'ONLY', 'postgresql')
+ stmt = stmt.with_hint(tbl2, 'ONLY', 'postgresql')
+ expected = ('SELECT testtbl1.id, testtbl2.id FROM ONLY testtbl1, '
+ 'ONLY testtbl2')
+ self.assert_compile(stmt, expected)
+
+ stmt = update(tbl1, values=dict(id=1))
+ stmt = stmt.with_hint('ONLY', dialect_name='postgresql')
+ expected = 'UPDATE ONLY testtbl1 SET id=%(id)s'
+ self.assert_compile(stmt, expected)
+
+ stmt = delete(tbl1).with_hint('ONLY', selectable=tbl1, dialect_name='postgresql')
+ expected = 'DELETE FROM ONLY testtbl1'
+ self.assert_compile(stmt, expected)
+
+ tbl3 = Table('testtbl3', m, Column('id', Integer), schema='testschema')
+ stmt = tbl3.select().with_hint(tbl3, 'ONLY', 'postgresql')
+ expected = 'SELECT testschema.testtbl3.id FROM ONLY testschema.testtbl3'
+ self.assert_compile(stmt, expected)
+
+ assert_raises(
+ exc.CompileError,
+ tbl3.select().with_hint(tbl3, "FAKE", "postgresql").compile,
+ dialect=postgresql.dialect()
+ )
class FloatCoercionTest(fixtures.TablesTest, AssertsExecutionResults):
__only_on__ = 'postgresql'
"WHERE mytable.myid = hoho(:hoho_1) AND mytable.name = :param_2 || "
"mytable.name || :param_3")
+ def test_aliased_update(self):
+ talias1 = table1.alias('t1')
+ self.assert_compile(
+ update(talias1, talias1.c.myid == 7),
+ "UPDATE mytable AS t1 SET name=:name WHERE t1.myid = :myid_1",
+ params = {table1.c.name:'fred'})
+ self.assert_compile(
+ update(talias1, table1.c.myid == 7),
+ "UPDATE mytable AS t1 SET name=:name FROM "
+ "mytable WHERE mytable.myid = :myid_1",
+ params = {table1.c.name:'fred'})
+
def test_correlated_update(self):
# test against a straight text subquery
u = update(table1, values = {
"DELETE FROM mytable WHERE mytable.myid = :myid_1 "
"AND mytable.name = :name_1")
+ def test_aliased_delete(self):
+ talias1 = table1.alias('t1')
+ self.assert_compile(
+ delete(talias1).where(talias1.c.myid == 7),
+ "DELETE FROM mytable AS t1 WHERE t1.myid = :myid_1")
+
def test_correlated_delete(self):
# test a non-correlated WHERE clause
s = select([table2.c.othername], table2.c.otherid == 7)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
