--- /dev/null
+.. change::
+ :tags: bug, engine
+ :tickets: 5929
+
+ Fixed bug where the "schema_translate_map" feature failed to be taken into
+ account for the use case of direct execution of
+ :class:`_schema.DefaultGenerator` objects such as sequences, which included
+ the case where they were "pre-executed" in order to generate primary key
+ values when implicit_returning was disabled.
return self._execute_scalar(
"SELECT gen_id(%s, 1) FROM rdb$database"
- % self.dialect.identifier_preparer.format_sequence(seq),
+ % self.identifier_preparer.format_sequence(seq),
type_,
)
In addition to the above DBAPI layers with native SQLAlchemy support, there
are third-party dialects for other DBAPI layers that are compatible
with SQL Server. See the "External Dialects" list on the
-:ref:`dialect_toplevel` page.
+:ref:`dialect_toplevel` page.
.. _mssql_identity:
self.cursor,
self._opt_encode(
"SET IDENTITY_INSERT %s ON"
- % self.dialect.identifier_preparer.format_table(tbl)
+ % self.identifier_preparer.format_table(tbl)
),
(),
self,
self.cursor,
self._opt_encode(
"SET IDENTITY_INSERT %s OFF"
- % self.dialect.identifier_preparer.format_table(
+ % self.identifier_preparer.format_table(
self.compiled.statement.table
)
),
self.cursor.execute(
self._opt_encode(
"SET IDENTITY_INSERT %s OFF"
- % self.dialect.identifier_preparer.format_table(
+ % self.identifier_preparer.format_table(
self.compiled.statement.table
)
)
return self._execute_scalar(
(
"SELECT NEXT VALUE FOR %s"
- % self.dialect.identifier_preparer.format_sequence(seq)
+ % self.identifier_preparer.format_sequence(seq)
),
type_,
)
from ...types import BOOLEAN
from ...types import DATE
from ...types import VARBINARY
+from ...util import compat
from ...util import topological
+if compat.TYPE_CHECKING:
+ from typing import Any
+
RESERVED_WORDS = set(
[
"accessible",
return self._execute_scalar(
(
"select nextval(%s)"
- % self.dialect.identifier_preparer.format_sequence(seq)
+ % self.identifier_preparer.format_sequence(seq)
),
type_,
)
return parser.parse(sql, charset)
def _detect_charset(self, connection):
+ # type: (Any) -> str
raise NotImplementedError()
def _detect_casing(self, connection):
return self.process(vc.column, **kw) + "(+)"
def visit_sequence(self, seq, **kw):
- return (
- self.dialect.identifier_preparer.format_sequence(seq) + ".nextval"
- )
+ return self.preparer.format_sequence(seq) + ".nextval"
def get_render_as_alias_suffix(self, alias_name_text):
"""Oracle doesn't like ``FROM table AS alias``"""
def fire_sequence(self, seq, type_):
return self._execute_scalar(
"SELECT "
- + self.dialect.identifier_preparer.format_sequence(seq)
+ + self.identifier_preparer.format_sequence(seq)
+ ".nextval FROM DUAL",
type_,
)
return self._execute_scalar(
(
"select nextval('%s')"
- % self.dialect.identifier_preparer.format_sequence(seq)
+ % self.identifier_preparer.format_sequence(seq)
),
type_,
)
else:
return "unknown"
+ @util.memoized_property
+ def identifier_preparer(self):
+ if self.compiled:
+ return self.compiled.preparer
+ elif "schema_translate_map" in self.execution_options:
+ return self.dialect.identifier_preparer._with_schema_translate(
+ self.execution_options["schema_translate_map"]
+ )
+ else:
+ return self.dialect.identifier_preparer
+
@util.memoized_property
def engine(self):
return self.root_connection.engine
):
stmt = self.dialect._encoder(stmt)[0]
+ if "schema_translate_map" in self.execution_options:
+ schema_translate_map = self.execution_options.get(
+ "schema_translate_map", {}
+ )
+
+ rst = self.identifier_preparer._render_schema_translates
+ stmt = rst(stmt, schema_translate_map)
+
if not parameters:
if self.dialect.positional:
parameters = self.dialect.execute_sequence_format()
)
util.drop_all_tables(eng, inspector)
-
if config.requirements.schemas.enabled_for_config(cfg):
util.drop_all_tables(eng, inspector, schema=cfg.test_schema)
util.drop_all_tables(eng, inspector, schema=cfg.test_schema_2)
with eng.begin() as conn:
for seq in inspector.get_sequence_names():
conn.execute(ddl.DropSequence(schema.Sequence(seq)))
+ if config.requirements.schemas.enabled_for_config(cfg):
+ for schema_name in [cfg.test_schema, cfg.test_schema_2]:
+ for seq in inspector.get_sequence_names(
+ schema=schema_name
+ ):
+ conn.execute(
+ ddl.DropSequence(
+ schema.Sequence(seq, schema=schema_name)
+ )
+ )
@register.init
Column("data", String(50)),
)
+ Table(
+ "seq_no_returning",
+ metadata,
+ Column(
+ "id",
+ Integer,
+ Sequence("noret_id_seq"),
+ primary_key=True,
+ ),
+ Column("data", String(50)),
+ implicit_returning=False,
+ )
+
+ if testing.requires.schemas.enabled:
+ Table(
+ "seq_no_returning_sch",
+ metadata,
+ Column(
+ "id",
+ Integer,
+ Sequence("noret_sch_id_seq", schema=config.test_schema),
+ primary_key=True,
+ ),
+ Column("data", String(50)),
+ implicit_returning=False,
+ schema=config.test_schema,
+ )
+
def test_insert_roundtrip(self, connection):
connection.execute(self.tables.seq_pk.insert(), dict(data="some data"))
self._assert_round_trip(self.tables.seq_pk, connection)
row = conn.execute(table.select()).first()
eq_(row, (testing.db.dialect.default_sequence_base, "some data"))
+ def test_insert_roundtrip_no_implicit_returning(self, connection):
+ connection.execute(
+ self.tables.seq_no_returning.insert(), dict(data="some data")
+ )
+ self._assert_round_trip(self.tables.seq_no_returning, connection)
+
+ @testing.combinations((True,), (False,), argnames="implicit_returning")
+ @testing.requires.schemas
+ def test_insert_roundtrip_translate(self, connection, implicit_returning):
+
+ seq_no_returning = Table(
+ "seq_no_returning_sch",
+ MetaData(),
+ Column(
+ "id",
+ Integer,
+ Sequence("noret_sch_id_seq", schema="alt_schema"),
+ primary_key=True,
+ ),
+ Column("data", String(50)),
+ implicit_returning=implicit_returning,
+ schema="alt_schema",
+ )
+
+ connection = connection.execution_options(
+ schema_translate_map={"alt_schema": config.test_schema}
+ )
+ connection.execute(seq_no_returning.insert(), dict(data="some data"))
+ self._assert_round_trip(seq_no_returning, connection)
+
+ @testing.requires.schemas
+ def test_nextval_direct_schema_translate(self, connection):
+ seq = Sequence("noret_sch_id_seq", schema="alt_schema")
+ connection = connection.execution_options(
+ schema_translate_map={"alt_schema": config.test_schema}
+ )
+
+ r = connection.execute(seq)
+ eq_(r, testing.db.dialect.default_sequence_base)
+
class SequenceCompilerTest(testing.AssertsCompiledSQL, fixtures.TestBase):
__requires__ = ("sequences",)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
