commit or rollback transaction with errors
on engine.begin().
+- mssql
+ - [feature] Added support for MSSQL INSERT,
+ UPDATE, and DELETE table hints, using
+ new with_hint() method on UpdateBase.
+ [ticket:2430]
+
- mysql
- [feature] Added support for the "isolation_level"
parameter to all MySQL dialects. Thanks
:show-inheritance:
.. autoclass:: Update
- :members: where, values
+ :members:
:show-inheritance:
.. autoclass:: UpdateBase
- :members: params, bind, returning
+ :members:
:show-inheritance:
.. autoclass:: ValuesBase
def get_from_hint_text(self, table, text):
return text
+ def get_crud_hint_text(self, table, text):
+ return text
+
def limit_clause(self, select):
# Limit in mssql is after the select keyword
return ""
return ', '.join(t._compiler_dispatch(self, asfrom=True, **kw)
for t in [from_table] + list(extra_froms))
- def update_from_clause(self, update_stmt, from_table, extra_froms, **kw):
+ def update_from_clause(self, update_stmt, from_table,
+ extra_froms, from_hints, **kw):
return None
def get_from_hint_text(self, table, text):
return None
+ def get_crud_hint_text(self, table, text):
+ return None
+
def visit_select(self, select, asfrom=False, parens=True,
iswrapper=False, fromhints=None,
compound_index=1, **kwargs):
text = "INSERT"
+
prefixes = [self.process(x) for x in insert_stmt._prefixes]
if prefixes:
text += " " + " ".join(prefixes)
text += " INTO " + preparer.format_table(insert_stmt.table)
+ if insert_stmt._hints:
+ dialect_hints = dict([
+ (table, hint_text)
+ for (table, dialect), hint_text in
+ insert_stmt._hints.items()
+ if dialect in ('*', self.dialect.name)
+ ])
+ if insert_stmt.table in dialect_hints:
+ text += " " + self.get_crud_hint_text(
+ insert_stmt.table,
+ dialect_hints[insert_stmt.table]
+ )
+
if colparams or not supports_default_values:
text += " (%s)" % ', '.join([preparer.format_column(c[0])
for c in colparams])
extra_froms, **kw):
"""Provide a hook to override the initial table clause
in an UPDATE statement.
-
+
MySQL overrides this.
"""
return self.preparer.format_table(from_table)
- def update_from_clause(self, update_stmt, from_table, extra_froms, **kw):
+ def update_from_clause(self, update_stmt,
+ from_table, extra_froms,
+ from_hints,
+ **kw):
"""Provide a hook to override the generation of an
UPDATE..FROM clause.
-
+
MySQL overrides this.
"""
return "FROM " + ', '.join(
- t._compiler_dispatch(self, asfrom=True, **kw)
+ t._compiler_dispatch(self, asfrom=True,
+ fromhints=from_hints, **kw)
for t in extra_froms)
def visit_update(self, update_stmt, **kw):
update_stmt.table,
extra_froms, **kw)
+ if update_stmt._hints:
+ dialect_hints = dict([
+ (table, hint_text)
+ for (table, dialect), hint_text in
+ update_stmt._hints.items()
+ if dialect in ('*', self.dialect.name)
+ ])
+ if update_stmt.table in dialect_hints:
+ text += " " + self.get_crud_hint_text(
+ update_stmt.table,
+ dialect_hints[update_stmt.table]
+ )
+ else:
+ dialect_hints = None
+
text += ' SET '
if extra_froms and self.render_table_with_column_in_update_from:
text += ', '.join(
extra_from_text = self.update_from_clause(
update_stmt,
update_stmt.table,
- extra_froms, **kw)
+ extra_froms,
+ dialect_hints, **kw)
if extra_from_text:
text += " " + extra_from_text
text = "DELETE FROM " + self.preparer.format_table(delete_stmt.table)
+ if delete_stmt._hints:
+ dialect_hints = dict([
+ (table, hint_text)
+ for (table, dialect), hint_text in
+ delete_stmt._hints.items()
+ 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]
+ )
+ else:
+ dialect_hints = None
+
if delete_stmt._returning:
self.returning = delete_stmt._returning
if self.returning_precedes_values:
The text of the hint is rendered in the appropriate
location for the database backend in use, relative
to the given :class:`.Table` or :class:`.Alias` passed as the
- *selectable* argument. The dialect implementation
+ ``selectable`` argument. The dialect implementation
typically uses Python string substitution syntax
with the token ``%(name)s`` to render the name of
the table or alias. E.g. when using Oracle, the
_execution_options = \
Executable._execution_options.union({'autocommit': True})
kwargs = util.immutabledict()
+ _hints = util.immutabledict()
def _process_colparams(self, parameters):
if isinstance(parameters, (list, tuple)):
"""
self._returning = cols
+ @_generative
+ def with_hint(self, text, selectable=None, dialect_name="*"):
+ """Add a table hint for a single table to this
+ INSERT/UPDATE/DELETE statement.
+
+ .. note::
+
+ :meth:`.UpdateBase.with_hint` currently applies only to
+ Microsoft SQL Server. For MySQL INSERT hints, use
+ :meth:`.Insert.prefix_with`. UPDATE/DELETE hints for
+ MySQL will be added in a future release.
+
+ The text of the hint is rendered in the appropriate
+ location for the database backend in use, relative
+ to the :class:`.Table` that is the subject of this
+ statement, or optionally to that of the given
+ :class:`.Table` passed as the ``selectable`` argument.
+
+ The ``dialect_name`` option will limit the rendering of a particular
+ hint to a particular backend. Such as, to add a hint
+ that only takes effect for SQL Server::
+
+ mytable.insert().with_hint("WITH (PAGLOCK)", dialect_name="mssql")
+
+ New in 0.7.6.
+
+ :param text: Text of the hint.
+ :param selectable: optional :class:`.Table` that specifies
+ an element of the FROM clause within an UPDATE or DELETE
+ to be the subject of the hint - applies only to certain backends.
+ :param dialect_name: defaults to ``*``, if specified as the name
+ of a particular dialect, will apply these hints only when
+ that dialect is in use.
+ """
+ if selectable is None:
+ selectable = self.table
+
+ self._hints = self._hints.union({(selectable, dialect_name):text})
+
class ValuesBase(UpdateBase):
"""Supplies support for :meth:`.ValuesBase.values` to INSERT and UPDATE constructs."""
'n WHERE sometable.somecolumn = '
':somecolumn_1', dict(somecolumn=10))
+ def test_insert_hint(self):
+ t = table('sometable', column('somecolumn'))
+ for targ in (None, t):
+ for darg in ("*", "mssql"):
+ self.assert_compile(
+ t.insert().
+ values(somecolumn="x").
+ with_hint("WITH (PAGLOCK)",
+ selectable=targ,
+ dialect_name=darg),
+ "INSERT INTO sometable WITH (PAGLOCK) "
+ "(somecolumn) VALUES (:somecolumn)"
+ )
+
+ def test_update_hint(self):
+ t = table('sometable', column('somecolumn'))
+ for targ in (None, t):
+ for darg in ("*", "mssql"):
+ self.assert_compile(
+ t.update().where(t.c.somecolumn=="q").
+ values(somecolumn="x").
+ with_hint("WITH (PAGLOCK)",
+ selectable=targ,
+ dialect_name=darg),
+ "UPDATE sometable WITH (PAGLOCK) "
+ "SET somecolumn=:somecolumn "
+ "WHERE sometable.somecolumn = :somecolumn_1"
+ )
+
+ def test_update_exclude_hint(self):
+ t = table('sometable', column('somecolumn'))
+ self.assert_compile(
+ t.update().where(t.c.somecolumn=="q").
+ values(somecolumn="x").
+ with_hint("XYZ", "mysql"),
+ "UPDATE sometable SET somecolumn=:somecolumn "
+ "WHERE sometable.somecolumn = :somecolumn_1"
+ )
+
+ def test_delete_hint(self):
+ t = table('sometable', column('somecolumn'))
+ for targ in (None, t):
+ for darg in ("*", "mssql"):
+ self.assert_compile(
+ t.delete().where(t.c.somecolumn=="q").
+ with_hint("WITH (PAGLOCK)",
+ selectable=targ,
+ dialect_name=darg),
+ "DELETE FROM sometable WITH (PAGLOCK) "
+ "WHERE sometable.somecolumn = :somecolumn_1"
+ )
+
+ def test_delete_exclude_hint(self):
+ t = table('sometable', column('somecolumn'))
+ self.assert_compile(
+ t.delete().\
+ where(t.c.somecolumn=="q").\
+ with_hint("XYZ", dialect_name="mysql"),
+ "DELETE FROM sometable WHERE "
+ "sometable.somecolumn = :somecolumn_1"
+ )
+
+ def test_update_from_hint(self):
+ t = table('sometable', column('somecolumn'))
+ t2 = table('othertable', column('somecolumn'))
+ for darg in ("*", "mssql"):
+ self.assert_compile(
+ t.update().where(t.c.somecolumn==t2.c.somecolumn).
+ values(somecolumn="x").
+ with_hint("WITH (PAGLOCK)",
+ selectable=t2,
+ dialect_name=darg),
+ "UPDATE sometable SET somecolumn=:somecolumn "
+ "FROM othertable WITH (PAGLOCK) "
+ "WHERE sometable.somecolumn = othertable.somecolumn"
+ )
+
+ # TODO: not supported yet.
+ #def test_delete_from_hint(self):
+ # t = table('sometable', column('somecolumn'))
+ # t2 = table('othertable', column('somecolumn'))
+ # for darg in ("*", "mssql"):
+ # self.assert_compile(
+ # t.delete().where(t.c.somecolumn==t2.c.somecolumn).
+ # with_hint("WITH (PAGLOCK)",
+ # selectable=t2,
+ # dialect_name=darg),
+ # ""
+ # )
+
# TODO: should this be for *all* MS-SQL dialects ?
def test_mxodbc_binds(self):
"""mxodbc uses MS-SQL native binds, which aren't allowed in
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
