--- /dev/null
+.. change::
+ :tags: deprecations, engine
+ :tickets: 6962
+
+ The :paramref:`_sa.create_engine.implicit_returning` parameter is
+ deprecated on the :func:`_sa.create_engine` function only; the parameter
+ remains available on the :class:`_schema.Table` object. This parameter was
+ originally intended to enable the "implicit returning" feature of
+ SQLAlchemy when it was first developed and was not enabled by default.
+ Under modern use, there's no reason this parameter should be disabled, and
+ it has been observed to cause confusion as it degrades performance and
+ makes it more difficult for the ORM to retrieve recently inserted server
+ defaults. The parameter remains available on :class:`_schema.Table` to
+ specifically suit database-level edge cases which make RETURNING
+ infeasible, the sole example currently being SQL Server's limitation that
+ INSERT RETURNING may not be used on a table that has INSERT triggers on it.
+
+
+.. change::
+ :tags: bug, oracle
+ :tickets: 6962
+
+ Related to the deprecation for
+ :paramref:`_sa.create_engine.implicit_returning`, the "implicit_returning"
+ feature is now enabled for the Oracle dialect in all cases; previously, the
+ feature would be turned off when an Oracle 8/8i version were detected,
+ however online documentation indicates both versions support the same
+ RETURNING syntax as modern versions.
supports_unicode_binds = False
max_identifier_length = 128
+ implicit_returning = True
+
supports_simple_order_by_label = False
cte_follows_insert = True
def initialize(self, connection):
super(OracleDialect, self).initialize(connection)
- self.implicit_returning = self.__dict__.get(
- "implicit_returning", self.server_version_info > (10,)
- )
+ # Oracle 8i has RETURNING:
+ # https://docs.oracle.com/cd/A87860_01/doc/index.htm
+
+ # so does Oracle8:
+ # https://docs.oracle.com/cd/A64702_01/doc/index.htm
if self._is_oracle_8:
self.colspecs = self.colspecs.copy()
'expressions, or an "empty set" SELECT, at statement execution'
"time.",
),
+ implicit_returning=(
+ "2.0",
+ "The :paramref:`_sa.create_engine.implicit_returning` parameter "
+ "is deprecated and will be removed in a future release. ",
+ ),
)
def create_engine(url, **kwargs):
"""Create a new :class:`_engine.Engine` instance.
:paramref:`_engine.Connection.execution_options.logging_token`
-
-
:param max_identifier_length: integer; override the max_identifier_length
determined by the dialect. if ``None`` or zero, has no effect. This
is the database's configured maximum number of characters that may be
"sqlalchemy.pool" logger. Defaults to a hexstring of the object's
id.
-
.. seealso::
:ref:`dbengine_logging` - further detail on how to configure
logging.
-
:param pool_pre_ping: boolean, if True will enable the connection pool
"pre-ping" feature that tests connections for liveness upon
each checkout.
in order to get their value. This is currently oriented towards
PostgreSQL.
- ``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 inserted_primary_key.
- If an insert statement has returning() specified explicitly,
- the "implicit" functionality is not used and inserted_primary_key
- will not be available.
-
``colspecs``
A dictionary of TypeEngine classes from sqlalchemy.types mapped
to subclasses that are specific to the dialect class. This
self.is_consumed = True
self.errormessage = None
else:
- self.errormessage = self._failure_message(params) % {
+ self.errormessage = self._failure_message(
+ execute_observed, params
+ ) % {
"received_statement": _received_statement,
"received_parameters": _received_parameters,
}
else:
return None
- def _failure_message(self, expected_params):
+ def _failure_message(self, execute_observed, expected_params):
return (
"Testing for compiled statement\n%r partial params %s, "
"received\n%%(received_statement)r with params "
self.params = params
self.dialect = dialect
- def _failure_message(self, expected_params):
+ def _failure_message(self, execute_observed, expected_params):
return (
"Testing for compiled statement ~%r partial params %s, "
"received %%(received_statement)r with params "
return received_stmt, execute_observed.context.compiled_parameters
- def _compare_sql(self, execute_observed, received_statement):
+ def _dialect_adjusted_statement(self, paramstyle):
stmt = re.sub(r"[\n\t]", "", self.statement)
- # convert our comparison statement to have the
- # paramstyle of the received
- paramstyle = execute_observed.context.dialect.paramstyle
if paramstyle == "pyformat":
stmt = re.sub(r":([\w_]+)", r"%(\1)s", stmt)
else:
elif paramstyle == "numeric":
repl = None
stmt = re.sub(r":([\w_]+)", repl, stmt)
+ return stmt
+ def _compare_sql(self, execute_observed, received_statement):
+ paramstyle = execute_observed.context.dialect.paramstyle
+ stmt = self._dialect_adjusted_statement(paramstyle)
return received_statement == stmt
+ def _failure_message(self, execute_observed, expected_params):
+ paramstyle = execute_observed.context.dialect.paramstyle
+ return (
+ "Testing for compiled statement\n%r partial params %s, "
+ "received\n%%(received_statement)r with params "
+ "%%(received_parameters)r"
+ % (
+ self._dialect_adjusted_statement(paramstyle).replace(
+ "%", "%%"
+ ),
+ repr(expected_params).replace("%", "%%"),
+ )
+ )
+
class CountStatements(AssertRule):
def __init__(self, count):
-from .. import config
-from .. import engines
from .. import fixtures
from ..assertions import eq_
from ..config import requirements
__requires__ = "implements_get_lastrowid", "autoincrement_insert"
- __engine_options__ = {"implicit_returning": False}
-
@classmethod
def define_tables(cls, metadata):
Table(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
Column("data", String(50)),
+ implicit_returning=False,
)
Table(
metadata,
Column("id", Integer, primary_key=True, autoincrement=False),
Column("data", String(50)),
+ implicit_returning=False,
)
def _assert_round_trip(self, table, conn):
)
@requirements.autoincrement_insert
- def test_autoclose_on_insert(self):
- if requirements.returning.enabled:
- engine = engines.testing_engine(
- options={"implicit_returning": False}
- )
- else:
- engine = config.db
-
- with engine.begin() as conn:
- r = conn.execute(
- self.tables.autoinc_pk.insert(), dict(data="some data")
- )
+ def test_autoclose_on_insert(self, connection):
+ r = connection.execute(
+ self.tables.autoinc_pk.insert(), dict(data="some data")
+ )
assert r._soft_closed
assert not r.closed
assert r.is_insert
__requires__ = "returning", "autoincrement_insert"
__backend__ = True
- __engine_options__ = {"implicit_returning": True}
-
def _assert_round_trip(self, table, conn):
row = conn.execute(table.select()).first()
eq_(
col.type.length, binary_table.c[col.name].type.length
)
- def test_autoincrement(self, metadata, connection):
+ @testing.combinations(True, False, argnames="implicit_returning")
+ def test_autoincrement(self, metadata, connection, implicit_returning):
Table(
"ai_1",
metadata,
Column("int_y", Integer, primary_key=True, autoincrement=True),
Column("int_n", Integer, DefaultClause("0"), primary_key=True),
+ implicit_returning=implicit_returning,
)
Table(
"ai_2",
metadata,
Column("int_y", Integer, primary_key=True, autoincrement=True),
Column("int_n", Integer, DefaultClause("0"), primary_key=True),
+ implicit_returning=implicit_returning,
)
Table(
"ai_3",
metadata,
Column("int_n", Integer, DefaultClause("0"), primary_key=True),
Column("int_y", Integer, primary_key=True, autoincrement=True),
+ implicit_returning=implicit_returning,
)
-
Table(
"ai_4",
metadata,
Column("int_n", Integer, DefaultClause("0"), primary_key=True),
Column("int_n2", Integer, DefaultClause("0"), primary_key=True),
+ implicit_returning=implicit_returning,
)
Table(
"ai_5",
metadata,
Column("int_y", Integer, primary_key=True, autoincrement=True),
Column("int_n", Integer, DefaultClause("0"), primary_key=True),
+ implicit_returning=implicit_returning,
)
Table(
"ai_6",
metadata,
Column("o1", String(1), DefaultClause("x"), primary_key=True),
Column("int_y", Integer, primary_key=True, autoincrement=True),
+ implicit_returning=implicit_returning,
)
Table(
"ai_7",
Column("o1", String(1), DefaultClause("x"), primary_key=True),
Column("o2", String(1), DefaultClause("x"), primary_key=True),
Column("int_y", Integer, autoincrement=True, primary_key=True),
+ implicit_returning=implicit_returning,
)
Table(
"ai_8",
metadata,
Column("o1", String(1), DefaultClause("x"), primary_key=True),
Column("o2", String(1), DefaultClause("x"), primary_key=True),
+ implicit_returning=implicit_returning,
)
metadata.create_all(connection)
eq_(col.autoincrement, "auto")
is_not(tbl._autoincrement_column, col)
- eng = [
- engines.testing_engine(options={"implicit_returning": False}),
- engines.testing_engine(options={"implicit_returning": True}),
- ]
-
- for counter, engine in enumerate(eng):
- connection.execute(tbl.insert())
- if "int_y" in tbl.c:
- eq_(
- connection.execute(select(tbl.c.int_y)).scalar(),
- counter + 1,
- )
- assert (
- list(connection.execute(tbl.select()).first()).count(
- counter + 1
- )
- == 1
- )
- else:
- assert 1 not in list(
- connection.execute(tbl.select()).first()
- )
- connection.execute(tbl.delete())
+ connection.execute(tbl.insert())
+ if "int_y" in tbl.c:
+ eq_(
+ connection.execute(select(tbl.c.int_y)).scalar(),
+ 1,
+ )
+ assert (
+ list(connection.execute(tbl.select()).first()).count(1)
+ == 1
+ )
+ else:
+ assert 1 not in list(connection.execute(tbl.select()).first())
class StringTest(fixtures.TestBase, AssertsCompiledSQL):
assert not dialect._use_nchar_for_unicode
dialect.initialize(Mock())
- assert not dialect.implicit_returning
+
+ # oracle 8 / 8i support returning
+ assert dialect.implicit_returning
+
assert not dialect._supports_char_length
assert not dialect.use_ansi
self.assert_compile(String(50), "VARCHAR2(50)", dialect=dialect)
self.assert_compile(Unicode(50), "VARCHAR2(50)", dialect=dialect)
self.assert_compile(UnicodeText(), "CLOB", dialect=dialect)
- dialect = self._dialect((8, 2, 5), implicit_returning=True)
- dialect.initialize(testing.db.connect())
- assert dialect.implicit_returning
-
def test_default_flags(self):
"""test with no initialization or server version info"""
from sqlalchemy.sql import operators
from sqlalchemy.sql import table
from sqlalchemy.sql import util as sql_util
-from sqlalchemy.testing import engines
-from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing.assertions import assert_raises
from sqlalchemy.testing.assertions import assert_raises_message
class SequenceTest(fixtures.TestBase, AssertsCompiledSQL):
- __prefer__ = "postgresql"
+ __dialect__ = "postgresql"
def test_format(self):
seq = Sequence("my_seq_no_schema")
== '"Some_Schema"."My_Seq"'
)
- @testing.only_on("postgresql", "foo")
- @testing.provide_metadata
- def test_reverse_eng_name(self):
- metadata = self.metadata
- engine = engines.testing_engine(options=dict(implicit_returning=False))
- for tname, cname in [
- ("tb1" * 30, "abc"),
- ("tb2", "abc" * 30),
- ("tb3" * 30, "abc" * 30),
- ("tb4", "abc"),
- ]:
- t = Table(
- tname[:57],
- metadata,
- Column(cname[:57], Integer, primary_key=True),
- )
- t.create(engine)
- with engine.begin() as conn:
- r = conn.execute(t.insert())
- eq_(r.inserted_primary_key, (1,))
-
@testing.combinations(
(None, ""),
(Integer, "AS INTEGER "),
connection.execute(text("CREATE SEQUENCE fooseq"))
t.create(connection, checkfirst=True)
+ @testing.combinations(True, False, argnames="implicit_returning")
+ def test_sequence_detection_tricky_names(
+ self, metadata, connection, implicit_returning
+ ):
+ for tname, cname in [
+ ("tb1" * 30, "abc"),
+ ("tb2", "abc" * 30),
+ ("tb3" * 30, "abc" * 30),
+ ("tb4", "abc"),
+ ]:
+ t = Table(
+ tname[:57],
+ metadata,
+ Column(cname[:57], Integer, primary_key=True),
+ implicit_returning=implicit_returning,
+ )
+ t.create(connection)
+ r = connection.execute(t.insert())
+ eq_(r.inserted_primary_key, (1,))
+
@testing.provide_metadata
def test_schema_roundtrips(self, connection):
meta = self.metadata
__only_on__ = "postgresql"
__backend__ = True
- def setup_test(self):
- self.metadata = MetaData()
+ @testing.combinations((False,), (True,), argnames="implicit_returning")
+ def test_foreignkey_missing_insert(
+ self, metadata, connection, implicit_returning
+ ):
- def teardown_test(self):
- with testing.db.begin() as conn:
- self.metadata.drop_all(conn)
-
- @testing.combinations((False,), (True,))
- def test_foreignkey_missing_insert(self, implicit_returning):
- engine = engines.testing_engine(
- options={"implicit_returning": implicit_returning}
+ Table(
+ "t1",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ implicit_returning=implicit_returning,
)
-
- Table("t1", self.metadata, Column("id", Integer, primary_key=True))
t2 = Table(
"t2",
- self.metadata,
+ metadata,
Column("id", Integer, ForeignKey("t1.id"), primary_key=True),
+ implicit_returning=implicit_returning,
)
- self.metadata.create_all(engine)
+ metadata.create_all(connection)
# want to ensure that "null value in column "id" violates not-
# null constraint" is raised (IntegrityError on psycoopg2, but
# the case here due to the foreign key.
with expect_warnings(".*has no Python-side or server-side default.*"):
- with engine.begin() as conn:
- assert_raises(
- (exc.IntegrityError, exc.ProgrammingError),
- conn.execute,
- t2.insert(),
- )
-
- def test_sequence_insert(self):
- table = Table(
- "testtable",
- self.metadata,
- Column("id", Integer, Sequence("my_seq"), primary_key=True),
- Column("data", String(30)),
- )
- self.metadata.create_all(testing.db)
- self._assert_data_with_sequence(table, "my_seq")
+ assert_raises(
+ (exc.IntegrityError, exc.ProgrammingError),
+ connection.execute,
+ t2.insert(),
+ )
- @testing.requires.returning
- def test_sequence_returning_insert(self):
+ @testing.combinations(True, False, argnames="implicit_returning")
+ def test_sequence_insert(self, metadata, connection, implicit_returning):
table = Table(
"testtable",
- self.metadata,
+ metadata,
Column("id", Integer, Sequence("my_seq"), primary_key=True),
Column("data", String(30)),
+ implicit_returning=implicit_returning,
)
- self.metadata.create_all(testing.db)
- self._assert_data_with_sequence_returning(table, "my_seq")
-
- def test_opt_sequence_insert(self):
- table = Table(
- "testtable",
- self.metadata,
- Column(
- "id",
- Integer,
- Sequence("my_seq", optional=True),
- primary_key=True,
- ),
- Column("data", String(30)),
- )
- self.metadata.create_all(testing.db)
- self._assert_data_autoincrement(table)
+ metadata.create_all(connection)
+ if implicit_returning:
+ self._assert_data_with_sequence_returning(
+ connection, table, "my_seq"
+ )
+ else:
+ self._assert_data_with_sequence(connection, table, "my_seq")
- @testing.requires.returning
- def test_opt_sequence_returning_insert(self):
+ @testing.combinations(True, False, argnames="implicit_returning")
+ def test_opt_sequence_insert(
+ self, metadata, connection, implicit_returning
+ ):
table = Table(
"testtable",
- self.metadata,
+ metadata,
Column(
"id",
Integer,
primary_key=True,
),
Column("data", String(30)),
+ implicit_returning=implicit_returning,
)
- self.metadata.create_all(testing.db)
- self._assert_data_autoincrement_returning(table)
-
- def test_autoincrement_insert(self):
- table = Table(
- "testtable",
- self.metadata,
- Column("id", Integer, primary_key=True),
- Column("data", String(30)),
- )
- self.metadata.create_all(testing.db)
- self._assert_data_autoincrement(table)
+ metadata.create_all(connection)
+ if implicit_returning:
+ self._assert_data_autoincrement_returning(connection, table)
+ else:
+ self._assert_data_autoincrement(connection, table)
- @testing.requires.returning
- def test_autoincrement_returning_insert(self):
+ @testing.combinations(True, False, argnames="implicit_returning")
+ def test_autoincrement_insert(
+ self, metadata, connection, implicit_returning
+ ):
table = Table(
"testtable",
- self.metadata,
+ metadata,
Column("id", Integer, primary_key=True),
Column("data", String(30)),
+ implicit_returning=implicit_returning,
)
- self.metadata.create_all(testing.db)
- self._assert_data_autoincrement_returning(table)
+ metadata.create_all(connection)
+ if implicit_returning:
+ self._assert_data_autoincrement_returning(connection, table)
+ else:
+ self._assert_data_autoincrement(connection, table)
- def test_noautoincrement_insert(self):
+ @testing.combinations(True, False, argnames="implicit_returning")
+ def test_noautoincrement_insert(
+ self, metadata, connection, implicit_returning
+ ):
table = Table(
"testtable",
- self.metadata,
+ metadata,
Column("id", Integer, primary_key=True, autoincrement=False),
Column("data", String(30)),
+ implicit_returning=implicit_returning,
)
- self.metadata.create_all(testing.db)
- self._assert_data_noautoincrement(table)
-
- def _assert_data_autoincrement(self, table):
- engine = engines.testing_engine(options={"implicit_returning": False})
+ metadata.create_all(connection)
+ self._assert_data_noautoincrement(connection, table)
- with self.sql_execution_asserter(engine) as asserter:
+ def _assert_data_autoincrement(self, connection, table):
+ """
+ invoked by:
+ * test_opt_sequence_insert
+ * test_autoincrement_insert
+ """
- with engine.begin() as conn:
- # execute with explicit id
+ with self.sql_execution_asserter(connection) as asserter:
+ conn = connection
+ # execute with explicit id
- r = conn.execute(table.insert(), {"id": 30, "data": "d1"})
- eq_(r.inserted_primary_key, (30,))
+ r = conn.execute(table.insert(), {"id": 30, "data": "d1"})
+ eq_(r.inserted_primary_key, (30,))
- # execute with prefetch id
+ # execute with prefetch id
- r = conn.execute(table.insert(), {"data": "d2"})
- eq_(r.inserted_primary_key, (1,))
+ r = conn.execute(table.insert(), {"data": "d2"})
+ eq_(r.inserted_primary_key, (1,))
- # executemany with explicit ids
+ # executemany with explicit ids
- conn.execute(
- table.insert(),
- [
- {"id": 31, "data": "d3"},
- {"id": 32, "data": "d4"},
- ],
- )
+ conn.execute(
+ table.insert(),
+ [
+ {"id": 31, "data": "d3"},
+ {"id": 32, "data": "d4"},
+ ],
+ )
- # executemany, uses SERIAL
+ # executemany, uses SERIAL
- conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
+ conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
- # single execute, explicit id, inline
+ # single execute, explicit id, inline
- conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
+ conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
- # single execute, inline, uses SERIAL
+ # single execute, inline, uses SERIAL
- conn.execute(table.insert().inline(), {"data": "d8"})
+ conn.execute(table.insert().inline(), {"data": "d8"})
asserter.assert_(
DialectSQL(
),
)
- with engine.begin() as conn:
- eq_(
- conn.execute(table.select()).fetchall(),
- [
- (30, "d1"),
- (1, "d2"),
- (31, "d3"),
- (32, "d4"),
- (2, "d5"),
- (3, "d6"),
- (33, "d7"),
- (4, "d8"),
- ],
- )
+ eq_(
+ conn.execute(table.select()).fetchall(),
+ [
+ (30, "d1"),
+ (1, "d2"),
+ (31, "d3"),
+ (32, "d4"),
+ (2, "d5"),
+ (3, "d6"),
+ (33, "d7"),
+ (4, "d8"),
+ ],
+ )
- conn.execute(table.delete())
+ conn.execute(table.delete())
# test the same series of events using a reflected version of
# the table
m2 = MetaData()
- table = Table(table.name, m2, autoload_with=engine)
-
- with self.sql_execution_asserter(engine) as asserter:
- with engine.begin() as conn:
- conn.execute(table.insert(), {"id": 30, "data": "d1"})
- r = conn.execute(table.insert(), {"data": "d2"})
- eq_(r.inserted_primary_key, (5,))
- conn.execute(
- table.insert(),
- [
- {"id": 31, "data": "d3"},
- {"id": 32, "data": "d4"},
- ],
- )
- conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
- conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
- conn.execute(table.insert().inline(), {"data": "d8"})
+ table = Table(
+ table.name, m2, autoload_with=connection, implicit_returning=False
+ )
+
+ with self.sql_execution_asserter(connection) as asserter:
+ conn.execute(table.insert(), {"id": 30, "data": "d1"})
+ r = conn.execute(table.insert(), {"data": "d2"})
+ eq_(r.inserted_primary_key, (5,))
+ conn.execute(
+ table.insert(),
+ [
+ {"id": 31, "data": "d3"},
+ {"id": 32, "data": "d4"},
+ ],
+ )
+ conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
+ conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
+ conn.execute(table.insert().inline(), {"data": "d8"})
asserter.assert_(
DialectSQL(
"INSERT INTO testtable (data) VALUES (:data)", [{"data": "d8"}]
),
)
- with engine.begin() as conn:
- eq_(
- conn.execute(table.select()).fetchall(),
- [
- (30, "d1"),
- (5, "d2"),
- (31, "d3"),
- (32, "d4"),
- (6, "d5"),
- (7, "d6"),
- (33, "d7"),
- (8, "d8"),
- ],
- )
- conn.execute(table.delete())
- def _assert_data_autoincrement_returning(self, table):
- engine = engines.testing_engine(options={"implicit_returning": True})
+ eq_(
+ conn.execute(table.select()).fetchall(),
+ [
+ (30, "d1"),
+ (5, "d2"),
+ (31, "d3"),
+ (32, "d4"),
+ (6, "d5"),
+ (7, "d6"),
+ (33, "d7"),
+ (8, "d8"),
+ ],
+ )
- with self.sql_execution_asserter(engine) as asserter:
- with engine.begin() as conn:
+ def _assert_data_autoincrement_returning(self, connection, table):
+ """
+ invoked by:
+ * test_opt_sequence_returning_insert
+ * test_autoincrement_returning_insert
+ """
+ with self.sql_execution_asserter(connection) as asserter:
+ conn = connection
- # execute with explicit id
+ # execute with explicit id
- r = conn.execute(table.insert(), {"id": 30, "data": "d1"})
- eq_(r.inserted_primary_key, (30,))
+ r = conn.execute(table.insert(), {"id": 30, "data": "d1"})
+ eq_(r.inserted_primary_key, (30,))
- # execute with prefetch id
+ # execute with prefetch id
- r = conn.execute(table.insert(), {"data": "d2"})
- eq_(r.inserted_primary_key, (1,))
+ r = conn.execute(table.insert(), {"data": "d2"})
+ eq_(r.inserted_primary_key, (1,))
- # executemany with explicit ids
+ # executemany with explicit ids
- conn.execute(
- table.insert(),
- [
- {"id": 31, "data": "d3"},
- {"id": 32, "data": "d4"},
- ],
- )
+ conn.execute(
+ table.insert(),
+ [
+ {"id": 31, "data": "d3"},
+ {"id": 32, "data": "d4"},
+ ],
+ )
- # executemany, uses SERIAL
+ # executemany, uses SERIAL
- conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
+ conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
- # single execute, explicit id, inline
+ # single execute, explicit id, inline
- conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
+ conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
- # single execute, inline, uses SERIAL
+ # single execute, inline, uses SERIAL
- conn.execute(table.insert().inline(), {"data": "d8"})
+ conn.execute(table.insert().inline(), {"data": "d8"})
asserter.assert_(
DialectSQL(
),
)
- with engine.begin() as conn:
- eq_(
- conn.execute(table.select()).fetchall(),
- [
- (30, "d1"),
- (1, "d2"),
- (31, "d3"),
- (32, "d4"),
- (2, "d5"),
- (3, "d6"),
- (33, "d7"),
- (4, "d8"),
- ],
- )
- conn.execute(table.delete())
+ eq_(
+ conn.execute(table.select()).fetchall(),
+ [
+ (30, "d1"),
+ (1, "d2"),
+ (31, "d3"),
+ (32, "d4"),
+ (2, "d5"),
+ (3, "d6"),
+ (33, "d7"),
+ (4, "d8"),
+ ],
+ )
+ conn.execute(table.delete())
# test the same series of events using a reflected version of
# the table
m2 = MetaData()
- table = Table(table.name, m2, autoload_with=engine)
-
- with self.sql_execution_asserter(engine) as asserter:
- with engine.begin() as conn:
- conn.execute(table.insert(), {"id": 30, "data": "d1"})
- r = conn.execute(table.insert(), {"data": "d2"})
- eq_(r.inserted_primary_key, (5,))
- conn.execute(
- table.insert(),
- [
- {"id": 31, "data": "d3"},
- {"id": 32, "data": "d4"},
- ],
- )
- conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
- conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
- conn.execute(table.insert().inline(), {"data": "d8"})
+ table = Table(
+ table.name,
+ m2,
+ autoload_with=connection,
+ implicit_returning=True,
+ )
+
+ with self.sql_execution_asserter(connection) as asserter:
+ conn.execute(table.insert(), {"id": 30, "data": "d1"})
+ r = conn.execute(table.insert(), {"data": "d2"})
+ eq_(r.inserted_primary_key, (5,))
+ conn.execute(
+ table.insert(),
+ [
+ {"id": 31, "data": "d3"},
+ {"id": 32, "data": "d4"},
+ ],
+ )
+ conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
+ conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
+ conn.execute(table.insert().inline(), {"data": "d8"})
asserter.assert_(
DialectSQL(
),
)
- with engine.begin() as conn:
- eq_(
- conn.execute(table.select()).fetchall(),
+ eq_(
+ conn.execute(table.select()).fetchall(),
+ [
+ (30, "d1"),
+ (5, "d2"),
+ (31, "d3"),
+ (32, "d4"),
+ (6, "d5"),
+ (7, "d6"),
+ (33, "d7"),
+ (8, "d8"),
+ ],
+ )
+
+ def _assert_data_with_sequence(self, connection, table, seqname):
+ """
+ invoked by:
+ * test_sequence_insert
+ """
+
+ with self.sql_execution_asserter(connection) as asserter:
+ conn = connection
+ conn.execute(table.insert(), {"id": 30, "data": "d1"})
+ conn.execute(table.insert(), {"data": "d2"})
+ conn.execute(
+ table.insert(),
[
- (30, "d1"),
- (5, "d2"),
- (31, "d3"),
- (32, "d4"),
- (6, "d5"),
- (7, "d6"),
- (33, "d7"),
- (8, "d8"),
+ {"id": 31, "data": "d3"},
+ {"id": 32, "data": "d4"},
],
)
- conn.execute(table.delete())
-
- def _assert_data_with_sequence(self, table, seqname):
- engine = engines.testing_engine(options={"implicit_returning": False})
-
- with self.sql_execution_asserter(engine) as asserter:
- with engine.begin() as conn:
- conn.execute(table.insert(), {"id": 30, "data": "d1"})
- conn.execute(table.insert(), {"data": "d2"})
- conn.execute(
- table.insert(),
- [
- {"id": 31, "data": "d3"},
- {"id": 32, "data": "d4"},
- ],
- )
- conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
- conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
- conn.execute(table.insert().inline(), {"data": "d8"})
+ conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
+ conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
+ conn.execute(table.insert().inline(), {"data": "d8"})
asserter.assert_(
DialectSQL(
[{"data": "d8"}],
),
)
- with engine.begin() as conn:
- eq_(
- conn.execute(table.select()).fetchall(),
+ eq_(
+ conn.execute(table.select()).fetchall(),
+ [
+ (30, "d1"),
+ (1, "d2"),
+ (31, "d3"),
+ (32, "d4"),
+ (2, "d5"),
+ (3, "d6"),
+ (33, "d7"),
+ (4, "d8"),
+ ],
+ )
+
+ def _assert_data_with_sequence_returning(self, connection, table, seqname):
+ """
+ invoked by:
+ * test_sequence_returning_insert
+ """
+
+ with self.sql_execution_asserter(connection) as asserter:
+ conn = connection
+ conn.execute(table.insert(), {"id": 30, "data": "d1"})
+ conn.execute(table.insert(), {"data": "d2"})
+ conn.execute(
+ table.insert(),
[
- (30, "d1"),
- (1, "d2"),
- (31, "d3"),
- (32, "d4"),
- (2, "d5"),
- (3, "d6"),
- (33, "d7"),
- (4, "d8"),
+ {"id": 31, "data": "d3"},
+ {"id": 32, "data": "d4"},
],
)
-
- # cant test reflection here since the Sequence must be
- # explicitly specified
-
- def _assert_data_with_sequence_returning(self, table, seqname):
- engine = engines.testing_engine(options={"implicit_returning": True})
-
- with self.sql_execution_asserter(engine) as asserter:
- with engine.begin() as conn:
- conn.execute(table.insert(), {"id": 30, "data": "d1"})
- conn.execute(table.insert(), {"data": "d2"})
- conn.execute(
- table.insert(),
- [
- {"id": 31, "data": "d3"},
- {"id": 32, "data": "d4"},
- ],
- )
- conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
- conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
- conn.execute(table.insert().inline(), {"data": "d8"})
+ conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
+ conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
+ conn.execute(table.insert().inline(), {"data": "d8"})
asserter.assert_(
DialectSQL(
),
)
- with engine.begin() as conn:
- eq_(
- conn.execute(table.select()).fetchall(),
- [
- (30, "d1"),
- (1, "d2"),
- (31, "d3"),
- (32, "d4"),
- (2, "d5"),
- (3, "d6"),
- (33, "d7"),
- (4, "d8"),
- ],
- )
-
- # cant test reflection here since the Sequence must be
- # explicitly specified
+ eq_(
+ connection.execute(table.select()).fetchall(),
+ [
+ (30, "d1"),
+ (1, "d2"),
+ (31, "d3"),
+ (32, "d4"),
+ (2, "d5"),
+ (3, "d6"),
+ (33, "d7"),
+ (4, "d8"),
+ ],
+ )
- def _assert_data_noautoincrement(self, table):
- engine = engines.testing_engine(options={"implicit_returning": False})
+ def _assert_data_noautoincrement(self, connection, table):
+ """
+ invoked by:
+ * test_noautoincrement_insert
+ """
# turning off the cache because we are checking for compile-time
# warnings
- engine = engine.execution_options(compiled_cache=None)
+ connection.execution_options(compiled_cache=None)
- with engine.begin() as conn:
- conn.execute(table.insert(), {"id": 30, "data": "d1"})
+ conn = connection
+ conn.execute(table.insert(), {"id": 30, "data": "d1"})
- with engine.begin() as conn:
+ with conn.begin_nested() as nested:
with expect_warnings(
".*has no Python-side or server-side default.*"
):
table.insert(),
{"data": "d2"},
)
+ nested.rollback()
- with engine.begin() as conn:
+ with conn.begin_nested() as nested:
with expect_warnings(
".*has no Python-side or server-side default.*"
):
table.insert(),
[{"data": "d2"}, {"data": "d3"}],
)
+ nested.rollback()
- with engine.begin() as conn:
+ with conn.begin_nested() as nested:
with expect_warnings(
".*has no Python-side or server-side default.*"
):
table.insert(),
{"data": "d2"},
)
+ nested.rollback()
- with engine.begin() as conn:
+ with conn.begin_nested() as nested:
with expect_warnings(
".*has no Python-side or server-side default.*"
):
table.insert(),
[{"data": "d2"}, {"data": "d3"}],
)
+ nested.rollback()
- with engine.begin() as conn:
- conn.execute(
- table.insert(),
- [{"id": 31, "data": "d2"}, {"id": 32, "data": "d3"}],
- )
- conn.execute(table.insert().inline(), {"id": 33, "data": "d4"})
- eq_(
- conn.execute(table.select()).fetchall(),
- [(30, "d1"), (31, "d2"), (32, "d3"), (33, "d4")],
- )
- conn.execute(table.delete())
+ conn.execute(
+ table.insert(),
+ [{"id": 31, "data": "d2"}, {"id": 32, "data": "d3"}],
+ )
+ conn.execute(table.insert().inline(), {"id": 33, "data": "d4"})
+ eq_(
+ conn.execute(table.select()).fetchall(),
+ [(30, "d1"), (31, "d2"), (32, "d3"), (33, "d4")],
+ )
+ conn.execute(table.delete())
# test the same series of events using a reflected version of
# the table
m2 = MetaData()
- table = Table(table.name, m2, autoload_with=engine)
- with engine.begin() as conn:
- conn.execute(table.insert(), {"id": 30, "data": "d1"})
+ table = Table(table.name, m2, autoload_with=connection)
+ conn = connection
+
+ conn.execute(table.insert(), {"id": 30, "data": "d1"})
- with engine.begin() as conn:
+ with conn.begin_nested() as nested:
with expect_warnings(
".*has no Python-side or server-side default.*"
):
table.insert(),
{"data": "d2"},
)
+ nested.rollback()
- with engine.begin() as conn:
+ with conn.begin_nested() as nested:
with expect_warnings(
".*has no Python-side or server-side default.*"
):
table.insert(),
[{"data": "d2"}, {"data": "d3"}],
)
+ nested.rollback()
- with engine.begin() as conn:
- conn.execute(
- table.insert(),
- [{"id": 31, "data": "d2"}, {"id": 32, "data": "d3"}],
- )
- conn.execute(table.insert().inline(), {"id": 33, "data": "d4"})
- eq_(
- conn.execute(table.select()).fetchall(),
- [(30, "d1"), (31, "d2"), (32, "d3"), (33, "d4")],
- )
+ conn.execute(
+ table.insert(),
+ [{"id": 31, "data": "d2"}, {"id": 32, "data": "d3"}],
+ )
+ conn.execute(table.insert().inline(), {"id": 33, "data": "d4"})
+ eq_(
+ conn.execute(table.select()).fetchall(),
+ [(30, "d1"), (31, "d2"), (32, "d3"), (33, "d4")],
+ )
class MatchTest(fixtures.TablesTest, AssertsCompiledSQL):
from sqlalchemy import event
from sqlalchemy import exc
from sqlalchemy import ForeignKey
+from sqlalchemy import insert
from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy.engine.mock import MockConnection
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assertions
from sqlalchemy.testing import config
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import is_true
from sqlalchemy.testing import mock
from sqlalchemy.testing.assertions import expect_deprecated
+from sqlalchemy.testing.assertions import expect_raises_message
from sqlalchemy.testing.engines import testing_engine
from sqlalchemy.testing.mock import Mock
from sqlalchemy.testing.schema import Column
with e1.connect() as conn:
result = conn.execute(select(1))
result.close()
+
+
+ce_implicit_returning = (
+ r"The create_engine.implicit_returning parameter is deprecated "
+ r"and will be removed in a future release."
+)
+
+
+class ImplicitReturningFlagTest(fixtures.TestBase):
+ __backend__ = True
+
+ @testing.combinations(True, False, None, argnames="implicit_returning")
+ def test_implicit_returning_engine_parameter(self, implicit_returning):
+ if implicit_returning is None:
+ e = engines.testing_engine()
+ else:
+ with assertions.expect_deprecated(ce_implicit_returning):
+ e = engines.testing_engine(
+ options={"implicit_returning": implicit_returning}
+ )
+
+ if implicit_returning is None:
+ eq_(
+ e.dialect.implicit_returning,
+ testing.db.dialect.implicit_returning,
+ )
+ else:
+ eq_(e.dialect.implicit_returning, implicit_returning)
+
+ t = Table(
+ "t",
+ MetaData(),
+ Column("id", Integer, primary_key=True),
+ Column("data", String(50)),
+ )
+
+ t2 = Table(
+ "t",
+ MetaData(),
+ Column("id", Integer, primary_key=True),
+ Column("data", String(50)),
+ implicit_returning=False,
+ )
+
+ with e.connect() as conn:
+ stmt = insert(t).values(data="data")
+
+ if implicit_returning:
+ if not testing.requires.returning.enabled:
+ with expect_raises_message(
+ exc.CompileError, "RETURNING is not supported"
+ ):
+ stmt.compile(conn)
+ else:
+ eq_(stmt.compile(conn).returning, [t.c.id])
+ elif (
+ implicit_returning is None
+ and testing.db.dialect.implicit_returning
+ ):
+ eq_(stmt.compile(conn).returning, [t.c.id])
+ else:
+ eq_(stmt.compile(conn).returning, [])
+
+ # table setting it to False disables it
+ stmt2 = insert(t2).values(data="data")
+ eq_(stmt2.compile(conn).returning, [])
# TODO: this test is kind of a mess
for engine in [
- engines.testing_engine(options=dict(implicit_returning=False)),
- engines.testing_engine(
- options=dict(implicit_returning=False)
- ).connect(),
+ engines.testing_engine(),
+ engines.testing_engine().connect(),
]:
event.listen(engine, "before_execute", execute)
event.listen(engine, "before_cursor_execute", cursor_execute)
default=func.lower("Foo"),
primary_key=True,
),
+ implicit_returning=False,
)
if isinstance(engine, Connection):
"""for mssql, autoincrement means IDENTITY, not sequence"""
return skip_if("mssql")
+ @property
+ def supports_autoincrement_w_composite_pk(self):
+ """integer autoincrement works for tables with composite primary
+ keys"""
+ return fails_if("sqlite")
+
@property
def identity_columns(self):
return only_if(["postgresql >= 10", "oracle >= 12", "mssql"])
from sqlalchemy.sql import text
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
-from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
eq_(conn.execute(select(q.c.x, q.c.y)).first(), expected)
-class PKDefaultTest(fixtures.TablesTest):
+class PKDefaultTest(fixtures.TestBase):
__requires__ = ("subqueries",)
__backend__ = True
- @classmethod
- def define_tables(cls, metadata):
- t2 = Table("t2", metadata, Column("nextid", Integer))
+ @testing.fixture
+ def table_fixture(self, metadata, connection):
+ def go(implicit_returning):
+ t2 = Table(
+ "t2",
+ metadata,
+ Column("nextid", Integer),
+ implicit_returning=implicit_returning,
+ )
- Table(
- "t1",
- metadata,
- Column(
- "id",
- Integer,
- primary_key=True,
- default=sa.select(func.max(t2.c.nextid)).scalar_subquery(),
- ),
- Column("data", String(30)),
- )
+ t1 = Table(
+ "t1",
+ metadata,
+ Column(
+ "id",
+ Integer,
+ primary_key=True,
+ default=sa.select(func.max(t2.c.nextid)).scalar_subquery(),
+ ),
+ Column("data", String(30)),
+ implicit_returning=implicit_returning,
+ )
- Table(
- "date_table",
- metadata,
- Column(
- "date_id",
- DateTime(timezone=True),
- default=text("current_timestamp"),
- primary_key=True,
- ),
- )
+ date_table = Table(
+ "date_table",
+ metadata,
+ Column(
+ "date_id",
+ DateTime(timezone=True),
+ default=text("current_timestamp"),
+ primary_key=True,
+ ),
+ implicit_returning=implicit_returning,
+ )
- @testing.requires.returning
- def test_with_implicit_returning(self):
- self._test(True)
+ metadata.create_all(connection)
+ return t1, t2, date_table
- def test_regular(self):
- self._test(False)
+ return go
- def _test(self, returning):
- t2, t1, date_table = (
- self.tables.t2,
- self.tables.t1,
- self.tables.date_table,
- )
+ @testing.combinations(
+ (True, testing.requires.returning),
+ (False,),
+ argnames="implicit_returning",
+ )
+ def test_pk_default(self, connection, table_fixture, implicit_returning):
+ t1, t2, date_table = table_fixture(implicit_returning)
- if not returning and not testing.db.dialect.implicit_returning:
- engine = testing.db
- else:
- engine = engines.testing_engine(
- options={"implicit_returning": returning}
- )
- with engine.begin() as conn:
- conn.execute(t2.insert(), dict(nextid=1))
- r = conn.execute(t1.insert(), dict(data="hi"))
- eq_((1,), r.inserted_primary_key)
+ conn = connection
+
+ conn.execute(t2.insert(), dict(nextid=1))
+ r = conn.execute(t1.insert(), dict(data="hi"))
+ eq_((1,), r.inserted_primary_key)
- conn.execute(t2.insert(), dict(nextid=2))
- r = conn.execute(t1.insert(), dict(data="there"))
- eq_((2,), r.inserted_primary_key)
+ conn.execute(t2.insert(), dict(nextid=2))
+ r = conn.execute(t1.insert(), dict(data="there"))
+ eq_((2,), r.inserted_primary_key)
- r = conn.execute(date_table.insert())
- assert isinstance(r.inserted_primary_key[0], datetime.datetime)
+ r = conn.execute(date_table.insert())
+ assert isinstance(r.inserted_primary_key[0], datetime.datetime)
class PKIncrementTest(fixtures.TablesTest):
from sqlalchemy import INT
from sqlalchemy import Integer
from sqlalchemy import literal
-from sqlalchemy import MetaData
from sqlalchemy import Sequence
from sqlalchemy import sql
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy import VARCHAR
from sqlalchemy.testing import assert_raises_message
-from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
+from sqlalchemy.testing import mock
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
],
)
- def _test_lastrow_accessor(self, table_, values, assertvalues):
+ def _test_lastrow_accessor(self, connection, table_, values, assertvalues):
"""Tests the inserted_primary_key and lastrow_has_id() functions."""
- def insert_values(engine, table_, values):
+ def insert_values(table_, values):
"""
Inserts a row into a table, returns the full list of values
INSERTed including defaults that fired off on the DB side and
"""
# verify implicit_returning is working
- if engine.dialect.implicit_returning:
+ if (
+ connection.dialect.implicit_returning
+ and table_.implicit_returning
+ ):
ins = table_.insert()
- comp = ins.compile(engine, column_keys=list(values))
+ comp = ins.compile(connection, column_keys=list(values))
if not set(values).issuperset(
c.key for c in table_.primary_key
):
is_(bool(comp.returning), True)
- with engine.begin() as connection:
- result = connection.execute(table_.insert(), values)
- ret = values.copy()
-
- ipk = result.inserted_primary_key
- for col, id_ in zip(table_.primary_key, ipk):
- ret[col.key] = id_
-
- if result.lastrow_has_defaults():
- criterion = and_(
- *[
- col == id_
- for col, id_ in zip(
- table_.primary_key, result.inserted_primary_key
- )
- ]
- )
- row = connection.execute(
- table_.select().where(criterion)
- ).first()
- for c in table_.c:
- ret[c.key] = row._mapping[c]
+ result = connection.execute(table_.insert(), values)
+ ret = values.copy()
+
+ ipk = result.inserted_primary_key
+ for col, id_ in zip(table_.primary_key, ipk):
+ ret[col.key] = id_
+
+ if result.lastrow_has_defaults():
+ criterion = and_(
+ *[
+ col == id_
+ for col, id_ in zip(
+ table_.primary_key, result.inserted_primary_key
+ )
+ ]
+ )
+ row = connection.execute(
+ table_.select().where(criterion)
+ ).first()
+ for c in table_.c:
+ ret[c.key] = row._mapping[c]
return ret, ipk
- if testing.against("postgresql", "oracle", "mssql"):
- assert testing.db.dialect.implicit_returning
-
- if testing.db.dialect.implicit_returning:
- test_engines = [
- engines.testing_engine(options={"implicit_returning": False}),
- engines.testing_engine(options={"implicit_returning": True}),
- ]
- else:
- test_engines = [testing.db]
-
- for engine in test_engines:
- try:
- table_.create(bind=engine, checkfirst=True)
- i, ipk = insert_values(engine, table_, values)
- eq_(i, assertvalues)
+ table_.create(connection, checkfirst=True)
+ i, ipk = insert_values(table_, values)
+ eq_(i, assertvalues)
- # named tuple tests
- for col in table_.primary_key:
- eq_(getattr(ipk, col.key), assertvalues[col.key])
- eq_(ipk._mapping[col.key], assertvalues[col.key])
+ # named tuple tests
+ for col in table_.primary_key:
+ eq_(getattr(ipk, col.key), assertvalues[col.key])
+ eq_(ipk._mapping[col.key], assertvalues[col.key])
- eq_(
- ipk._fields, tuple([col.key for col in table_.primary_key])
- )
-
- finally:
- table_.drop(bind=engine)
+ eq_(ipk._fields, tuple([col.key for col in table_.primary_key]))
- @testing.skip_if("sqlite")
- def test_lastrow_accessor_one(self):
- metadata = MetaData()
+ @testing.requires.supports_autoincrement_w_composite_pk
+ @testing.combinations(
+ (True, testing.requires.returning),
+ (False,),
+ argnames="implicit_returning",
+ )
+ def test_lastrow_accessor_one(
+ self, metadata, connection, implicit_returning
+ ):
self._test_lastrow_accessor(
+ connection,
Table(
"t1",
metadata,
test_needs_autoincrement=True,
),
Column("foo", String(30), primary_key=True),
+ implicit_returning=implicit_returning,
),
{"foo": "hi"},
{"id": 1, "foo": "hi"},
)
- @testing.skip_if("sqlite")
- def test_lastrow_accessor_two(self):
- metadata = MetaData()
+ @testing.requires.supports_autoincrement_w_composite_pk
+ @testing.combinations(
+ (True, testing.requires.returning),
+ (False,),
+ argnames="implicit_returning",
+ )
+ def test_lastrow_accessor_two(
+ self, metadata, connection, implicit_returning
+ ):
self._test_lastrow_accessor(
+ connection,
Table(
"t2",
metadata,
),
Column("foo", String(30), primary_key=True),
Column("bar", String(30), server_default="hi"),
+ implicit_returning=implicit_returning,
),
{"foo": "hi"},
{"id": 1, "foo": "hi", "bar": "hi"},
)
- def test_lastrow_accessor_three(self):
- metadata = MetaData()
+ @testing.combinations(
+ (True, testing.requires.returning),
+ (False,),
+ argnames="implicit_returning",
+ )
+ def test_lastrow_accessor_three(
+ self, metadata, connection, implicit_returning
+ ):
self._test_lastrow_accessor(
+ connection,
Table(
"t3",
metadata,
Column("id", String(40), primary_key=True),
Column("foo", String(30), primary_key=True),
Column("bar", String(30)),
+ implicit_returning=implicit_returning,
),
{"id": "hi", "foo": "thisisfoo", "bar": "thisisbar"},
{"id": "hi", "foo": "thisisfoo", "bar": "thisisbar"},
)
@testing.requires.sequences
- def test_lastrow_accessor_four(self):
- metadata = MetaData()
+ @testing.combinations(
+ (True, testing.requires.returning),
+ (False,),
+ argnames="implicit_returning",
+ )
+ def test_lastrow_accessor_four(
+ self, metadata, connection, implicit_returning
+ ):
self._test_lastrow_accessor(
+ connection,
Table(
"t4",
metadata,
),
Column("foo", String(30), primary_key=True),
Column("bar", String(30), server_default="hi"),
+ implicit_returning=implicit_returning,
),
{"foo": "hi", "id": 1},
{"id": 1, "foo": "hi", "bar": "hi"},
)
@testing.requires.sequences
- def test_lastrow_accessor_four_a(self):
- metadata = MetaData()
+ @testing.combinations(
+ (True, testing.requires.returning),
+ (False,),
+ argnames="implicit_returning",
+ )
+ def test_lastrow_accessor_four_a(
+ self, metadata, connection, implicit_returning
+ ):
self._test_lastrow_accessor(
+ connection,
Table(
"t4",
metadata,
primary_key=True,
),
Column("foo", String(30)),
+ implicit_returning=implicit_returning,
),
{"foo": "hi"},
{"id": 1, "foo": "hi"},
)
- def test_lastrow_accessor_five(self):
- metadata = MetaData()
+ @testing.combinations(
+ (True, testing.requires.returning),
+ (False,),
+ argnames="implicit_returning",
+ )
+ def test_lastrow_accessor_five(
+ self, metadata, connection, implicit_returning
+ ):
self._test_lastrow_accessor(
+ connection,
Table(
"t5",
metadata,
Column("id", String(10), primary_key=True),
Column("bar", String(30), server_default="hi"),
+ implicit_returning=implicit_returning,
),
{"id": "id1"},
{"id": "id1", "bar": "hi"},
)
- @testing.skip_if("sqlite")
- def test_lastrow_accessor_six(self):
- metadata = MetaData()
+ @testing.requires.supports_autoincrement_w_composite_pk
+ @testing.combinations(
+ (True, testing.requires.returning),
+ (False,),
+ argnames="implicit_returning",
+ )
+ def test_lastrow_accessor_six(
+ self, metadata, connection, implicit_returning
+ ):
self._test_lastrow_accessor(
+ connection,
Table(
"t6",
metadata,
test_needs_autoincrement=True,
),
Column("bar", Integer, primary_key=True),
+ implicit_returning=implicit_returning,
),
{"bar": 0},
{"id": 1, "bar": 0},
# TODO: why not in the sqlite suite?
@testing.only_on("sqlite+pysqlite")
- @testing.provide_metadata
- def test_lastrowid_zero(self):
+ def test_lastrowid_zero(self, metadata, connection):
from sqlalchemy.dialects import sqlite
- eng = engines.testing_engine()
-
class ExcCtx(sqlite.base.SQLiteExecutionContext):
def get_lastrowid(self):
return 0
- eng.dialect.execution_ctx_cls = ExcCtx
t = Table(
"t",
self.metadata,
Column("x", Integer, primary_key=True),
Column("y", Integer),
)
- with eng.begin() as conn:
- t.create(conn)
- r = conn.execute(t.insert().values(y=5))
+ t.create(connection)
+ with mock.patch.object(
+ connection.dialect, "execution_ctx_cls", ExcCtx
+ ):
+ r = connection.execute(t.insert().values(y=5))
eq_(r.inserted_primary_key, (0,))
- @testing.fails_on(
- "sqlite", "sqlite autoincrement doesn't work with composite pks"
- )
- @testing.provide_metadata
- def test_misordered_lastrow(self, connection):
- metadata = self.metadata
-
+ @testing.requires.supports_autoincrement_w_composite_pk
+ def test_misordered_lastrow(self, connection, metadata):
related = Table(
"related",
metadata,
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import AssertsExecutionResults
-from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing.schema import Column
result.inserted_primary_key_rows,
[(10,), (11,), (12,), (13,), (14,), (15,)],
)
-
-
-class ImplicitReturningFlag(fixtures.TestBase):
- __backend__ = True
-
- def test_flag_turned_off(self):
- e = engines.testing_engine(options={"implicit_returning": False})
- assert e.dialect.implicit_returning is False
- c = e.connect()
- c.close()
- assert e.dialect.implicit_returning is False
-
- def test_flag_turned_on(self):
- e = engines.testing_engine(options={"implicit_returning": True})
- assert e.dialect.implicit_returning is True
- c = e.connect()
- c.close()
- assert e.dialect.implicit_returning is True
-
- def test_flag_turned_default(self):
- supports = [False]
-
- def go():
- supports[0] = True
-
- testing.requires.returning(go)()
- e = engines.testing_engine()
-
- # version detection on connect sets it
- c = e.connect()
- c.close()
- assert e.dialect.implicit_returning is supports[0]
from sqlalchemy.schema import DropSequence
from sqlalchemy.sql import select
from sqlalchemy.testing import assert_raises_message
-from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_false
connection.execute(t1.insert().values(x=s.next_value()))
self._assert_seq_result(connection.scalar(t1.select()))
- @testing.provide_metadata
- def test_inserted_pk_no_returning(self):
+ def test_inserted_pk_no_returning(self, metadata, connection):
"""test inserted_primary_key contains [None] when
pk_col=next_value(), implicit returning is not used."""
# I'm not really sure what this test wants to accomlish.
- metadata = self.metadata
- t1 = Table("t", metadata, Column("x", Integer, primary_key=True))
+ t1 = Table(
+ "t",
+ metadata,
+ Column("x", Integer, primary_key=True),
+ implicit_returning=False,
+ )
s = Sequence("my_sequence_here", metadata=metadata)
- e = engines.testing_engine(options={"implicit_returning": False})
- with e.begin() as conn:
-
- t1.create(conn)
- s.create(conn)
+ conn = connection
+ t1.create(conn)
+ s.create(conn)
- r = conn.execute(t1.insert().values(x=s.next_value()))
+ r = conn.execute(t1.insert().values(x=s.next_value()))
- if testing.requires.emulated_lastrowid_even_with_sequences.enabled:
- eq_(r.inserted_primary_key, (1,))
- else:
- eq_(r.inserted_primary_key, (None,))
+ if testing.requires.emulated_lastrowid_even_with_sequences.enabled:
+ eq_(r.inserted_primary_key, (1,))
+ else:
+ eq_(r.inserted_primary_key, (None,))
@testing.combinations(
("implicit_returning",),
argnames="returning",
)
@testing.requires.multivalues_inserts
- def test_seq_multivalues_inline(self, metadata, testing_engine, returning):
+ def test_seq_multivalues_inline(self, metadata, connection, returning):
+ _implicit_returning = "no_implicit_returning" not in returning
t1 = Table(
"t",
metadata,
Column("x", Integer, Sequence("my_seq"), primary_key=True),
Column("data", String(50)),
+ implicit_returning=_implicit_returning,
)
- e = engines.testing_engine(
- options={
- "implicit_returning": "no_implicit_returning" not in returning
- }
+ metadata.create_all(connection)
+ conn = connection
+
+ stmt = t1.insert().values(
+ [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
)
- metadata.create_all(e)
- with e.begin() as conn:
+ if returning == "explicit_returning":
+ stmt = stmt.returning(t1.c.x)
+ elif "return_defaults" in returning:
+ stmt = stmt.return_defaults()
- stmt = t1.insert().values(
- [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
- )
- if returning == "explicit_returning":
- stmt = stmt.returning(t1.c.x)
- elif "return_defaults" in returning:
- stmt = stmt.return_defaults()
-
- r = conn.execute(stmt)
- if returning == "explicit_returning":
- eq_(r.all(), [(1,), (2,), (3,)])
- elif "return_defaults" in returning:
- eq_(r.returned_defaults_rows, None)
+ r = conn.execute(stmt)
+ if returning == "explicit_returning":
+ eq_(r.all(), [(1,), (2,), (3,)])
+ elif "return_defaults" in returning:
+ eq_(r.returned_defaults_rows, None)
- # TODO: not sure what this is
- eq_(r.inserted_primary_key_rows, [(None,)])
+ # TODO: not sure what this is
+ eq_(r.inserted_primary_key_rows, [(None,)])
- eq_(
- conn.execute(t1.select().order_by(t1.c.x)).all(),
- [(1, "d1"), (2, "d2"), (3, "d3")],
- )
+ eq_(
+ conn.execute(t1.select().order_by(t1.c.x)).all(),
+ [(1, "d1"), (2, "d2"), (3, "d3")],
+ )
@testing.combinations(
("implicit_returning",),
argnames="returning",
)
def test_seq_multivalues_executemany(
- self, metadata, testing_engine, returning
+ self, connection, metadata, returning
):
+ _implicit_returning = "no_implicit_returning" not in returning
t1 = Table(
"t",
metadata,
Column("x", Integer, Sequence("my_seq"), primary_key=True),
Column("data", String(50)),
+ implicit_returning=_implicit_returning,
)
- e = engines.testing_engine(
- options={
- "implicit_returning": "no_implicit_returning" not in returning
- }
- )
- metadata.create_all(e)
- with e.begin() as conn:
+ metadata.create_all(connection)
+ conn = connection
- stmt = t1.insert()
- if returning == "explicit_returning":
- stmt = stmt.returning(t1.c.x)
- elif "return_defaults" in returning:
- stmt = stmt.return_defaults()
+ stmt = t1.insert()
+ if returning == "explicit_returning":
+ stmt = stmt.returning(t1.c.x)
+ elif "return_defaults" in returning:
+ stmt = stmt.return_defaults()
- r = conn.execute(
- stmt, [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
- )
- if returning == "explicit_returning":
- eq_(r.all(), [(1,), (2,), (3,)])
- elif "return_defaults" in returning:
- if "no_implicit_returning" in returning:
- eq_(r.returned_defaults_rows, None)
- eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
- else:
- eq_(r.returned_defaults_rows, [(1,), (2,), (3,)])
- eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
-
- eq_(
- conn.execute(t1.select().order_by(t1.c.x)).all(),
- [(1, "d1"), (2, "d2"), (3, "d3")],
- )
+ r = conn.execute(
+ stmt, [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
+ )
+ if returning == "explicit_returning":
+ eq_(r.all(), [(1,), (2,), (3,)])
+ elif "return_defaults" in returning:
+ if "no_implicit_returning" in returning:
+ eq_(r.returned_defaults_rows, None)
+ eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
+ else:
+ eq_(r.returned_defaults_rows, [(1,), (2,), (3,)])
+ eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
+
+ eq_(
+ conn.execute(t1.select().order_by(t1.c.x)).all(),
+ [(1, "d1"), (2, "d2"), (3, "d3")],
+ )
@testing.requires.returning
- @testing.provide_metadata
- def test_inserted_pk_implicit_returning(self):
+ def test_inserted_pk_implicit_returning(self, connection, metadata):
"""test inserted_primary_key contains the result when
pk_col=next_value(), when implicit returning is used."""
- metadata = self.metadata
s = Sequence("my_sequence")
t1 = Table(
"t",
Integer,
primary_key=True,
),
+ implicit_returning=True,
)
- t1.create(testing.db)
+ t1.create(connection)
- e = engines.testing_engine(options={"implicit_returning": True})
- with e.begin() as conn:
- r = conn.execute(t1.insert().values(x=s.next_value()))
- self._assert_seq_result(r.inserted_primary_key[0])
+ r = connection.execute(t1.insert().values(x=s.next_value()))
+ self._assert_seq_result(r.inserted_primary_key[0])
class SequenceTest(fixtures.TestBase, testing.AssertsCompiledSQL):
__requires__ = ("sequences",)
__backend__ = True
- @classmethod
- def define_tables(cls, metadata):
- Table(
- "cartitems",
- metadata,
- Column(
- "cart_id",
- Integer,
- Sequence("cart_id_seq"),
- primary_key=True,
- autoincrement=False,
- ),
- Column("description", String(40)),
- Column("createdate", sa.DateTime()),
- )
+ @testing.fixture
+ def table_fixture(self, metadata, connection, implicit_returning):
+ def go(implicit_returning):
+ cartitems = Table(
+ "cartitems",
+ metadata,
+ Column(
+ "cart_id",
+ Integer,
+ Sequence("cart_id_seq"),
+ primary_key=True,
+ autoincrement=False,
+ ),
+ Column("description", String(40)),
+ Column("createdate", sa.DateTime()),
+ implicit_returning=implicit_returning,
+ )
- # a little bit of implicit case sensitive naming test going on here
- Table(
- "Manager",
- metadata,
- Column(
- "obj_id",
- Integer,
- Sequence("obj_id_seq"),
- ),
- Column("name", String(128)),
- Column(
- "id",
- Integer,
- Sequence("Manager_id_seq", optional=True),
- primary_key=True,
- ),
- )
+ # a little bit of implicit case sensitive naming test going on here
+ Manager = Table(
+ "Manager",
+ metadata,
+ Column(
+ "obj_id",
+ Integer,
+ Sequence("obj_id_seq"),
+ ),
+ Column("name", String(128)),
+ Column(
+ "id",
+ Integer,
+ Sequence("Manager_id_seq", optional=True),
+ primary_key=True,
+ ),
+ implicit_returning=implicit_returning,
+ )
+ metadata.create_all(connection)
+ return Manager, cartitems
- def test_insert_via_seq(self, connection):
- cartitems = self.tables.cartitems
+ return go
+
+ @testing.combinations(
+ (True, testing.requires.returning),
+ (False,),
+ argnames="implicit_returning",
+ )
+ def test_insert_via_seq(
+ self, table_fixture, connection, implicit_returning
+ ):
+ Manager, cartitems = table_fixture(implicit_returning)
connection.execute(cartitems.insert(), dict(description="hi"))
connection.execute(cartitems.insert(), dict(description="there"))
expected,
)
- def test_seq_nonpk(self):
+ @testing.combinations(
+ (True, testing.requires.returning),
+ (False,),
+ argnames="implicit_returning",
+ )
+ def test_seq_nonpk(self, connection, table_fixture, implicit_returning):
"""test sequences fire off as defaults on non-pk columns"""
- sometable = self.tables.Manager
-
- engine = engines.testing_engine(options={"implicit_returning": False})
+ sometable, cartitems = table_fixture(implicit_returning)
- with engine.begin() as conn:
- result = conn.execute(sometable.insert(), dict(name="somename"))
+ conn = connection
+ result = conn.execute(sometable.insert(), dict(name="somename"))
- eq_(result.postfetch_cols(), [sometable.c.obj_id])
+ eq_(result.postfetch_cols(), [sometable.c.obj_id])
- result = conn.execute(sometable.insert(), dict(name="someother"))
+ result = conn.execute(sometable.insert(), dict(name="someother"))
- conn.execute(
- sometable.insert(), [{"name": "name3"}, {"name": "name4"}]
- )
+ conn.execute(
+ sometable.insert(), [{"name": "name3"}, {"name": "name4"}]
+ )
- dsb = testing.db.dialect.default_sequence_base
- eq_(
- list(
- conn.execute(sometable.select().order_by(sometable.c.id))
+ dsb = testing.db.dialect.default_sequence_base
+ eq_(
+ list(conn.execute(sometable.select().order_by(sometable.c.id))),
+ [
+ (
+ dsb,
+ "somename",
+ dsb,
),
- [
- (
- dsb,
- "somename",
- dsb,
- ),
- (
- dsb + 1,
- "someother",
- dsb + 1,
- ),
- (
- dsb + 2,
- "name3",
- dsb + 2,
- ),
- (
- dsb + 3,
- "name4",
- dsb + 3,
- ),
- ],
- )
+ (
+ dsb + 1,
+ "someother",
+ dsb + 1,
+ ),
+ (
+ dsb + 2,
+ "name3",
+ dsb + 2,
+ ),
+ (
+ dsb + 3,
+ "name4",
+ dsb + 3,
+ ),
+ ],
+ )
class SequenceAsServerDefaultTest(
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
