check constraints in 0.9 as part of :ticket:`2742`, which more commonly
feature Core SQL expressions as opposed to plain string expressions.
+ .. change:: 2626
+ :tags: bug, mssql
+ :tickets: 2626
+
+ The SQL Server dialect now allows for a database and/or owner name
+ with a dot inside of it, using brackets explicitly in the string around
+ the owner and optionally the database name as well. In addition,
+ sending the :class:`.quoted_name` construct for the schema name will
+ not split on the dot and will deliver the full string as the "owner".
+ :class:`.quoted_name` is also now available from the ``sqlalchemy.sql``
+ import space.
+
+ .. seealso::
+
+ :ref:`change_2626`
+
.. change:: 3923
:tags: bug, sql
:tickets: 3923
:ticket:`3276`
+
+Dialect Improvements and Changes - SQL Server
+=============================================
+
+.. _change_2626:
+
+SQL Server schema names with embedded dots supported
+-----------------------------------------------------
+
+The SQL Server dialect has a behavior such that a schema name with a dot inside
+of it is assumed to be a "database"."owner" identifier pair, which is
+necessarily split up into these separate components during table and component
+reflection operations, as well as when rendering quoting for the schema name so
+that the two symbols are quoted separately. The schema argument can
+now be passed using brackets to manually specify where this split
+occurs, allowing database and/or owner names that themselves contain one
+or more dots::
+
+ Table(
+ "some_table", metadata,
+ Column("q", String(50)),
+ schema="[MyDataBase.dbo]"
+ )
+
+The above table will consider the "owner" to be ``MyDataBase.dbo``, which
+will also be quoted upon render, and the "database" as None. To individually
+refer to database name and owner, use two pairs of brackets::
+
+ Table(
+ "some_table", metadata,
+ Column("q", String(50)),
+ schema="[MyDataBase.SomeDB].[MyDB.owner]"
+ )
+
+Additionally, the :class:`.quoted_name` construct is now honored when
+passed to "schema" by the SQL Server dialect; the given symbol will
+not be split on the dot if the quote flag is True and will be interpreted
+as the "owner".
+
+.. seealso::
+
+ :ref:`multipart_schema_names`
+
+:ticket:`2626`
.. versionadded:: 1.0.0
+.. _multipart_schema_names:
+
+Multipart Schema Names
+----------------------
+
+SQL Server schemas sometimes require multiple parts to their "schema"
+qualifier, that is, including the database name and owner name as separate
+tokens, such as ``mydatabase.dbo.some_table``. These multipart names can be set
+at once using the :paramref:`.Table.schema` argument of :class:`.Table`::
+
+ Table(
+ "some_table", metadata,
+ Column("q", String(50)),
+ schema="mydatabase.dbo"
+ )
+
+When performing operations such as table or component reflection, a schema
+argument that contains a dot will be split into separate
+"database" and "owner" components in order to correctly query the SQL
+Server information schema tables, as these two values are stored separately.
+Additionally, when rendering the schema name for DDL or SQL, the two
+components will be quoted separately for case sensitive names and other
+special characters. Given an argument as below::
+
+ Table(
+ "some_table", metadata,
+ Column("q", String(50)),
+ schema="MyDataBase.dbo"
+ )
+
+The above schema would be rendered as ``[MyDataBase].dbo``, and also in
+reflection, would be reflected using "dbo" as the owner and "MyDataBase"
+as the database name.
+
+To control how the schema name is broken into database / owner,
+specify brackets (which in SQL Server are quoting characters) in the name.
+Below, the "owner" will be considered as ``MyDataBase.dbo`` and the
+"database" will be None::
+
+ Table(
+ "some_table", metadata,
+ Column("q", String(50)),
+ schema="[MyDataBase.dbo]"
+ )
+
+To individually specify both database and owner name with special characters
+or embedded dots, use two sets of brackets::
+
+ Table(
+ "some_table", metadata,
+ Column("q", String(50)),
+ schema="[MyDataBase.Period].[MyOwner.Dot]"
+ )
+
+
+.. versionchanged:: 1.2 the SQL Server dialect now treats brackets as
+ identifier delimeters splitting the schema into separate database
+ and owner tokens, to allow dots within either name itself.
+
.. _legacy_schema_rendering:
Legacy Schema Mode
import re
from ... import sql, schema as sa_schema, exc, util
-from ...sql import compiler, expression, util as sql_util
+from ...sql import compiler, expression, util as sql_util, quoted_name
from ... import engine
from ...engine import reflection, default
from ... import types as sqltypes
def _escape_identifier(self, value):
return value
+
def quote_schema(self, schema, force=None):
"""Prepare a quoted table and schema name."""
- result = '.'.join([self.quote(x, force) for x in schema.split('.')])
+
+ dbname, owner = _schema_elements(schema)
+ if dbname:
+ result = "%s.%s" % (
+ self.quote(dbname, force), self.quote(owner, force))
+ elif owner:
+ result = self.quote(owner, force)
+ else:
+ result = ""
return result
if not schema:
return None, dialect.default_schema_name
elif "." in schema:
- return schema.split(".", 1)
+ return _schema_elements(schema)
else:
return None, schema
+def _schema_elements(schema):
+ if isinstance(schema, quoted_name) and schema.quote:
+ return None, schema
+
+ push = []
+ symbol = ""
+ bracket = False
+ for token in re.split(r"(\[|\]|\.)", schema):
+ if not token:
+ continue
+ if token == '[':
+ bracket = True
+ elif token == ']':
+ bracket = False
+ elif not bracket and token == ".":
+ push.append(symbol)
+ symbol = ""
+ else:
+ symbol += token
+ if symbol:
+ push.append(symbol)
+ if len(push) > 1:
+ return push[0], "".join(push[1:])
+ elif len(push):
+ return None, push[0]
+ else:
+ return None, None
+
+
class MSDialect(default.DefaultDialect):
name = 'mssql'
supports_default_values = True
outerjoin,
outparam,
over,
+ quoted_name,
select,
subquery,
table,
can be quoted. Such as to use the :meth:`.Engine.has_table` method with
an unconditionally quoted name::
- from sqlaclchemy import create_engine
- from sqlalchemy.sql.elements import quoted_name
+ from sqlalchemy import create_engine
+ from sqlalchemy.sql import quoted_name
engine = create_engine("oracle+cx_oracle://some_dsn")
engine.has_table(quoted_name("some_table", True))
.. versionadded:: 0.9.0
+ .. versionchanged:: 1.2 The :class:`.quoted_name` construct is now
+ importable from ``sqlalchemy.sql``, in addition to the previous
+ location of ``sqlalchemy.sql.elements``.
+
"""
__slots__ = 'quote', 'lower', 'upper'
'nullslast',
'or_', 'outparam', 'outerjoin', 'over', 'select', 'subquery',
'table', 'text',
- 'tuple_', 'type_coerce', 'union', 'union_all', 'update', 'within_group',
+ 'tuple_', 'type_coerce', 'quoted_name', 'union', 'union_all', 'update',
+ 'within_group',
'TableSample', 'tablesample']
BindParameter, CollectionAggregate, UnaryExpression, BooleanClauseList, \
Label, Cast, Case, ColumnClause, TextClause, Over, Null, \
True_, False_, BinaryExpression, Tuple, TypeClause, Extract, \
- Grouping, WithinGroup, not_, \
+ Grouping, WithinGroup, not_, quoted_name, \
collate, literal_column, between,\
literal, outparam, TypeCoerce, ClauseList, FunctionFilter
# -*- encoding: utf-8
from sqlalchemy.testing import eq_, is_
from sqlalchemy import schema
-from sqlalchemy.sql import table, column
-from sqlalchemy.databases import mssql
+from sqlalchemy.sql import table, column, quoted_name
+from sqlalchemy.dialects import mssql
from sqlalchemy.dialects.mssql import mxodbc
from sqlalchemy.testing import fixtures, AssertsCompiledSQL
from sqlalchemy import sql
update, delete, insert, extract, union, func, PrimaryKeyConstraint, \
UniqueConstraint, Index, Sequence, literal
from sqlalchemy import testing
+from sqlalchemy.dialects.mssql import base
class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
"myid FROM mytable) AS foo, mytable WHERE "
"foo.myid = mytable.myid")
+ def test_force_schema_quoted_name_w_dot_case_insensitive(self):
+ metadata = MetaData()
+ tbl = Table(
+ 'test', metadata,
+ Column('id', Integer, primary_key=True),
+ schema=quoted_name("foo.dbo", True)
+ )
+ self.assert_compile(
+ select([tbl]),
+ "SELECT [foo.dbo].test.id FROM [foo.dbo].test"
+ )
+
+ def test_force_schema_quoted_w_dot_case_insensitive(self):
+ metadata = MetaData()
+ tbl = Table(
+ 'test', metadata,
+ Column('id', Integer, primary_key=True),
+ schema=quoted_name("foo.dbo", True)
+ )
+ self.assert_compile(
+ select([tbl]),
+ "SELECT [foo.dbo].test.id FROM [foo.dbo].test"
+ )
+
+ def test_force_schema_quoted_name_w_dot_case_sensitive(self):
+ metadata = MetaData()
+ tbl = Table(
+ 'test', metadata,
+ Column('id', Integer, primary_key=True),
+ schema=quoted_name("Foo.dbo", True)
+ )
+ self.assert_compile(
+ select([tbl]),
+ "SELECT [Foo.dbo].test.id FROM [Foo.dbo].test"
+ )
+
+ def test_force_schema_quoted_w_dot_case_sensitive(self):
+ metadata = MetaData()
+ tbl = Table(
+ 'test', metadata,
+ Column('id', Integer, primary_key=True),
+ schema="[Foo.dbo]"
+ )
+ self.assert_compile(
+ select([tbl]),
+ "SELECT [Foo.dbo].test.id FROM [Foo.dbo].test"
+ )
+
+ def test_schema_autosplit_w_dot_case_insensitive(self):
+ metadata = MetaData()
+ tbl = Table(
+ 'test', metadata,
+ Column('id', Integer, primary_key=True),
+ schema="foo.dbo"
+ )
+ self.assert_compile(
+ select([tbl]),
+ "SELECT foo.dbo.test.id FROM foo.dbo.test"
+ )
+
+ def test_schema_autosplit_w_dot_case_sensitive(self):
+ metadata = MetaData()
+ tbl = Table(
+ 'test', metadata,
+ Column('id', Integer, primary_key=True),
+ schema="Foo.dbo"
+ )
+ self.assert_compile(
+ select([tbl]),
+ "SELECT [Foo].dbo.test.id FROM [Foo].dbo.test"
+ )
+
+ def test_owner_database_pairs(self):
+ dialect = mssql.dialect()
+
+ for identifier, expected_schema, expected_owner in [
+ ("foo", None, "foo"),
+ ("foo.bar", "foo", "bar"),
+ ("Foo.Bar", "Foo", "Bar"),
+ ("[Foo.Bar]", None, "Foo.Bar"),
+ ("[Foo.Bar].[bat]", "Foo.Bar", "bat"),
+ ]:
+ schema, owner = base._owner_plus_db(dialect, identifier)
+
+ eq_(owner, expected_owner)
+ eq_(schema, expected_schema)
+
def test_delete_schema(self):
metadata = MetaData()
tbl = Table('test', metadata, Column('id', Integer,
"SELECT TOP 0 t.x, t.y FROM t WHERE t.x = :x_1 ORDER BY t.y",
checkparams={'x_1': 5}
)
- c = s.compile(dialect=mssql.MSDialect())
+ c = s.compile(dialect=mssql.dialect())
eq_(len(c._result_columns), 2)
assert t.c.x in set(c._create_result_map()['x'][1])
checkparams={'param_1': 20, 'x_1': 5}
)
- c = s.compile(dialect=mssql.MSDialect())
+ c = s.compile(dialect=mssql.dialect())
eq_(len(c._result_columns), 2)
assert t.c.x in set(c._create_result_map()['x'][1])
"WHERE mssql_rn > :param_1 AND mssql_rn <= :param_2 + :param_1",
checkparams={'param_1': 20, 'param_2': 10, 'x_1': 5}
)
- c = s.compile(dialect=mssql.MSDialect())
+ c = s.compile(dialect=mssql.dialect())
eq_(len(c._result_columns), 2)
assert t.c.x in set(c._create_result_map()['x'][1])
assert t.c.y in set(c._create_result_map()['y'][1])
"WHERE mssql_rn > :param_1 AND mssql_rn <= :param_2 + :param_1",
checkparams={'param_1': 20, 'param_2': 10, 'x_1': 5}
)
- c = s.compile(dialect=mssql.MSDialect())
+ c = s.compile(dialect=mssql.dialect())
eq_(len(c._result_columns), 4)
result_map = c._create_result_map()
checkparams={'param_1': 20, 'param_2': 10, 'x_1': 5}
)
- c = s.compile(dialect=mssql.MSDialect())
+ c = s.compile(dialect=mssql.dialect())
eq_(len(c._result_columns), 2)
assert t1.c.x in set(c._create_result_map()['x'][1])
assert t1.c.y in set(c._create_result_map()['y'][1])
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
