--- /dev/null
+.. change::
+ :tags: mssql, engine
+ :tickets: 4809
+
+ Deprecated the ``legacy_schema_aliasing`` parameter to
+ :meth:`_sa.create_engine`. This is a long-outdated parameter that has
+ defaulted to False since version 1.1.
in version 1.0.5 to allow disabling of legacy mode for schemas now
defaults to False.
+.. deprecated:: 1.4
+
+ The ``legacy_schema_aliasing`` flag is now
+ deprecated and will be removed in a future release.
.. _mssql_indexes:
_supports_offset_fetch = False
_supports_nvarchar_max = False
+ legacy_schema_aliasing = False
+
server_version_info = ()
statement_compiler = MSSQLCompiler
schema_name="dbo",
isolation_level=None,
deprecate_large_types=None,
- legacy_schema_aliasing=False,
json_serializer=None,
json_deserializer=None,
+ legacy_schema_aliasing=None,
**opts
):
self.query_timeout = int(query_timeout or 0)
self.use_scope_identity = use_scope_identity
self.deprecate_large_types = deprecate_large_types
- self.legacy_schema_aliasing = legacy_schema_aliasing
+
+ if legacy_schema_aliasing is not None:
+ util.warn_deprecated(
+ "The legacy_schema_aliasing parameter is "
+ "deprecated and will be removed in a future release.",
+ "1.4",
+ )
+ self.legacy_schema_aliasing = legacy_schema_aliasing
super(MSDialect, self).__init__(**opts)
from ...testing.provision import drop_db
from ...testing.provision import log
from ...testing.provision import run_reap_dbs
-from ...testing.provision import update_db_opts
-
-
-@update_db_opts.for_db("mssql")
-def _mssql_update_db_opts(db_url, db_opts):
- db_opts["legacy_schema_aliasing"] = False
@create_db.for_db("mssql")
--- /dev/null
+# -*- encoding: utf-8
+from sqlalchemy import Column
+from sqlalchemy import engine_from_config
+from sqlalchemy import Integer
+from sqlalchemy import MetaData
+from sqlalchemy import select
+from sqlalchemy import String
+from sqlalchemy import Table
+from sqlalchemy import testing
+from sqlalchemy.dialects.mssql import base as mssql
+from sqlalchemy.sql import column
+from sqlalchemy.sql import table
+from sqlalchemy.testing import assertions
+from sqlalchemy.testing import AssertsCompiledSQL
+from sqlalchemy.testing import engines
+from sqlalchemy.testing import eq_
+from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import is_
+from sqlalchemy.testing.mock import Mock
+
+
+def _legacy_schema_aliasing_warning():
+ return assertions.expect_deprecated("The legacy_schema_aliasing parameter")
+
+
+class LegacySchemaAliasingTest(fixtures.TestBase, AssertsCompiledSQL):
+ """Legacy behavior tried to prevent schema-qualified tables
+ from being rendered as dotted names, and were instead aliased.
+
+ This behavior no longer seems to be required.
+
+ """
+
+ def setup(self):
+ metadata = MetaData()
+ self.t1 = table(
+ "t1",
+ column("a", Integer),
+ column("b", String),
+ column("c", String),
+ )
+ self.t2 = Table(
+ "t2",
+ metadata,
+ Column("a", Integer),
+ Column("b", Integer),
+ Column("c", Integer),
+ schema="schema",
+ )
+
+ def _assert_sql(self, element, legacy_sql, modern_sql=None):
+ dialect = self._legacy_dialect()
+
+ self.assert_compile(element, legacy_sql, dialect=dialect)
+
+ dialect = mssql.dialect()
+ self.assert_compile(element, modern_sql or "foob", dialect=dialect)
+
+ def _legacy_dialect(self):
+ with _legacy_schema_aliasing_warning():
+ return mssql.dialect(legacy_schema_aliasing=True)
+
+ @testing.combinations(
+ (
+ {
+ "sqlalchemy.url": "mssql://foodsn",
+ "sqlalchemy.legacy_schema_aliasing": "true",
+ },
+ True,
+ ),
+ (
+ {
+ "sqlalchemy.url": "mssql://foodsn",
+ "sqlalchemy.legacy_schema_aliasing": "false",
+ },
+ False,
+ ),
+ )
+ def test_legacy_schema_flag(self, cfg, expected):
+ with testing.expect_deprecated("The legacy_schema_aliasing parameter"):
+ e = engine_from_config(
+ cfg, module=Mock(version="MS SQL Server 11.0.92")
+ )
+ is_(e.dialect.legacy_schema_aliasing, expected)
+
+ def test_result_map(self):
+ s = self.t2.select()
+ c = s.compile(dialect=self._legacy_dialect())
+ assert self.t2.c.a in set(c._create_result_map()["a"][1])
+
+ def test_result_map_use_labels(self):
+ s = self.t2.select(use_labels=True)
+ c = s.compile(dialect=self._legacy_dialect())
+ assert self.t2.c.a in set(c._create_result_map()["schema_t2_a"][1])
+
+ def test_straight_select(self):
+ self._assert_sql(
+ self.t2.select(),
+ "SELECT t2_1.a, t2_1.b, t2_1.c FROM [schema].t2 AS t2_1",
+ "SELECT [schema].t2.a, [schema].t2.b, "
+ "[schema].t2.c FROM [schema].t2",
+ )
+
+ def test_straight_select_use_labels(self):
+ self._assert_sql(
+ self.t2.select(use_labels=True),
+ "SELECT t2_1.a AS schema_t2_a, t2_1.b AS schema_t2_b, "
+ "t2_1.c AS schema_t2_c FROM [schema].t2 AS t2_1",
+ "SELECT [schema].t2.a AS schema_t2_a, "
+ "[schema].t2.b AS schema_t2_b, "
+ "[schema].t2.c AS schema_t2_c FROM [schema].t2",
+ )
+
+ def test_join_to_schema(self):
+ t1, t2 = self.t1, self.t2
+ self._assert_sql(
+ t1.join(t2, t1.c.a == t2.c.a).select(),
+ "SELECT t1.a, t1.b, t1.c, t2_1.a, t2_1.b, t2_1.c FROM t1 "
+ "JOIN [schema].t2 AS t2_1 ON t2_1.a = t1.a",
+ "SELECT t1.a, t1.b, t1.c, [schema].t2.a, [schema].t2.b, "
+ "[schema].t2.c FROM t1 JOIN [schema].t2 ON [schema].t2.a = t1.a",
+ )
+
+ def test_union_schema_to_non(self):
+ t1, t2 = self.t1, self.t2
+ s = (
+ select(t2.c.a, t2.c.b)
+ .apply_labels()
+ .union(select(t1.c.a, t1.c.b).apply_labels())
+ .alias()
+ .select()
+ )
+ self._assert_sql(
+ s,
+ "SELECT anon_1.schema_t2_a, anon_1.schema_t2_b FROM "
+ "(SELECT t2_1.a AS schema_t2_a, t2_1.b AS schema_t2_b "
+ "FROM [schema].t2 AS t2_1 UNION SELECT t1.a AS t1_a, "
+ "t1.b AS t1_b FROM t1) AS anon_1",
+ "SELECT anon_1.schema_t2_a, anon_1.schema_t2_b FROM "
+ "(SELECT [schema].t2.a AS schema_t2_a, [schema].t2.b AS "
+ "schema_t2_b FROM [schema].t2 UNION SELECT t1.a AS t1_a, "
+ "t1.b AS t1_b FROM t1) AS anon_1",
+ )
+
+ def test_column_subquery_to_alias(self):
+ a1 = self.t2.alias("a1")
+ s = select([self.t2, select(a1.c.a).scalar_subquery()])
+ self._assert_sql(
+ s,
+ "SELECT t2_1.a, t2_1.b, t2_1.c, "
+ "(SELECT a1.a FROM [schema].t2 AS a1) "
+ "AS anon_1 FROM [schema].t2 AS t2_1",
+ "SELECT [schema].t2.a, [schema].t2.b, [schema].t2.c, "
+ "(SELECT a1.a FROM [schema].t2 AS a1) AS anon_1 FROM [schema].t2",
+ )
+
+
+class LegacySchemaAliasingBackendTest(
+ testing.AssertsExecutionResults, fixtures.TestBase
+):
+ __only_on__ = "mssql"
+
+ @testing.provide_metadata
+ def test_insertid_schema(self):
+ meta = self.metadata
+
+ with _legacy_schema_aliasing_warning():
+ eng = engines.testing_engine(
+ options=dict(legacy_schema_aliasing=False)
+ )
+
+ tbl = Table(
+ "test",
+ meta,
+ Column("id", Integer, primary_key=True),
+ schema=testing.config.test_schema,
+ )
+
+ with eng.connect() as conn:
+ tbl.create(conn)
+ conn.execute(tbl.insert(), {"id": 1})
+ eq_(conn.scalar(tbl.select()), 1)
+
+ @testing.provide_metadata
+ def test_insertid_schema_legacy(self):
+ meta = self.metadata
+
+ tbl = Table(
+ "test",
+ meta,
+ Column("id", Integer, primary_key=True),
+ schema=testing.config.test_schema,
+ )
+
+ with _legacy_schema_aliasing_warning():
+ eng = engines.testing_engine(
+ options=dict(legacy_schema_aliasing=True)
+ )
+
+ with eng.connect() as conn:
+
+ tbl.create(conn)
+ conn.execute(tbl.insert(), {"id": 1})
+ eq_(conn.scalar(tbl.select()), 1)
+
+ @testing.provide_metadata
+ def test_delete_schema(self, connection):
+ meta = self.metadata
+
+ is_(connection.dialect.legacy_schema_aliasing, False)
+
+ tbl = Table(
+ "test",
+ meta,
+ Column("id", Integer, primary_key=True),
+ schema=testing.config.test_schema,
+ )
+ tbl.create(connection)
+ connection.execute(tbl.insert(), {"id": 1})
+ eq_(connection.scalar(tbl.select()), 1)
+ connection.execute(tbl.delete(tbl.c.id == 1))
+ eq_(connection.scalar(tbl.select()), None)
+
+ @testing.provide_metadata
+ def test_delete_schema_legacy(self):
+ meta = self.metadata
+ with _legacy_schema_aliasing_warning():
+ eng = engines.testing_engine(
+ options=dict(legacy_schema_aliasing=True)
+ )
+
+ tbl = Table(
+ "test",
+ meta,
+ Column("id", Integer, primary_key=True),
+ schema=testing.config.test_schema,
+ )
+
+ with eng.connect() as conn:
+ tbl.create(conn)
+ conn.execute(tbl.insert(), {"id": 1})
+ eq_(conn.scalar(tbl.select()), 1)
+ conn.execute(tbl.delete(tbl.c.id == 1))
+ eq_(conn.scalar(tbl.select()), None)
# -*- encoding: utf-8
from sqlalchemy import Column
-from sqlalchemy import engine_from_config
from sqlalchemy import event
from sqlalchemy import exc
from sqlalchemy import Integer
)
-class EngineFromConfigTest(fixtures.TestBase):
- def test_legacy_schema_flag(self):
- cfg = {
- "sqlalchemy.url": "mssql://foodsn",
- "sqlalchemy.legacy_schema_aliasing": "false",
- }
- e = engine_from_config(
- cfg, module=Mock(version="MS SQL Server 11.0.92")
- )
- eq_(e.dialect.legacy_schema_aliasing, False)
-
-
class FastExecutemanyTest(fixtures.TestBase):
__only_on__ = "mssql"
__backend__ = True
from sqlalchemy import testing
from sqlalchemy import util
from sqlalchemy.dialects.mssql import base as mssql
-from sqlalchemy.sql import column
-from sqlalchemy.sql import table
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing.assertsql import DialectSQL
from sqlalchemy.util import ue
-
metadata = None
cattable = None
matchtable = None
-class LegacySchemaAliasingTest(fixtures.TestBase, AssertsCompiledSQL):
- """Legacy behavior tried to prevent schema-qualified tables
- from being rendered as dotted names, and were instead aliased.
-
- This behavior no longer seems to be required.
-
- """
-
- def setup(self):
- metadata = MetaData()
- self.t1 = table(
- "t1",
- column("a", Integer),
- column("b", String),
- column("c", String),
- )
- self.t2 = Table(
- "t2",
- metadata,
- Column("a", Integer),
- Column("b", Integer),
- Column("c", Integer),
- schema="schema",
- )
-
- def _assert_sql(self, element, legacy_sql, modern_sql=None):
- dialect = mssql.dialect(legacy_schema_aliasing=True)
-
- self.assert_compile(element, legacy_sql, dialect=dialect)
-
- dialect = mssql.dialect()
- self.assert_compile(element, modern_sql or "foob", dialect=dialect)
-
- def _legacy_dialect(self):
- return mssql.dialect(legacy_schema_aliasing=True)
-
- def test_result_map(self):
- s = self.t2.select()
- c = s.compile(dialect=self._legacy_dialect())
- assert self.t2.c.a in set(c._create_result_map()["a"][1])
-
- def test_result_map_use_labels(self):
- s = self.t2.select(use_labels=True)
- c = s.compile(dialect=self._legacy_dialect())
- assert self.t2.c.a in set(c._create_result_map()["schema_t2_a"][1])
-
- def test_straight_select(self):
- self._assert_sql(
- self.t2.select(),
- "SELECT t2_1.a, t2_1.b, t2_1.c FROM [schema].t2 AS t2_1",
- "SELECT [schema].t2.a, [schema].t2.b, "
- "[schema].t2.c FROM [schema].t2",
- )
-
- def test_straight_select_use_labels(self):
- self._assert_sql(
- self.t2.select(use_labels=True),
- "SELECT t2_1.a AS schema_t2_a, t2_1.b AS schema_t2_b, "
- "t2_1.c AS schema_t2_c FROM [schema].t2 AS t2_1",
- "SELECT [schema].t2.a AS schema_t2_a, "
- "[schema].t2.b AS schema_t2_b, "
- "[schema].t2.c AS schema_t2_c FROM [schema].t2",
- )
-
- def test_join_to_schema(self):
- t1, t2 = self.t1, self.t2
- self._assert_sql(
- t1.join(t2, t1.c.a == t2.c.a).select(),
- "SELECT t1.a, t1.b, t1.c, t2_1.a, t2_1.b, t2_1.c FROM t1 "
- "JOIN [schema].t2 AS t2_1 ON t2_1.a = t1.a",
- "SELECT t1.a, t1.b, t1.c, [schema].t2.a, [schema].t2.b, "
- "[schema].t2.c FROM t1 JOIN [schema].t2 ON [schema].t2.a = t1.a",
- )
-
- def test_union_schema_to_non(self):
- t1, t2 = self.t1, self.t2
- s = (
- select(t2.c.a, t2.c.b)
- .apply_labels()
- .union(select(t1.c.a, t1.c.b).apply_labels())
- .alias()
- .select()
- )
- self._assert_sql(
- s,
- "SELECT anon_1.schema_t2_a, anon_1.schema_t2_b FROM "
- "(SELECT t2_1.a AS schema_t2_a, t2_1.b AS schema_t2_b "
- "FROM [schema].t2 AS t2_1 UNION SELECT t1.a AS t1_a, "
- "t1.b AS t1_b FROM t1) AS anon_1",
- "SELECT anon_1.schema_t2_a, anon_1.schema_t2_b FROM "
- "(SELECT [schema].t2.a AS schema_t2_a, [schema].t2.b AS "
- "schema_t2_b FROM [schema].t2 UNION SELECT t1.a AS t1_a, "
- "t1.b AS t1_b FROM t1) AS anon_1",
- )
-
- def test_column_subquery_to_alias(self):
- a1 = self.t2.alias("a1")
- s = select([self.t2, select(a1.c.a).scalar_subquery()])
- self._assert_sql(
- s,
- "SELECT t2_1.a, t2_1.b, t2_1.c, "
- "(SELECT a1.a FROM [schema].t2 AS a1) "
- "AS anon_1 FROM [schema].t2 AS t2_1",
- "SELECT [schema].t2.a, [schema].t2.b, [schema].t2.c, "
- "(SELECT a1.a FROM [schema].t2 AS a1) AS anon_1 FROM [schema].t2",
- )
-
-
class IdentityInsertTest(fixtures.TestBase, AssertsCompiledSQL):
__only_on__ = "mssql"
__dialect__ = mssql.MSDialect()
)
@testing.provide_metadata
- def test_insertid_schema(self):
- meta = self.metadata
- eng = engines.testing_engine(
- options=dict(legacy_schema_aliasing=False)
- )
- meta.bind = eng
- conn = eng.connect()
- conn.exec_driver_sql("create schema paj")
-
- @event.listens_for(meta, "after_drop")
- def cleanup(target, connection, **kw):
- connection.exec_driver_sql("drop schema paj")
-
- tbl = Table(
- "test", meta, Column("id", Integer, primary_key=True), schema="paj"
- )
- tbl.create(conn)
- conn.execute(tbl.insert(), {"id": 1})
- eq_(conn.scalar(tbl.select()), 1)
-
- @testing.provide_metadata
- def test_insertid_schema_legacy(self):
+ def test_insertid_schema(self, connection):
meta = self.metadata
- eng = engines.testing_engine(options=dict(legacy_schema_aliasing=True))
- meta.bind = eng
- conn = eng.connect()
- conn.exec_driver_sql("create schema paj")
-
- @event.listens_for(meta, "after_drop")
- def cleanup(target, connection, **kw):
- connection.exec_driver_sql("drop schema paj")
tbl = Table(
- "test", meta, Column("id", Integer, primary_key=True), schema="paj"
+ "test",
+ meta,
+ Column("id", Integer, primary_key=True),
+ schema=testing.config.test_schema,
)
- tbl.create()
- tbl.insert().execute({"id": 1})
- eq_(tbl.select().scalar(), 1)
+ tbl.create(connection)
+ connection.execute(tbl.insert(), {"id": 1})
+ eq_(connection.scalar(tbl.select()), 1)
@testing.provide_metadata
def test_returning_no_autoinc(self, connection):
eq_(result.fetchall(), [(1, "somestring")])
@testing.provide_metadata
- def test_delete_schema(self):
+ def test_delete_schema(self, connection):
meta = self.metadata
- eng = engines.testing_engine(
- options=dict(legacy_schema_aliasing=False)
- )
- meta.bind = eng
- conn = eng.connect()
- conn.exec_driver_sql("create schema paj")
-
- @event.listens_for(meta, "after_drop")
- def cleanup(target, connection, **kw):
- connection.exec_driver_sql("drop schema paj")
tbl = Table(
- "test", meta, Column("id", Integer, primary_key=True), schema="paj"
- )
- tbl.create(conn)
- conn.execute(tbl.insert(), {"id": 1})
- eq_(conn.scalar(tbl.select()), 1)
- conn.execute(tbl.delete(tbl.c.id == 1))
- eq_(conn.scalar(tbl.select()), None)
-
- @testing.provide_metadata
- def test_delete_schema_legacy(self):
- meta = self.metadata
- eng = engines.testing_engine(options=dict(legacy_schema_aliasing=True))
- meta.bind = eng
- conn = eng.connect()
- conn.exec_driver_sql("create schema paj")
-
- @event.listens_for(meta, "after_drop")
- def cleanup(target, connection, **kw):
- connection.exec_driver_sql("drop schema paj")
-
- tbl = Table(
- "test", meta, Column("id", Integer, primary_key=True), schema="paj"
+ "test",
+ meta,
+ Column("id", Integer, primary_key=True),
+ schema=testing.config.test_schema,
)
- tbl.create(conn)
- conn.execute(tbl.insert(), {"id": 1})
- eq_(conn.scalar(tbl.select()), 1)
- conn.execute(tbl.delete(tbl.c.id == 1))
- eq_(conn.scalar(tbl.select()), None)
+ tbl.create(connection)
+ connection.execute(tbl.insert(), {"id": 1})
+ eq_(connection.scalar(tbl.select()), 1)
+ connection.execute(tbl.delete(tbl.c.id == 1))
+ eq_(connection.scalar(tbl.select()), None)
@testing.provide_metadata
def test_insertid_reserved(self, connection):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
