generated sequence value (i.e. MySQL, MS-SQL) now work correctly
when there is a composite primary key where the "autoincrement" column
is not the first primary key column in the table.
+
+ - the last_inserted_ids() method has been renamed to the descriptor
+ "inserted_primary_key".
- engines
- transaction isolation level may be specified with
* the ``inline=True`` flag is not set on the ``Insert()`` or ``Update()`` construct.
-For a statement execution which is not an executemany, the returned ``ResultProxy`` will contain a collection accessible via ``result.postfetch_cols()`` which contains a list of all ``Column`` objects which had an inline-executed default. Similarly, all parameters which were bound to the statement, including all Python and SQL expressions which were pre-executed, are present in the ``last_inserted_params()`` or ``last_updated_params()`` collections on ``ResultProxy``. The ``last_inserted_ids()`` collection contains a list of primary key values for the row inserted.
+For a statement execution which is not an executemany, the returned ``ResultProxy`` will contain a collection accessible via ``result.postfetch_cols()`` which contains a list of all ``Column`` objects which had an inline-executed default. Similarly, all parameters which were bound to the statement, including all Python and SQL expressions which were pre-executed, are present in the ``last_inserted_params()`` or ``last_updated_params()`` collections on ``ResultProxy``. The ``inserted_primary_key`` collection contains a list of primary key values for the row inserted.
DDL-Level Defaults
-------------------
.. sourcecode:: pycon+sql
- >>> result.last_inserted_ids()
+ >>> result.inserted_primary_key
[1]
-The value of ``1`` was automatically generated by SQLite, but only because we did not specify the ``id`` column in our ``Insert`` statement; otherwise, our explicit value would have been used. In either case, SQLAlchemy always knows how to get at a newly generated primary key value, even though the method of generating them is different across different databases; each databases' ``Dialect`` knows the specific steps needed to determine the correct value (or values; note that ``last_inserted_ids()`` returns a list so that it supports composite primary keys).
+The value of ``1`` was automatically generated by SQLite, but only because we did not specify the ``id`` column in our ``Insert`` statement; otherwise, our explicit value would have been used. In either case, SQLAlchemy always knows how to get at a newly generated primary key value, even though the method of generating them is different across different databases; each databases' ``Dialect`` knows the specific steps needed to determine the correct value (or values; note that ``inserted_primary_key`` returns a list so that it supports composite primary keys).
Executing Multiple Statements
==============================
# 4 - offset of the error into the SQL statement
# 5 - rowid after insert
def post_exec(self):
- if getattr(self.compiled, "isinsert", False) and self.last_inserted_ids() is None:
+ if getattr(self.compiled, "isinsert", False) and self.inserted_primary_key is None:
self._last_inserted_ids = [self.cursor.sqlerrd[1]]
elif hasattr( self.compiled , 'offset' ):
self.cursor.offset( self.compiled.offset )
(attype, name))
coltype = sqltypes.NULLTYPE
# adjust the default value
+ autoincrement = False
if default is not None:
match = re.search(r"""(nextval\(')([^']+)('.*$)""", default)
if match is not None:
+ autoincrement = True
# the default is related to a Sequence
sch = schema
if '.' not in match.group(2) and sch is not None:
default = match.group(1) + ('"%s"' % sch) + '.' + match.group(2) + match.group(3)
column_info = dict(name=name, type=coltype, nullable=nullable,
- default=default)
+ default=default, autoincrement=autoincrement)
columns.append(column_info)
return columns
implicit_returning
use RETURNING or equivalent during INSERT execution in order to load
newly generated primary keys and other column defaults in one execution,
- which are then available via last_inserted_ids().
+ which are then available via inserted_primary_key.
If an insert statement has returning() specified explicitly,
- the "implicit" functionality is not used and last_inserted_ids()
+ the "implicit" functionality is not used and inserted_primary_key
will not be available.
dbapi_type_map
raise NotImplementedError()
- def last_inserted_ids(self):
- """Return the list of the primary key values for the last insert statement executed.
-
- This does not apply to straight textual clauses; only to
- ``sql.Insert`` objects compiled against a ``schema.Table``
- object. The order of items in the list is the same as that of
- the Table's 'primary_key' attribute.
- """
-
- raise NotImplementedError()
-
def last_inserted_params(self):
"""Return a dictionary of the full parameter dictionary for the last compiled INSERT statement.
self._cursor_executemany(context.cursor, context.statement, context.parameters, context=context)
else:
self._cursor_execute(context.cursor, context.statement, context.parameters[0], context=context)
+
if context.compiled:
context.post_exec()
if context.isinsert and not context.executemany:
consistent across backends.
Usage of this method is normally unnecessary; the
- last_inserted_ids() method provides a
+ inserted_primary_key method provides a
tuple of primary key values for a newly inserted row,
regardless of database backend.
self.rowcount
self.close() # autoclose
elif self.context.isinsert and \
+ not self.context.executemany and \
not self.context._is_explicit_returning:
# an insert, no explicit returning(), may need
# to fetch rows which were created via implicit
# returning, then close
- self.context.last_inserted_ids(self)
+ self.context._fetch_implicit_returning(self)
self.close()
return self
raise StopIteration
else:
yield row
+
+ @util.memoized_property
+ def inserted_primary_key(self):
+ """Return the primary key for the row just inserted.
+
+ This only applies to single row insert() constructs which
+ did not explicitly specify returning().
- def last_inserted_ids(self):
- """Return ``last_inserted_ids()`` from the underlying ExecutionContext.
-
- See ExecutionContext for details.
"""
- return self.context.last_inserted_ids(self)
+ if not self.context.isinsert:
+ raise exc.InvalidRequestError("Statement is not an insert() expression construct.")
+ elif self.context._is_explicit_returning:
+ raise exc.InvalidRequestError("Can't call inserted_primary_key when returning() is used.")
+
+ return self.context._inserted_primary_key
+ @util.deprecated("Use inserted_primary_key")
+ def last_inserted_ids(self):
+ """deprecated. use inserted_primary_key."""
+
+ return self.inserted_primary_key
+
def last_updated_params(self):
"""Return ``last_updated_params()`` from the underlying ExecutionContext.
import re, random
from sqlalchemy.engine import base, reflection
from sqlalchemy.sql import compiler, expression
-from sqlalchemy import exc, types as sqltypes
+from sqlalchemy import exc, types as sqltypes, util
AUTOCOMMIT_REGEXP = re.compile(r'\s*(?:UPDATE|INSERT|CREATE|DELETE|DROP|ALTER)',
re.I | re.UNICODE)
self.isinsert = self.isupdate = self.isdelete = self.executemany = self.should_autocommit = False
self.cursor = self.create_cursor()
- @property
+ @util.memoized_property
def _is_explicit_returning(self):
return self.compiled and \
getattr(self.compiled.statement, '_returning', False)
def post_insert(self):
if self.dialect.postfetch_lastrowid and \
- (not len(self._last_inserted_ids) or \
- None in self._last_inserted_ids):
+ (not len(self._inserted_primary_key) or \
+ None in self._inserted_primary_key):
table = self.compiled.statement.table
lastrowid = self.get_lastrowid()
- self._last_inserted_ids = [c is table._autoincrement_column and lastrowid or v
- for c, v in zip(table.primary_key, self._last_inserted_ids)
+ self._inserted_primary_key = [c is table._autoincrement_column and lastrowid or v
+ for c, v in zip(table.primary_key, self._inserted_primary_key)
]
- def last_inserted_ids(self, resultproxy):
- if not self.isinsert:
- raise exc.InvalidRequestError("Statement is not an insert() expression construct.")
+ def _fetch_implicit_returning(self, resultproxy):
if self.dialect.implicit_returning and \
not self.compiled.statement._returning and \
table = self.compiled.statement.table
row = resultproxy.first()
- self._last_inserted_ids = [v is not None and v or row[c]
- for c, v in zip(table.primary_key, self._last_inserted_ids)
+ self._inserted_primary_key = [v is not None and v or row[c]
+ for c, v in zip(table.primary_key, self._inserted_primary_key)
]
- return self._last_inserted_ids
-
- else:
- return self._last_inserted_ids
def last_inserted_params(self):
return self._last_inserted_params
def __process_defaults(self):
"""Generate default values for compiled insert/update statements,
- and generate last_inserted_ids() collection.
+ and generate inserted_primary_key collection.
"""
if self.executemany:
compiled_parameters[c.key] = val
if self.isinsert:
- self._last_inserted_ids = [compiled_parameters.get(c.key, None)
+ self._inserted_primary_key = [compiled_parameters.get(c.key, None)
for c in self.compiled.statement.table.primary_key]
self._last_inserted_params = compiled_parameters
else:
statement = table.insert()
for state, params, mapper, connection, value_params in insert:
c = connection.execute(statement.values(value_params), params)
- primary_key = c.last_inserted_ids()
+ primary_key = c.inserted_primary_key
if primary_key is not None:
# set primary key attributes
# create a list of column assignment clauses as tuples
values = []
+ #need_pks = self.isinsert and not self.inline
+
+ #implicit_returning = need_pks and \
+ # self.dialect.implicit_returning and \
+ # stmt.table.implicit_returning and \
+ # not self.statement._returning and \
+
implicit_returning = self.dialect.implicit_returning and \
stmt.table.implicit_returning
) and \
not self.inline and \
not self.statement._returning:
-
+
if implicit_returning:
if isinstance(c.default, schema.Sequence):
proc = self.process(c.default)
Opens=datetime.time(8, 15, 59),
LastEscape=datetime.datetime(2004, 7, 29, 5, 6, 7),
Admission=4.95,
- ).last_inserted_ids()[0]
+ ).inserted_primary_key[0]
sdz = Zoo.insert().execute(Name =u'San Diego Zoo',
Founded = datetime.date(1935, 9, 13),
Opens = datetime.time(9, 0, 0),
Admission = 0,
- ).last_inserted_ids()[0]
+ ).inserted_primary_key[0]
Zoo.insert().execute(
Name = u'Montr\xe9al Biod\xf4me',
)
seaworld = Zoo.insert().execute(
- Name =u'Sea_World', Admission = 60).last_inserted_ids()[0]
+ Name =u'Sea_World', Admission = 60).inserted_primary_key[0]
# Let's add a crazy futuristic Zoo to test large date values.
lp = Zoo.insert().execute(Name =u'Luna Park',
Founded = datetime.date(2072, 7, 17),
Opens = datetime.time(0, 0, 0),
Admission = 134.95,
- ).last_inserted_ids()[0]
+ ).inserted_primary_key[0]
# Animals
leopardid = Animal.insert().execute(Species=u'Leopard', Lifespan=73.5,
- ).last_inserted_ids()[0]
+ ).inserted_primary_key[0]
Animal.update(Animal.c.ID==leopardid).execute(ZooID=wap,
LastEscape=datetime.datetime(2004, 12, 21, 8, 15, 0, 999907))
- lion = Animal.insert().execute(Species=u'Lion', ZooID=wap).last_inserted_ids()[0]
+ lion = Animal.insert().execute(Species=u'Lion', ZooID=wap).inserted_primary_key[0]
Animal.insert().execute(Species=u'Slug', Legs=1, Lifespan=.75)
tiger = Animal.insert().execute(Species=u'Tiger', ZooID=sdz
- ).last_inserted_ids()[0]
+ ).inserted_primary_key[0]
# Override Legs.default with itself just to make sure it works.
Animal.insert().execute(Species=u'Bear', Legs=4)
Animal.insert().execute(Species=u'Centipede', Legs=100)
emp = Animal.insert().execute(Species=u'Emperor Penguin', Legs=2,
- ZooID=seaworld).last_inserted_ids()[0]
+ ZooID=seaworld).inserted_primary_key[0]
adelie = Animal.insert().execute(Species=u'Adelie Penguin', Legs=2,
- ZooID=seaworld).last_inserted_ids()[0]
+ ZooID=seaworld).inserted_primary_key[0]
Animal.insert().execute(Species=u'Millipede', Legs=1000000, ZooID=sdz)
# Add a mother and child to test relationships
bai_yun = Animal.insert().execute(Species=u'Ape', Name=u'Bai Yun',
- Legs=2).last_inserted_ids()[0]
+ Legs=2).inserted_primary_key[0]
Animal.insert().execute(Species=u'Ape', Name=u'Hua Mei', Legs=2,
MotherID=bai_yun)
eq_([(9, 'Python')], list(cats))
result = cattable.insert().values(description='PHP').execute()
- eq_([10], result.last_inserted_ids())
+ eq_([10], result.inserted_primary_key)
lastcat = cattable.select().order_by(desc(cattable.c.id)).execute()
eq_((10, 'PHP'), lastcat.first())
try:
tr = con.begin()
r = con.execute(t2.insert(), descr='hello')
- self.assert_(r.last_inserted_ids() == [200])
+ self.assert_(r.inserted_primary_key == [200])
r = con.execute(t1.insert(), descr='hello')
- self.assert_(r.last_inserted_ids() == [100])
+ self.assert_(r.inserted_primary_key == [100])
finally:
tr.commit()
def go():
# execute with explicit id
r = table.insert().execute({'id':30, 'data':'d1'})
- assert r.last_inserted_ids() == [30]
+ assert r.inserted_primary_key == [30]
# execute with prefetch id
r = table.insert().execute({'data':'d2'})
- assert r.last_inserted_ids() == [1]
+ assert r.inserted_primary_key == [1]
# executemany with explicit ids
table.insert().execute({'id':31, 'data':'d3'}, {'id':32, 'data':'d4'})
def go():
table.insert().execute({'id':30, 'data':'d1'})
r = table.insert().execute({'data':'d2'})
- assert r.last_inserted_ids() == [5]
+ assert r.inserted_primary_key == [5]
table.insert().execute({'id':31, 'data':'d3'}, {'id':32, 'data':'d4'})
table.insert().execute({'data':'d5'}, {'data':'d6'})
table.insert(inline=True).execute({'id':33, 'data':'d7'})
def go():
# execute with explicit id
r = table.insert().execute({'id':30, 'data':'d1'})
- assert r.last_inserted_ids() == [30]
+ assert r.inserted_primary_key == [30]
# execute with prefetch id
r = table.insert().execute({'data':'d2'})
- assert r.last_inserted_ids() == [1]
+ assert r.inserted_primary_key == [1]
# executemany with explicit ids
table.insert().execute({'id':31, 'data':'d3'}, {'id':32, 'data':'d4'})
def go():
table.insert().execute({'id':30, 'data':'d1'})
r = table.insert().execute({'data':'d2'})
- assert r.last_inserted_ids() == [5]
+ assert r.inserted_primary_key == [5]
table.insert().execute({'id':31, 'data':'d3'}, {'id':32, 'data':'d4'})
table.insert().execute({'data':'d5'}, {'data':'d6'})
table.insert(inline=True).execute({'id':33, 'data':'d7'})
t = Table("speedy_users", meta, autoload=True)
r = t.insert().execute(user_name='user', user_password='lala')
- assert r.last_inserted_ids() == [1]
+ assert r.inserted_primary_key == [1]
l = t.select().execute().fetchall()
assert l == [(1, 'user', 'lala')]
finally:
("DROP TABLE t1", {}, None)
]
- if engine.dialect.preexecute_pk_sequences:
+ if True: # or engine.dialect.preexecute_pk_sequences:
cursor = [
("CREATE TABLE t1", {}, ()),
("INSERT INTO t1 (c1, c2)", {'c2': 'some data', 'c1': 5}, [5, 'some data']),
@testing.fails_on('firebird', 'Data type unknown')
def test_update(self):
r = t.insert().execute()
- pk = r.last_inserted_ids()[0]
+ pk = r.inserted_primary_key[0]
t.update(t.c.col1==pk).execute(col4=None, col5=None)
ctexec = currenttime.scalar()
l = t.select(t.c.col1==pk).execute()
@testing.fails_on('firebird', 'Data type unknown')
def test_update_values(self):
r = t.insert().execute()
- pk = r.last_inserted_ids()[0]
+ pk = r.inserted_primary_key[0]
t.update(t.c.col1==pk, values={'col3': 55}).execute()
l = t.select(t.c.col1==pk).execute()
l = l.first()
engine = engines.testing_engine(options={'implicit_returning':returning})
engine.execute(t2.insert(), nextid=1)
r = engine.execute(t1.insert(), data='hi')
- eq_([1], r.last_inserted_ids())
+ eq_([1], r.inserted_primary_key)
engine.execute(t2.insert(), nextid=2)
r = engine.execute(t1.insert(), data='there')
- eq_([2], r.last_inserted_ids())
+ eq_([2], r.inserted_primary_key)
class PKIncrementTest(_base.TablesTest):
run_define_tables = 'each'
def _test_autoincrement(self, bind):
ids = set()
rs = bind.execute(aitable.insert(), int1=1)
- last = rs.last_inserted_ids()[0]
+ last = rs.inserted_primary_key[0]
self.assert_(last)
self.assert_(last not in ids)
ids.add(last)
rs = bind.execute(aitable.insert(), str1='row 2')
- last = rs.last_inserted_ids()[0]
+ last = rs.inserted_primary_key[0]
self.assert_(last)
self.assert_(last not in ids)
ids.add(last)
rs = bind.execute(aitable.insert(), int1=3, str1='row 3')
- last = rs.last_inserted_ids()[0]
+ last = rs.inserted_primary_key[0]
self.assert_(last)
self.assert_(last not in ids)
ids.add(last)
rs = bind.execute(aitable.insert(values={'int1':func.length('four')}))
- last = rs.last_inserted_ids()[0]
+ last = rs.inserted_primary_key[0]
self.assert_(last)
self.assert_(last not in ids)
ids.add(last)
single.create()
r = single.insert().execute()
- id_ = r.last_inserted_ids()[0]
+ id_ = r.inserted_primary_key[0]
eq_(id_, 1)
eq_(1, sa.select([func.count(sa.text('*'))], from_obj=single).scalar())
nodes.create()
r = nodes.insert().execute(data='foo')
- id_ = r.last_inserted_ids()[0]
+ id_ = r.inserted_primary_key[0]
nodes.insert().execute(data='bar', parent_id=id_)
@testing.fails_on('sqlite', 'FIXME: unknown')
cartitems.insert().execute(description='there')
r = cartitems.insert().execute(description='lala')
- assert r.last_inserted_ids() and r.last_inserted_ids()[0] is not None
- id_ = r.last_inserted_ids()[0]
+ assert r.inserted_primary_key and r.inserted_primary_key[0] is not None
+ id_ = r.inserted_primary_key[0]
eq_(1,
sa.select([func.count(cartitems.c.cart_id)],
assert t.select().execute().first()['value'] == 5
r = t.insert(values=dict(value=func.length("sfsaafsda"))).execute()
- id = r.last_inserted_ids()[0]
+ id = r.inserted_primary_key[0]
assert t.select(t.c.id==id).execute().first()['value'] == 9
t.update(values={t.c.value:func.length("asdf")}).execute()
assert t.select().execute().first()['value'] == 4
assert users.select(users.c.user_id==7).execute().first()['user_name'] == 'fred'
def test_lastrow_accessor(self):
- """Tests the last_inserted_ids() and lastrow_has_id() functions."""
+ """Tests the inserted_primary_key and lastrow_has_id() functions."""
def insert_values(engine, table, values):
"""
result = engine.execute(table.insert(), **values)
ret = values.copy()
- for col, id in zip(table.primary_key, result.last_inserted_ids()):
+ for col, id in zip(table.primary_key, result.inserted_primary_key):
ret[col.key] = id
if result.lastrow_has_defaults():
- criterion = and_(*[col==id for col, id in zip(table.primary_key, result.last_inserted_ids())])
+ criterion = and_(*[col==id for col, id in zip(table.primary_key, result.inserted_primary_key)])
row = engine.execute(table.select(criterion)).first()
for c in table.c:
ret[c.key] = row[c]
metadata.create_all()
r = related.insert().values(id=12).execute()
- id = r.last_inserted_ids()[0]
+ id = r.inserted_primary_key[0]
assert id==12
r = t6.insert().values(manual_id=id).execute()
- eq_(r.last_inserted_ids(), [12, 1])
+ eq_(r.inserted_primary_key, [12, 1])
def test_autoclose_on_insert(self):
if testing.against('firebird', 'postgresql', 'oracle', 'mssql'):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
