def _create_sequence(cr, seq_name, number_increment, number_next):
""" Create a PostreSQL sequence. """
if number_increment == 0:
raise UserError(_('Step must not be zero.'))
sql = "CREATE SEQUENCE %s INCREMENT BY %%s START WITH %%s" % seq_name
cr.execute(sql, (number_increment, number_next))
def _drop_sequences(cr, seq_names):
""" Drop the PostreSQL sequences if they exist. """
names = sql.SQL(',').join(map(sql.Identifier, seq_names))
# RESTRICT is the default; it prevents dropping the sequence if an
# object depends on it.
cr.execute(sql.SQL("DROP SEQUENCE IF EXISTS {} RESTRICT").format(names))
def _alter_sequence(cr, seq_name, number_increment=None, number_next=None):
""" Alter a PostreSQL sequence. """
if number_increment == 0:
raise UserError(_("Step must not be zero."))
cr.execute("SELECT relname FROM pg_class WHERE relkind=%s AND relname=%s", ('S', seq_name))
if not cr.fetchone():
# sequence is not created yet, we're inside create() so ignore it, will be set later
return
statement = sql.SQL("ALTER SEQUENCE") + sql.Identifier(seq_name)
params = []
if number_increment is not None:
statement += sql.SQL("INCREMENT BY") + sql.Placeholder()
params.append(number_increment)
if number_next is not None:
statement += sql.SQL("RESTART WITH") + sql.Placeholder()
params.append(number_next)
cr.execute(statement.join(' '), params)
def _select_nextval(cr, seq_name):
cr.execute("SELECT nextval(%s)", [seq_name])
return cr.fetchone()
def _predict_nextval(self, seq_id):
"""Predict next value for PostgreSQL sequence without consuming it"""
# Cannot use currval() as it requires prior call to nextval()
seqname = 'ir_sequence_%s' % seq_id
seqtable = sql.Identifier(seqname)
query = sql.SQL("""SELECT last_value,
(SELECT increment_by
FROM pg_sequences
WHERE sequencename = %s),
is_called
FROM {}""")
params = [seqname]
if self.env.cr._cnx.server_version < 100000:
query = sql.SQL("SELECT last_value, increment_by, is_called FROM {}")
params = []
self.env.cr.execute(query.format(seqtable), params)
(last_value, increment_by, is_called) = self.env.cr.fetchone()
if is_called:
return last_value + increment_by
# sequence has just been RESTARTed to return last_value next time
return last_value