--- /dev/null
+.. change::
+ :tags: usecase, sql
+ :tickets: 7998
+
+ Altered the compilation mechanics of the :class:`.Insert` construct such
+ that the "autoincrement primary key" column value will be fetched via
+ ``cursor.lastrowid`` or RETURNING even if present in the parameter set or
+ within the :meth:`.Insert.values` method as a plain bound value, for
+ single-row INSERT statements on specific backends that are known to
+ generate autoincrementing values even when explicit NULL is passed. This
+ restores a behavior that was in the 1.3 series for both the use case of
+ separate parameter set as well as :meth:`.Insert.values`. In 1.4, the
+ parameter set behavior unintentionally changed to no longer do this, but
+ the :meth:`.Insert.values` method would still fetch autoincrement values up
+ until 1.4.21 where :ticket:`6770` changed the behavior yet again again
+ unintentionally as this use case was never covered.
+
+ The behavior is now defined as "working" to suit the case where databases
+ such as SQLite, MySQL and MariaDB will ignore an explicit NULL primary key
+ value and nonetheless invoke an autoincrement generator.
supports_sane_rowcount = True
supports_sane_multi_rowcount = False
supports_multivalues_insert = True
+ insert_null_pk_still_autoincrements = True
supports_comments = True
inline_comments = True
supports_multivalues_insert = True
tuple_in_values = True
supports_statement_cache = True
+ insert_null_pk_still_autoincrements = True
default_paramstyle = "qmark"
execution_ctx_cls = SQLiteExecutionContext
preexecute_autoincrement_sequences = False
supports_identity_columns = False
postfetch_lastrowid = True
+ insert_null_pk_still_autoincrements = False
implicit_returning = False
full_returning = False
insert_executemany_returning = False
for col in table.primary_key
]
+ autoinc_getter = None
autoinc_col = table._autoincrement_column
if autoinc_col is not None:
# apply type post processors to the lastrowid
- proc = autoinc_col.type._cached_result_processor(
+ lastrowid_processor = autoinc_col.type._cached_result_processor(
self.dialect, None
)
+ autoinc_key = param_key_getter(autoinc_col)
+
+ # if a bind value is present for the autoincrement column
+ # in the parameters, we need to do the logic dictated by
+ # #7998; honor a non-None user-passed parameter over lastrowid.
+ # previously in the 1.4 series we weren't fetching lastrowid
+ # at all if the key were present in the parameters
+ if autoinc_key in self.binds:
+
+ def autoinc_getter(lastrowid, parameters):
+ param_value = parameters.get(autoinc_key, lastrowid)
+ if param_value is not None:
+ # they supplied non-None parameter, use that.
+ # SQLite at least is observed to return the wrong
+ # cursor.lastrowid for INSERT..ON CONFLICT so it
+ # can't be used in all cases
+ return param_value
+ else:
+ # use lastrowid
+ return lastrowid
+
else:
- proc = None
+ lastrowid_processor = None
row_fn = result.result_tuple([col.key for col in table.primary_key])
that were sent along with the INSERT.
"""
- if proc is not None:
- lastrowid = proc(lastrowid)
+ if lastrowid_processor is not None:
+ lastrowid = lastrowid_processor(lastrowid)
if lastrowid is None:
return row_fn(getter(parameters) for getter, col in getters)
else:
return row_fn(
- lastrowid if col is autoinc_col else getter(parameters)
+ (
+ autoinc_getter(lastrowid, parameters)
+ if autoinc_getter
+ else lastrowid
+ )
+ if col is autoinc_col
+ else getter(parameters)
for getter, col in getters
)
else:
cols = stmt.table.columns
+ if compile_state.isinsert and not compile_state._has_multi_parameters:
+ # new rules for #7998. fetch lastrowid or implicit returning
+ # for autoincrement column even if parameter is NULL, for DBs that
+ # override NULL param for primary key (sqlite, mysql/mariadb)
+ autoincrement_col = stmt.table._autoincrement_column
+ insert_null_pk_still_autoincrements = (
+ compiler.dialect.insert_null_pk_still_autoincrements
+ )
+ else:
+ autoincrement_col = insert_null_pk_still_autoincrements = None
+
for c in cols:
# scan through every column in the target table
implicit_returning,
implicit_return_defaults,
values,
+ autoincrement_col,
+ insert_null_pk_still_autoincrements,
kw,
)
implicit_returning,
implicit_return_defaults,
values,
+ autoincrement_col,
+ insert_null_pk_still_autoincrements,
kw,
):
value = parameters.pop(col_key)
)
if coercions._is_literal(value):
+
+ if (
+ insert_null_pk_still_autoincrements
+ and c.primary_key
+ and c is autoincrement_col
+ ):
+ # support use case for #7998, fetch autoincrement cols
+ # even if value was given
+ if implicit_returning:
+ compiler.implicit_returning.append(c)
+ elif compiler.dialect.postfetch_lastrowid:
+ compiler.postfetch_lastrowid = True
+
value = _create_bind_param(
compiler,
c,
**kw,
)
elif value._is_bind_parameter:
+ if (
+ insert_null_pk_still_autoincrements
+ and value.value is None
+ and c.primary_key
+ and c is autoincrement_col
+ ):
+ # support use case for #7998, fetch autoincrement cols
+ # even if value was given
+ if implicit_returning:
+ compiler.implicit_returning.append(c)
+ elif compiler.dialect.postfetch_lastrowid:
+ compiler.postfetch_lastrowid = True
+
value = _handle_values_anonymous_param(
compiler,
c,
and not stmt._returning
and not compile_state._has_multi_parameters
)
-
implicit_returning = (
need_pks
and compiler.dialect.implicit_returning
implicit_return_defaults = set(stmt._return_defaults_columns)
postfetch_lastrowid = need_pks and compiler.dialect.postfetch_lastrowid
-
return (
need_pks,
implicit_returning,
"mssql",
)
+ @property
+ def database_discards_null_for_autoincrement(self):
+ """target database autoincrements a primary key and populates
+ .lastrowid even if NULL is explicitly passed for the column.
+
+ """
+ return succeeds_if(
+ lambda config: (
+ config.db.dialect.insert_null_pk_still_autoincrements
+ )
+ )
+
@property
def emulated_lastrowid_even_with_sequences(self):
""" "target dialect retrieves cursor.lastrowid or an equivalent
checkparams={"name_1": "foo"},
)
+ @testing.combinations(
+ True, False, argnames="insert_null_still_autoincrements"
+ )
+ @testing.combinations("values", "params", "nothing", argnames="paramtype")
+ def test_explicit_null_implicit_returning_still_renders(
+ self, paramtype, insert_null_still_autoincrements
+ ):
+ """test for future support of #7998 with RETURNING"""
+ t = Table(
+ "t",
+ MetaData(),
+ Column("x", Integer, primary_key=True),
+ Column("q", Integer),
+ )
+
+ dialect = postgresql.dialect(implicit_returning=True)
+ dialect.insert_null_pk_still_autoincrements = (
+ insert_null_still_autoincrements
+ )
+
+ if paramtype == "values":
+ # for values present, we now have an extra check for this
+ stmt = t.insert().values(x=None, q=5)
+ if insert_null_still_autoincrements:
+ expected = (
+ "INSERT INTO t (x, q) VALUES (%(x)s, %(q)s) RETURNING t.x"
+ )
+ else:
+ expected = "INSERT INTO t (x, q) VALUES (%(x)s, %(q)s)"
+ params = None
+ elif paramtype == "params":
+ # for params, compiler doesnt have the value available to look
+ # at. we assume non-NULL
+ stmt = t.insert()
+ if insert_null_still_autoincrements:
+ expected = (
+ "INSERT INTO t (x, q) VALUES (%(x)s, %(q)s) RETURNING t.x"
+ )
+ else:
+ expected = "INSERT INTO t (x, q) VALUES (%(x)s, %(q)s)"
+ params = {"x": None, "q": 5}
+ elif paramtype == "nothing":
+ # no params, we assume full INSERT. this kind of compilation
+ # doesn't actually happen during execution since there are always
+ # parameters or values
+ stmt = t.insert()
+ expected = "INSERT INTO t (x, q) VALUES (%(x)s, %(q)s)"
+ params = None
+
+ self.assert_compile(stmt, expected, params=params, dialect=dialect)
+
def test_insert_multiple_values(self):
ins = self.tables.myothertable.insert().values(
[{"othername": "foo"}, {"othername": "bar"}]
class TableInsertTest(fixtures.TablesTest):
"""test for consistent insert behavior across dialects
- regarding the inline() method, lower-case 't' tables.
+ regarding the inline() method, values() method, lower-case 't' tables.
"""
returning=None,
inserted_primary_key=False,
table=None,
+ parameters=None,
):
- r = connection.execute(stmt)
+ if parameters is not None:
+ r = connection.execute(stmt, parameters)
+ else:
+ r = connection.execute(stmt)
if returning:
returned = r.first()
(testing.db.dialect.default_sequence_base, "data", 5),
inserted_primary_key=(),
)
+
+ @testing.requires.database_discards_null_for_autoincrement
+ def test_explicit_null_pk_values_db_ignores_it(self, connection):
+ """test new use case in #7998"""
+
+ # NOTE: this use case uses cursor.lastrowid on SQLite, MySQL, MariaDB,
+ # however when SQLAlchemy 2.0 adds support for RETURNING to SQLite
+ # and MariaDB, it should work there as well.
+
+ t = self.tables.foo_no_seq
+ self._test(
+ connection,
+ t.insert().values(id=None, data="data", x=5),
+ (testing.db.dialect.default_sequence_base, "data", 5),
+ inserted_primary_key=(testing.db.dialect.default_sequence_base,),
+ table=t,
+ )
+
+ @testing.requires.database_discards_null_for_autoincrement
+ def test_explicit_null_pk_params_db_ignores_it(self, connection):
+ """test new use case in #7998"""
+
+ # NOTE: this use case uses cursor.lastrowid on SQLite, MySQL, MariaDB,
+ # however when SQLAlchemy 2.0 adds support for RETURNING to SQLite
+ # and MariaDB, it should work there as well.
+
+ t = self.tables.foo_no_seq
+ self._test(
+ connection,
+ t.insert(),
+ (testing.db.dialect.default_sequence_base, "data", 5),
+ inserted_primary_key=(testing.db.dialect.default_sequence_base,),
+ table=t,
+ parameters=dict(id=None, data="data", x=5),
+ )
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
