--- /dev/null
+.. change::
+ :tags: bug, engine
+ :tickets: 10025
+
+ Adjusted the :paramref:`_sa.create_engine.schema_translate_map` feature
+ such that **all** schema names in the statement are now tokenized,
+ regardless of whether or not a specific name is in the immediate schema
+ translate map given, and to fallback to substituting the original name when
+ the key is not in the actual schema translate map at execution time. These
+ two changes allow for repeated use of a compiled object with schema
+ schema_translate_maps that include or dont include various keys on each
+ run, allowing cached SQL constructs to continue to function at runtime when
+ schema translate maps with different sets of keys are used each time. In
+ addition, added detection of schema_translate_map dictionaries which gain
+ or lose a ``None`` key across calls for the same statement, which affects
+ compilation of the statement and is not compatible with caching; an
+ exception is raised for these scenarios.
"""
+ _includes_none_schema_translate: bool = False
+
def __init__(
self,
dialect,
prep = self.__class__.__new__(self.__class__)
prep.__dict__.update(self.__dict__)
+ includes_none = None in schema_translate_map
+
def symbol_getter(obj):
name = obj.schema
- if name in schema_translate_map and obj._use_schema_map:
+ if obj._use_schema_map and (name is not None or includes_none):
if name is not None and ("[" in name or "]" in name):
raise exc.CompileError(
"Square bracket characters ([]) not supported "
return obj.schema
prep.schema_for_object = symbol_getter
+ prep._includes_none_schema_translate = includes_none
return prep
def _render_schema_translates(self, statement, schema_translate_map):
d = schema_translate_map
if None in d:
+ if not self._includes_none_schema_translate:
+ raise exc.InvalidRequestError(
+ "schema translate map which previously did not have "
+ "`None` present as a key now has `None` present; compiled "
+ "statement may lack adequate placeholders. Please use "
+ "consistent keys in successive "
+ "schema_translate_map dictionaries."
+ )
+
d["_none"] = d[None]
def replace(m):
name = m.group(2)
- effective_schema = d[name]
+ if name in d:
+ effective_schema = d[name]
+ else:
+ if name in (None, "_none"):
+ raise exc.InvalidRequestError(
+ "schema translate map which previously had `None` "
+ "present as a key now no longer has it present; don't "
+ "know how to apply schema for compiled statement. "
+ "Please use consistent keys in successive "
+ "schema_translate_map dictionaries."
+ )
+ effective_schema = name
+
if not effective_schema:
effective_schema = self.dialect.default_schema_name
if not effective_schema:
return t1, t2, t3
+ @testing.fixture
+ def same_named_tables(self, metadata, connection):
+ ts1 = Table(
+ "t1", metadata, Column("x", String(10)), schema=config.test_schema
+ )
+ tsnone = Table("t1", metadata, Column("x", String(10)), schema=None)
+
+ metadata.create_all(connection)
+
+ connection.execute(ts1.insert().values(x="ts1"))
+ connection.execute(tsnone.insert().values(x="tsnone"))
+ return ts1, tsnone
+
def test_create_table(self, plain_tables, connection):
map_ = {
None: config.test_schema,
CompiledSQL("DELETE FROM __[SCHEMA_bar].t3"),
)
+ def test_schema_translate_map_keys_change_name_added(
+ self, same_named_tables, connection
+ ):
+ """test #10024"""
+
+ metadata = MetaData()
+
+ translate_table = Table(
+ "t1", metadata, Column("x", String(10)), schema=config.test_schema
+ )
+
+ eq_(
+ connection.scalar(
+ select(translate_table),
+ execution_options={"schema_translate_map": {"foo": "bar"}},
+ ),
+ "ts1",
+ )
+
+ eq_(
+ connection.scalar(
+ select(translate_table),
+ execution_options={
+ "schema_translate_map": {
+ "foo": "bar",
+ config.test_schema: None,
+ }
+ },
+ ),
+ "tsnone",
+ )
+
+ def test_schema_translate_map_keys_change_name_removed(
+ self, same_named_tables, connection
+ ):
+ """test #10024"""
+
+ metadata = MetaData()
+
+ translate_table = Table(
+ "t1", metadata, Column("x", String(10)), schema=config.test_schema
+ )
+
+ eq_(
+ connection.scalar(
+ select(translate_table),
+ execution_options={
+ "schema_translate_map": {
+ "foo": "bar",
+ config.test_schema: None,
+ }
+ },
+ ),
+ "tsnone",
+ )
+
+ eq_(
+ connection.scalar(
+ select(translate_table),
+ execution_options={"schema_translate_map": {"foo": "bar"}},
+ ),
+ "ts1",
+ )
+
+ def test_schema_translate_map_keys_change_none_removed(
+ self, same_named_tables, connection
+ ):
+ """test #10024"""
+
+ connection.engine.clear_compiled_cache()
+
+ metadata = MetaData()
+
+ translate_table = Table("t1", metadata, Column("x", String(10)))
+
+ eq_(
+ connection.scalar(
+ select(translate_table),
+ execution_options={
+ "schema_translate_map": {None: config.test_schema}
+ },
+ ),
+ "ts1",
+ )
+
+ with expect_raises_message(
+ tsa.exc.StatementError,
+ "schema translate map which previously had `None` "
+ "present as a key now no longer has it present",
+ ):
+ connection.scalar(
+ select(translate_table),
+ execution_options={"schema_translate_map": {"foo": "bar"}},
+ ),
+
+ def test_schema_translate_map_keys_change_none_added(
+ self, same_named_tables, connection
+ ):
+ """test #10024"""
+
+ connection.engine.clear_compiled_cache()
+
+ metadata = MetaData()
+
+ translate_table = Table("t1", metadata, Column("x", String(10)))
+
+ eq_(
+ connection.scalar(
+ select(translate_table),
+ execution_options={"schema_translate_map": {"foo": "bar"}},
+ ),
+ "tsnone",
+ )
+
+ with expect_raises_message(
+ tsa.exc.StatementError,
+ "schema translate map which previously did not have `None` "
+ "present as a key now has `None` present; compiled statement may "
+ "lack adequate placeholders.",
+ ):
+ connection.scalar(
+ select(translate_table),
+ execution_options={
+ "schema_translate_map": {None: config.test_schema}
+ },
+ ),
+
def test_crud(self, plain_tables, connection):
# provided by metadata fixture provided by plain_tables fixture
self.metadata.create_all(connection)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
