Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP

Loading…

add psql FOR UPDATE OF functionality #42

Merged
merged 3 commits into from

2 participants

@mlassnig

Hello,

can you please assist me with this patch? I'm trying to add FOR UPDATE OF functionality, but I'm not sure I'm doing the right thing. Right now this relies on the user giving the right string in the of='tablename' part, but I would like this to be automatically extracted from a mapper somehow.

Thanks,
Mario

@zzzeek
Owner

the "of" part is PG specific so I'd rather make options like this more generic, e.g. query.with_lockmode() should accept *kw which it stores as "lockmode_opts", Select would have for_update_opts added to it which gets the *kw here, then the PG dialect would know what to do with that.

as far as automating this from the ORM, it's too broad to have "lockmode" universally set for a mapper, for one thing it would prevent the get() method from using the identity map as a positive "lockmode" here means a SELECT should definitely be emitted. If you want all SELECTs of a certain table to include some lockmode or option I'd use the before_execute event to receive a select(), inspect its ".froms" for tables, then add the option. Select itself should also have a "for_update()" method on it to make it easy to modify an existing select, I'm surprised this isn't there already.

@mlassnig

Hi Mike, thanks for you comments. It's not only PG, it's also for Oracle. But I see where you are heading. Let me have a try with your suggestions. (bear with me, I'm learning about the SQLAlchemy internals as I go along) :-)

@zzzeek
Owner

if FOR UPDATE OF is part of the SQL standard then we can add "of" as a kwarg.

@zzzeek
Owner

I found a little bit of mention of "FOR UPDATE OF" in the SQL standard, in terms of declaring a cursor which may be how it specifies this kind of thing, though it referred to a "column name list" as the target of the "OF". So I think we can call it "of". in any case it seems like the target of the "of" should be a list/tuple.

@mlassnig

I just investigated further. Oracle requires [table.column, table.column, ...], whereas postgresql requires [table, table, ...]. So that is something I will have to deal with in the dialect, I suppose.

@mlassnig

Hi, I think I got it. At least it does what I want :-) Testcases seem okay, it doesn't break anything else. Let me know if you want me to change something still. Cheers!

@mlassnig

Hi, is the patch fine like this? Do you need any more changes?

@zzzeek
Owner

can we turn select.for_update and query._lockmode into tuples themselves? e.g. ("update", {"of": [col1, col2, ...]}). then remove ".for_update_of" and "._lockmode_of".

@mlassnig

Are you sure this is a good idea? I just tried to do this, and suddenly I have to change all instances of for_update and lockmode all over the place, e.g, mysql/base, orm/{loading,query,session}, sql/compiler....

I mean, I can do it, but it suddenly doesn't feel as clean. What do you think?

@zzzeek
Owner

to be more specific, I don't want to change the public API here, so "lockmode" doesn't need to change in orm/session.py for example. It only changes specifically when it makes its way into the Select construct itself - where any information regarding "of" or other future arguments we might add are combined together with the given string to produce a tuple (so really, API isn't changing, we're just changing how the Select object represents the state internally. and the Query). the dialects themselves could perhaps be shielded from the format itself if we change the signature of "for_update_clause()" to receive the individual components of the select separately.

@zzzeek
Owner

or consider if we had a new kind of object, "LockmodeArgs()", that encapsulates information about a lockmode. That's essentially what we're doing here with the tuple, just without the class overhead. it's cleaner because as we add new elements and conditions to the concept of a "lockmode", we don't need to change APIs each time - the "LockmodeArgs()" object, disguised as a plain tuple, can handle the new changes wihtout impacting how the information is passed around, only how it is ultimately handled (And produced from public API arguments)

@mlassnig

Hi, I've added the LockmodeArgs. As you can see, I tried to keep as backwards compatible as possible. Let me know if it's okay, or if you want to change something still.

@zzzeek zzzeek commented on the diff
test/orm/test_lockmode.py
@@ -73,6 +73,23 @@ def test_postgres_update(self):
dialect=postgresql.dialect()
)
+ def test_postgres_update_of(self):
+ User = self.classes.User
+ sess = Session()
+ self.assert_compile(sess.query(User.id).with_lockmode('update', of=User.id),
+ "SELECT users.id AS users_id FROM users FOR UPDATE OF users",
+ dialect=postgresql.dialect()
+ )
+
+ def test_postgres_update_of_list(self):
+ User = self.classes.User
+ sess = Session()
+ self.assert_compile(sess.query(User.id).with_lockmode('update', of=[User.id, User.id, User.id]),
+ "SELECT users.id AS users_id FROM users FOR UPDATE OF users, users, users",
@zzzeek Owner
zzzeek added a note

do you actually want "users, users, users" here? I will likely add a set() so that tables are uniqued here.

Yes, it should be there multiple times. In a real scenario you would use different table.column entries, but for the test case it doesn't matter. (And both Ora and Psql accept multiple same entries anyway).

@zzzeek Owner
zzzeek added a note

Ok the idea here though is that if you have the same table multiple times, and you aren't displaying the column, just the table, the table should be de-duped on name. if i had with_lockmode(of=[User.id, User.name]), id want Oracle to say FOR UPDATE OF users.id, users.name but PG should say FOR UPDATE OF users.

in the most recent changesets you'll also see support for table aliases and select "adaptation" which is pretty important with the ORM.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
@zzzeek zzzeek merged commit e9aaf8e into from
@zzzeek
Owner

so i didn't actually plan for LockmodeArgs, I was suggesting a tuple still, but once I saw it I figured i'd go with it as this issue is more complicated anyway.

LockmodeArgs is mostly ForUpdateArgs in the selectable module, and both SelectBase and Query gain a with_for_update() method that acts consistently across both. for_update and with_lockmode() are deprecated. The only issue remaining is that the "lockmode" string is still accepted by Session.refresh(), I'm not sure how to deal with that, so at the moment it's just hanging around.

thanks for the pullreq!

@zzzeek
Owner

anyway please review the final product (it's quite different) and make sure it does everything we need it to, thanks !

@mlassnig

Thank you, that looks great indeed. It was also a great learning experience for me about the internals of SQLA. You might expect some more pull requests from me in the future :-)

@mlassnig mlassnig deleted the branch
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Commits on Nov 12, 2013
  1. @mlassnig
Commits on Nov 14, 2013
  1. @mlassnig

    added ORM support

    mlassnig authored
Commits on Nov 28, 2013
  1. @mlassnig

    added LockmodeArgs

    mlassnig authored
This page is out of date. Refresh to see the latest.
View
9 lib/sqlalchemy/dialects/mysql/base.py
@@ -1422,7 +1422,14 @@ def visit_join(self, join, asfrom=False, **kwargs):
self.process(join.onclause, **kwargs)))
def for_update_clause(self, select):
- if select.for_update == 'read':
+ # backwards compatibility
+ if isinstance(select.for_update, bool):
+ return ' FOR UPDATE'
+ elif isinstance(select.for_update, str):
+ if select.for_update == 'read':
+ return ' LOCK IN SHARE MODE'
+
+ if select.for_update.mode == 'read':
return ' LOCK IN SHARE MODE'
else:
return super(MySQLCompiler, self).for_update_clause(select)
View
24 lib/sqlalchemy/dialects/oracle/base.py
@@ -661,8 +661,28 @@ def limit_clause(self, select):
def for_update_clause(self, select):
if self.is_subquery():
return ""
- elif select.for_update == "nowait":
- return " FOR UPDATE NOWAIT"
+
+ tmp = ' FOR UPDATE'
+
+ # backwards compatibility
+ if isinstance(select.for_update, bool):
+ if select.for_update:
+ return tmp
+ elif isinstance(select.for_update, str):
+ if select.for_update == 'nowait':
+ return tmp + ' NOWAIT'
+ else:
+ return tmp
+
+ if isinstance(select.for_update.of, list):
+ tmp += ' OF ' + ', '.join(['.'.join(of) for of in select.for_update.of])
+ elif isinstance(select.for_update.of, tuple):
+ tmp += ' OF ' + '.'.join(select.for_update.of)
+
+ if select.for_update.mode == 'update_nowait':
+ return tmp + ' NOWAIT'
+ elif select.for_update.mode == 'update':
+ return tmp
else:
return super(OracleCompiler, self).for_update_clause(select)
View
36 lib/sqlalchemy/dialects/postgresql/base.py
@@ -230,7 +230,7 @@
"default", "deferrable", "desc", "distinct", "do", "else", "end",
"except", "false", "fetch", "for", "foreign", "from", "grant", "group",
"having", "in", "initially", "intersect", "into", "leading", "limit",
- "localtime", "localtimestamp", "new", "not", "null", "off", "offset",
+ "localtime", "localtimestamp", "new", "not", "null", "of", "off", "offset",
"old", "on", "only", "or", "order", "placing", "primary", "references",
"returning", "select", "session_user", "some", "symmetric", "table",
"then", "to", "trailing", "true", "union", "unique", "user", "using",
@@ -1014,12 +1014,34 @@ def get_select_precolumns(self, select):
return ""
def for_update_clause(self, select):
- if select.for_update == 'nowait':
- return " FOR UPDATE NOWAIT"
- elif select.for_update == 'read':
- return " FOR SHARE"
- elif select.for_update == 'read_nowait':
- return " FOR SHARE NOWAIT"
+
+ tmp = ' FOR UPDATE'
+
+ # backwards compatibility
+ if isinstance(select.for_update, bool):
+ return tmp
+ elif isinstance(select.for_update, str):
+ if select.for_update == 'nowait':
+ return tmp + ' NOWAIT'
+ elif select.for_update == 'read':
+ return ' FOR SHARE'
+ elif select.for_update == 'read_nowait':
+ return ' FOR SHARE NOWAIT'
+
+ if select.for_update.mode == 'read':
+ return ' FOR SHARE'
+ elif select.for_update.mode == 'read_nowait':
+ return ' FOR SHARE NOWAIT'
+
+ if isinstance(select.for_update.of, list):
+ tmp += ' OF ' + ', '.join([of[0] for of in select.for_update.of])
+ elif isinstance(select.for_update.of, tuple):
+ tmp += ' OF ' + select.for_update.of[0]
+
+ if select.for_update.mode == 'update_nowait':
+ return tmp + ' NOWAIT'
+ elif select.for_update.mode == 'update':
+ return tmp
else:
return super(PGCompiler, self).for_update_clause(select)
View
121 lib/sqlalchemy/orm/query.py
@@ -1124,33 +1124,42 @@ def execution_options(self, **kwargs):
self._execution_options = self._execution_options.union(kwargs)
@_generative()
- def with_lockmode(self, mode):
+ def with_lockmode(self, mode, of=None):
"""Return a new Query object with the specified locking mode.
:param mode: a string representing the desired locking mode. A
- corresponding value is passed to the ``for_update`` parameter of
- :meth:`~sqlalchemy.sql.expression.select` when the query is
- executed. Valid values are:
+ corresponding :meth:`~sqlalchemy.orm.query.LockmodeArgs` object
+ is passed to the ``for_update`` parameter of
+ :meth:`~sqlalchemy.sql.expression.select` when the
+ query is executed. Valid values are:
- ``'update'`` - passes ``for_update=True``, which translates to
- ``FOR UPDATE`` (standard SQL, supported by most dialects)
+ ``None`` - translates to no lockmode
- ``'update_nowait'`` - passes ``for_update='nowait'``, which
- translates to ``FOR UPDATE NOWAIT`` (supported by Oracle,
- PostgreSQL 8.1 upwards)
+ ``'update'`` - translates to ``FOR UPDATE``
+ (standard SQL, supported by most dialects)
- ``'read'`` - passes ``for_update='read'``, which translates to
- ``LOCK IN SHARE MODE`` (for MySQL), and ``FOR SHARE`` (for
- PostgreSQL)
+ ``'update_nowait'`` - translates to ``FOR UPDATE NOWAIT``
+ (supported by Oracle, PostgreSQL 8.1 upwards)
- ``'read_nowait'`` - passes ``for_update='read_nowait'``, which
- translates to ``FOR SHARE NOWAIT`` (supported by PostgreSQL).
+ ``'read'`` - translates to ``LOCK IN SHARE MODE`` (for MySQL),
+ and ``FOR SHARE`` (for PostgreSQL)
.. versionadded:: 0.7.7
``FOR SHARE`` and ``FOR SHARE NOWAIT`` (PostgreSQL).
+
+ :param of: either a column descriptor, or list of column
+ descriptors, representing the optional OF part of the
+ clause. This passes the descriptor to the
+ corresponding :meth:`~sqlalchemy.orm.query.LockmodeArgs` object,
+ and translates to ``FOR UPDATE OF table [NOWAIT]`` respectively
+ ``FOR UPDATE OF table, table [NOWAIT]`` (PostgreSQL), or
+ ``FOR UPDATE OF table.column [NOWAIT]`` respectively
+ ``FOR UPDATE OF table.column, table.column [NOWAIT]`` (Oracle).
+
+ .. versionadded:: 0.9.0b2
"""
- self._lockmode = mode
+ self._lockmode = LockmodeArgs(mode=mode, of=of)
@_generative()
def params(self, *args, **kwargs):
@@ -2683,13 +2692,6 @@ def update(self, values, synchronize_session='evaluate'):
update_op.exec_()
return update_op.rowcount
- _lockmode_lookup = {
- 'read': 'read',
- 'read_nowait': 'read_nowait',
- 'update': True,
- 'update_nowait': 'nowait',
- None: False
- }
def _compile_context(self, labels=True):
context = QueryContext(self)
@@ -2699,12 +2701,13 @@ def _compile_context(self, labels=True):
context.labels = labels
- if self._lockmode:
- try:
- context.for_update = self._lockmode_lookup[self._lockmode]
- except KeyError:
- raise sa_exc.ArgumentError(
- "Unknown lockmode %r" % self._lockmode)
+ if isinstance(self._lockmode, bool) and self._lockmode:
+ context.for_update = LockmodeArgs(mode='update')
+ elif isinstance(self._lockmode, LockmodeArgs):
+ if self._lockmode.mode not in LockmodeArgs.lockmodes:
+ raise sa_exc.ArgumentError('Unknown lockmode %r' % self._lockmode.mode)
+ context.for_update = self._lockmode
+
for entity in self._entities:
entity.setup_context(self, context)
@@ -3409,12 +3412,11 @@ def __str__(self):
return str(self.column)
-
class QueryContext(object):
multi_row_eager_loaders = False
adapter = None
froms = ()
- for_update = False
+ for_update = None
def __init__(self, query):
@@ -3489,3 +3491,62 @@ def process_query(self, query):
else:
alias = self.alias
query._from_obj_alias = sql_util.ColumnAdapter(alias)
+
+
+class LockmodeArgs(object):
+
+ lockmodes = [None,
+ 'read', 'read_nowait',
+ 'update', 'update_nowait'
+ ]
+
+ mode = None
+ of = None
+
+ def __init__(self, mode=None, of=None):
+ """ORM-level Lockmode
+
+ :class:`.LockmodeArgs` defines the locking strategy for the
+ dialects as given by ``FOR UPDATE [OF] [NOWAIT]``. The optional
+ OF component is translated by the dialects into the supported
+ tablename and columnname descriptors.
+
+ :param mode: Defines the lockmode to use.
+
+ ``None`` - translates to no lockmode
+
+ ``'update'`` - translates to ``FOR UPDATE``
+ (standard SQL, supported by most dialects)
+
+ ``'update_nowait'`` - translates to ``FOR UPDATE NOWAIT``
+ (supported by Oracle, PostgreSQL 8.1 upwards)
+
+ ``'read'`` - translates to ``LOCK IN SHARE MODE`` (for MySQL),
+ and ``FOR SHARE`` (for PostgreSQL)
+
+ ``'read_nowait'`` - translates to ``FOR SHARE NOWAIT``
+ (supported by PostgreSQL). ``FOR SHARE`` and
+ ``FOR SHARE NOWAIT`` (PostgreSQL).
+
+ :param of: either a column descriptor, or list of column
+ descriptors, representing the optional OF part of the
+ clause. This passes the descriptor to the
+ corresponding :meth:`~sqlalchemy.orm.query.LockmodeArgs` object,
+ and translates to ``FOR UPDATE OF table [NOWAIT]`` respectively
+ ``FOR UPDATE OF table, table [NOWAIT]`` (PostgreSQL), or
+ ``FOR UPDATE OF table.column [NOWAIT]`` respectively
+ ``FOR UPDATE OF table.column, table.column [NOWAIT]`` (Oracle).
+
+ .. versionadded:: 0.9.0b2
+ """
+
+ if isinstance(mode, bool) and mode:
+ mode = 'update'
+
+ self.mode = mode
+
+ # extract table names and column names
+ if isinstance(of, attributes.QueryableAttribute):
+ self.of = (of.expression.table.name, of.expression.name)
+ elif isinstance(of, (tuple, list)) and of != []:
+ self.of = [(o.expression.table.name, o.expression.name) for o in of]
View
7 lib/sqlalchemy/sql/compiler.py
@@ -1570,7 +1570,12 @@ def order_by_clause(self, select, **kw):
return ""
def for_update_clause(self, select):
- if select.for_update:
+ # backwards compatibility
+ if isinstance(select.for_update, bool):
+ return " FOR UPDATE" if select.for_update else ""
+ elif isinstance(select.for_update, str):
+ return " FOR UPDATE"
+ elif select.for_update.mode is not None:
return " FOR UPDATE"
else:
return ""
View
1  lib/sqlalchemy/sql/selectable.py
@@ -2785,4 +2785,3 @@ def __init__(self, element, values):
Annotated.__init__(self, element, values)
-
View
65 test/orm/test_lockmode.py
@@ -73,6 +73,23 @@ def test_postgres_update(self):
dialect=postgresql.dialect()
)
+ def test_postgres_update_of(self):
+ User = self.classes.User
+ sess = Session()
+ self.assert_compile(sess.query(User.id).with_lockmode('update', of=User.id),
+ "SELECT users.id AS users_id FROM users FOR UPDATE OF users",
+ dialect=postgresql.dialect()
+ )
+
+ def test_postgres_update_of_list(self):
+ User = self.classes.User
+ sess = Session()
+ self.assert_compile(sess.query(User.id).with_lockmode('update', of=[User.id, User.id, User.id]),
+ "SELECT users.id AS users_id FROM users FOR UPDATE OF users, users, users",
@zzzeek Owner
zzzeek added a note

do you actually want "users, users, users" here? I will likely add a set() so that tables are uniqued here.

Yes, it should be there multiple times. In a real scenario you would use different table.column entries, but for the test case it doesn't matter. (And both Ora and Psql accept multiple same entries anyway).

@zzzeek Owner
zzzeek added a note

Ok the idea here though is that if you have the same table multiple times, and you aren't displaying the column, just the table, the table should be de-duped on name. if i had with_lockmode(of=[User.id, User.name]), id want Oracle to say FOR UPDATE OF users.id, users.name but PG should say FOR UPDATE OF users.

in the most recent changesets you'll also see support for table aliases and select "adaptation" which is pretty important with the ORM.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
+ dialect=postgresql.dialect()
+ )
+
+
def test_postgres_update_nowait(self):
User = self.classes.User
sess = Session()
@@ -81,6 +98,22 @@ def test_postgres_update_nowait(self):
dialect=postgresql.dialect()
)
+ def test_postgres_update_nowait_of(self):
+ User = self.classes.User
+ sess = Session()
+ self.assert_compile(sess.query(User.id).with_lockmode('update_nowait', of=User.id),
+ "SELECT users.id AS users_id FROM users FOR UPDATE OF users NOWAIT",
+ dialect=postgresql.dialect()
+ )
+
+ def test_postgres_update_nowait_of_list(self):
+ User = self.classes.User
+ sess = Session()
+ self.assert_compile(sess.query(User.id).with_lockmode('update_nowait', of=[User.id, User.id, User.id]),
+ "SELECT users.id AS users_id FROM users FOR UPDATE OF users, users, users NOWAIT",
+ dialect=postgresql.dialect()
+ )
+
def test_oracle_update(self):
User = self.classes.User
sess = Session()
@@ -89,6 +122,22 @@ def test_oracle_update(self):
dialect=oracle.dialect()
)
+ def test_oracle_update_of(self):
+ User = self.classes.User
+ sess = Session()
+ self.assert_compile(sess.query(User.id).with_lockmode('update', of=User.id),
+ "SELECT users.id AS users_id FROM users FOR UPDATE OF users.id",
+ dialect=oracle.dialect()
+ )
+
+ def test_oracle_update_of_list(self):
+ User = self.classes.User
+ sess = Session()
+ self.assert_compile(sess.query(User.id).with_lockmode('update', of=[User.id, User.id, User.id]),
+ "SELECT users.id AS users_id FROM users FOR UPDATE OF users.id, users.id, users.id",
+ dialect=oracle.dialect()
+ )
+
def test_oracle_update_nowait(self):
User = self.classes.User
sess = Session()
@@ -97,6 +146,22 @@ def test_oracle_update_nowait(self):
dialect=oracle.dialect()
)
+ def test_oracle_update_nowait_of(self):
+ User = self.classes.User
+ sess = Session()
+ self.assert_compile(sess.query(User.id).with_lockmode('update_nowait', of=User.id),
+ "SELECT users.id AS users_id FROM users FOR UPDATE OF users.id NOWAIT",
+ dialect=oracle.dialect()
+ )
+
+ def test_oracle_update_nowait_of_list(self):
+ User = self.classes.User
+ sess = Session()
+ self.assert_compile(sess.query(User.id).with_lockmode('update_nowait', of=[User.id, User.id, User.id]),
+ "SELECT users.id AS users_id FROM users FOR UPDATE OF users.id, users.id, users.id NOWAIT",
+ dialect=oracle.dialect()
+ )
+
def test_mysql_read(self):
User = self.classes.User
sess = Session()
Something went wrong with that request. Please try again.