Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# dialects/mssql/base.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: ignore-errors
9"""
10.. dialect:: mssql
11 :name: Microsoft SQL Server
12 :normal_support: 2012+
13 :best_effort: 2005+
15.. _mssql_external_dialects:
17External Dialects
18-----------------
20In addition to the above DBAPI layers with native SQLAlchemy support, there
21are third-party dialects for other DBAPI layers that are compatible
22with SQL Server. See the "External Dialects" list on the
23:ref:`dialect_toplevel` page.
25.. _mssql_identity:
27Auto Increment Behavior / IDENTITY Columns
28------------------------------------------
30SQL Server provides so-called "auto incrementing" behavior using the
31``IDENTITY`` construct, which can be placed on any single integer column in a
32table. SQLAlchemy considers ``IDENTITY`` within its default "autoincrement"
33behavior for an integer primary key column, described at
34:paramref:`_schema.Column.autoincrement`. This means that by default,
35the first integer primary key column in a :class:`_schema.Table` will be
36considered to be the identity column - unless it is associated with a
37:class:`.Sequence` - and will generate DDL as such::
39 from sqlalchemy import Table, MetaData, Column, Integer
41 m = MetaData()
42 t = Table(
43 "t",
44 m,
45 Column("id", Integer, primary_key=True),
46 Column("x", Integer),
47 )
48 m.create_all(engine)
50The above example will generate DDL as:
52.. sourcecode:: sql
54 CREATE TABLE t (
55 id INTEGER NOT NULL IDENTITY,
56 x INTEGER NULL,
57 PRIMARY KEY (id)
58 )
60For the case where this default generation of ``IDENTITY`` is not desired,
61specify ``False`` for the :paramref:`_schema.Column.autoincrement` flag,
62on the first integer primary key column::
64 m = MetaData()
65 t = Table(
66 "t",
67 m,
68 Column("id", Integer, primary_key=True, autoincrement=False),
69 Column("x", Integer),
70 )
71 m.create_all(engine)
73To add the ``IDENTITY`` keyword to a non-primary key column, specify
74``True`` for the :paramref:`_schema.Column.autoincrement` flag on the desired
75:class:`_schema.Column` object, and ensure that
76:paramref:`_schema.Column.autoincrement`
77is set to ``False`` on any integer primary key column::
79 m = MetaData()
80 t = Table(
81 "t",
82 m,
83 Column("id", Integer, primary_key=True, autoincrement=False),
84 Column("x", Integer, autoincrement=True),
85 )
86 m.create_all(engine)
88.. versionchanged:: 1.4 Added :class:`_schema.Identity` construct
89 in a :class:`_schema.Column` to specify the start and increment
90 parameters of an IDENTITY. These replace
91 the use of the :class:`.Sequence` object in order to specify these values.
93.. deprecated:: 1.4
95 The ``mssql_identity_start`` and ``mssql_identity_increment`` parameters
96 to :class:`_schema.Column` are deprecated and should we replaced by
97 an :class:`_schema.Identity` object. Specifying both ways of configuring
98 an IDENTITY will result in a compile error.
99 These options are also no longer returned as part of the
100 ``dialect_options`` key in :meth:`_reflection.Inspector.get_columns`.
101 Use the information in the ``identity`` key instead.
103.. versionchanged:: 1.4 Removed the ability to use a :class:`.Sequence`
104 object to modify IDENTITY characteristics. :class:`.Sequence` objects
105 now only manipulate true T-SQL SEQUENCE types.
107.. note::
109 There can only be one IDENTITY column on the table. When using
110 ``autoincrement=True`` to enable the IDENTITY keyword, SQLAlchemy does not
111 guard against multiple columns specifying the option simultaneously. The
112 SQL Server database will instead reject the ``CREATE TABLE`` statement.
114.. note::
116 An INSERT statement which attempts to provide a value for a column that is
117 marked with IDENTITY will be rejected by SQL Server. In order for the
118 value to be accepted, a session-level option "SET IDENTITY_INSERT" must be
119 enabled. The SQLAlchemy SQL Server dialect will perform this operation
120 automatically when using a core :class:`_expression.Insert`
121 construct; if the
122 execution specifies a value for the IDENTITY column, the "IDENTITY_INSERT"
123 option will be enabled for the span of that statement's invocation.However,
124 this scenario is not high performing and should not be relied upon for
125 normal use. If a table doesn't actually require IDENTITY behavior in its
126 integer primary key column, the keyword should be disabled when creating
127 the table by ensuring that ``autoincrement=False`` is set.
129Controlling "Start" and "Increment"
130^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
132Specific control over the "start" and "increment" values for
133the ``IDENTITY`` generator are provided using the
134:paramref:`_schema.Identity.start` and :paramref:`_schema.Identity.increment`
135parameters passed to the :class:`_schema.Identity` object::
137 from sqlalchemy import Table, Integer, Column, Identity
139 test = Table(
140 "test",
141 metadata,
142 Column(
143 "id", Integer, primary_key=True, Identity(start=100, increment=10)
144 ),
145 Column("name", String(20)),
146 )
148The CREATE TABLE for the above :class:`_schema.Table` object would be:
150.. sourcecode:: sql
152 CREATE TABLE test (
153 id INTEGER NOT NULL IDENTITY(100,10) PRIMARY KEY,
154 name VARCHAR(20) NULL,
155 )
157.. note::
159 The :class:`_schema.Identity` object supports many other parameter in
160 addition to ``start`` and ``increment``. These are not supported by
161 SQL Server and will be ignored when generating the CREATE TABLE ddl.
164Using IDENTITY with Non-Integer numeric types
165^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
167SQL Server also allows ``IDENTITY`` to be used with ``NUMERIC`` columns. To
168implement this pattern smoothly in SQLAlchemy, the primary datatype of the
169column should remain as ``Integer``, however the underlying implementation
170type deployed to the SQL Server database can be specified as ``Numeric`` using
171:meth:`.TypeEngine.with_variant`::
173 from sqlalchemy import Column
174 from sqlalchemy import Integer
175 from sqlalchemy import Numeric
176 from sqlalchemy import String
177 from sqlalchemy.ext.declarative import declarative_base
179 Base = declarative_base()
182 class TestTable(Base):
183 __tablename__ = "test"
184 id = Column(
185 Integer().with_variant(Numeric(10, 0), "mssql"),
186 primary_key=True,
187 autoincrement=True,
188 )
189 name = Column(String)
191In the above example, ``Integer().with_variant()`` provides clear usage
192information that accurately describes the intent of the code. The general
193restriction that ``autoincrement`` only applies to ``Integer`` is established
194at the metadata level and not at the per-dialect level.
196When using the above pattern, the primary key identifier that comes back from
197the insertion of a row, which is also the value that would be assigned to an
198ORM object such as ``TestTable`` above, will be an instance of ``Decimal()``
199and not ``int`` when using SQL Server. The numeric return type of the
200:class:`_types.Numeric` type can be changed to return floats by passing False
201to :paramref:`_types.Numeric.asdecimal`. To normalize the return type of the
202above ``Numeric(10, 0)`` to return Python ints (which also support "long"
203integer values in Python 3), use :class:`_types.TypeDecorator` as follows::
205 from sqlalchemy import TypeDecorator
208 class NumericAsInteger(TypeDecorator):
209 "normalize floating point return values into ints"
211 impl = Numeric(10, 0, asdecimal=False)
212 cache_ok = True
214 def process_result_value(self, value, dialect):
215 if value is not None:
216 value = int(value)
217 return value
220 class TestTable(Base):
221 __tablename__ = "test"
222 id = Column(
223 Integer().with_variant(NumericAsInteger, "mssql"),
224 primary_key=True,
225 autoincrement=True,
226 )
227 name = Column(String)
229.. _mssql_insert_behavior:
231INSERT behavior
232^^^^^^^^^^^^^^^^
234Handling of the ``IDENTITY`` column at INSERT time involves two key
235techniques. The most common is being able to fetch the "last inserted value"
236for a given ``IDENTITY`` column, a process which SQLAlchemy performs
237implicitly in many cases, most importantly within the ORM.
239The process for fetching this value has several variants:
241* In the vast majority of cases, RETURNING is used in conjunction with INSERT
242 statements on SQL Server in order to get newly generated primary key values:
244 .. sourcecode:: sql
246 INSERT INTO t (x) OUTPUT inserted.id VALUES (?)
248 As of SQLAlchemy 2.0, the :ref:`engine_insertmanyvalues` feature is also
249 used by default to optimize many-row INSERT statements; for SQL Server
250 the feature takes place for both RETURNING and-non RETURNING
251 INSERT statements.
253 .. versionchanged:: 2.0.10 The :ref:`engine_insertmanyvalues` feature for
254 SQL Server was temporarily disabled for SQLAlchemy version 2.0.9 due to
255 issues with row ordering. As of 2.0.10 the feature is re-enabled, with
256 special case handling for the unit of work's requirement for RETURNING to
257 be ordered.
259* When RETURNING is not available or has been disabled via
260 ``implicit_returning=False``, either the ``scope_identity()`` function or
261 the ``@@identity`` variable is used; behavior varies by backend:
263 * when using PyODBC, the phrase ``; select scope_identity()`` will be
264 appended to the end of the INSERT statement; a second result set will be
265 fetched in order to receive the value. Given a table as::
267 t = Table(
268 "t",
269 metadata,
270 Column("id", Integer, primary_key=True),
271 Column("x", Integer),
272 implicit_returning=False,
273 )
275 an INSERT will look like:
277 .. sourcecode:: sql
279 INSERT INTO t (x) VALUES (?); select scope_identity()
281 * Other dialects such as pymssql will call upon
282 ``SELECT scope_identity() AS lastrowid`` subsequent to an INSERT
283 statement. If the flag ``use_scope_identity=False`` is passed to
284 :func:`_sa.create_engine`,
285 the statement ``SELECT @@identity AS lastrowid``
286 is used instead.
288A table that contains an ``IDENTITY`` column will prohibit an INSERT statement
289that refers to the identity column explicitly. The SQLAlchemy dialect will
290detect when an INSERT construct, created using a core
291:func:`_expression.insert`
292construct (not a plain string SQL), refers to the identity column, and
293in this case will emit ``SET IDENTITY_INSERT ON`` prior to the insert
294statement proceeding, and ``SET IDENTITY_INSERT OFF`` subsequent to the
295execution. Given this example::
297 m = MetaData()
298 t = Table(
299 "t", m, Column("id", Integer, primary_key=True), Column("x", Integer)
300 )
301 m.create_all(engine)
303 with engine.begin() as conn:
304 conn.execute(t.insert(), {"id": 1, "x": 1}, {"id": 2, "x": 2})
306The above column will be created with IDENTITY, however the INSERT statement
307we emit is specifying explicit values. In the echo output we can see
308how SQLAlchemy handles this:
310.. sourcecode:: sql
312 CREATE TABLE t (
313 id INTEGER NOT NULL IDENTITY(1,1),
314 x INTEGER NULL,
315 PRIMARY KEY (id)
316 )
318 COMMIT
319 SET IDENTITY_INSERT t ON
320 INSERT INTO t (id, x) VALUES (?, ?)
321 ((1, 1), (2, 2))
322 SET IDENTITY_INSERT t OFF
323 COMMIT
327This is an auxiliary use case suitable for testing and bulk insert scenarios.
329SEQUENCE support
330----------------
332The :class:`.Sequence` object creates "real" sequences, i.e.,
333``CREATE SEQUENCE``:
335.. sourcecode:: pycon+sql
337 >>> from sqlalchemy import Sequence
338 >>> from sqlalchemy.schema import CreateSequence
339 >>> from sqlalchemy.dialects import mssql
340 >>> print(
341 ... CreateSequence(Sequence("my_seq", start=1)).compile(
342 ... dialect=mssql.dialect()
343 ... )
344 ... )
345 {printsql}CREATE SEQUENCE my_seq START WITH 1
347For integer primary key generation, SQL Server's ``IDENTITY`` construct should
348generally be preferred vs. sequence.
350.. tip::
352 The default start value for T-SQL is ``-2**63`` instead of 1 as
353 in most other SQL databases. Users should explicitly set the
354 :paramref:`.Sequence.start` to 1 if that's the expected default::
356 seq = Sequence("my_sequence", start=1)
358.. versionadded:: 1.4 added SQL Server support for :class:`.Sequence`
360.. versionchanged:: 2.0 The SQL Server dialect will no longer implicitly
361 render "START WITH 1" for ``CREATE SEQUENCE``, which was the behavior
362 first implemented in version 1.4.
364MAX on VARCHAR / NVARCHAR
365-------------------------
367SQL Server supports the special string "MAX" within the
368:class:`_types.VARCHAR` and :class:`_types.NVARCHAR` datatypes,
369to indicate "maximum length possible". The dialect currently handles this as
370a length of "None" in the base type, rather than supplying a
371dialect-specific version of these types, so that a base type
372specified such as ``VARCHAR(None)`` can assume "unlengthed" behavior on
373more than one backend without using dialect-specific types.
375To build a SQL Server VARCHAR or NVARCHAR with MAX length, use None::
377 my_table = Table(
378 "my_table",
379 metadata,
380 Column("my_data", VARCHAR(None)),
381 Column("my_n_data", NVARCHAR(None)),
382 )
384Collation Support
385-----------------
387Character collations are supported by the base string types,
388specified by the string argument "collation"::
390 from sqlalchemy import VARCHAR
392 Column("login", VARCHAR(32, collation="Latin1_General_CI_AS"))
394When such a column is associated with a :class:`_schema.Table`, the
395CREATE TABLE statement for this column will yield:
397.. sourcecode:: sql
399 login VARCHAR(32) COLLATE Latin1_General_CI_AS NULL
401LIMIT/OFFSET Support
402--------------------
404MSSQL has added support for LIMIT / OFFSET as of SQL Server 2012, via the
405"OFFSET n ROWS" and "FETCH NEXT n ROWS" clauses. SQLAlchemy supports these
406syntaxes automatically if SQL Server 2012 or greater is detected.
408.. versionchanged:: 1.4 support added for SQL Server "OFFSET n ROWS" and
409 "FETCH NEXT n ROWS" syntax.
411For statements that specify only LIMIT and no OFFSET, all versions of SQL
412Server support the TOP keyword. This syntax is used for all SQL Server
413versions when no OFFSET clause is present. A statement such as::
415 select(some_table).limit(5)
417will render similarly to:
419.. sourcecode:: sql
421 SELECT TOP 5 col1, col2.. FROM table
423For versions of SQL Server prior to SQL Server 2012, a statement that uses
424LIMIT and OFFSET, or just OFFSET alone, will be rendered using the
425``ROW_NUMBER()`` window function. A statement such as::
427 select(some_table).order_by(some_table.c.col3).limit(5).offset(10)
429will render similarly to:
431.. sourcecode:: sql
433 SELECT anon_1.col1, anon_1.col2 FROM (SELECT col1, col2,
434 ROW_NUMBER() OVER (ORDER BY col3) AS
435 mssql_rn FROM table WHERE t.x = :x_1) AS
436 anon_1 WHERE mssql_rn > :param_1 AND mssql_rn <= :param_2 + :param_1
438Note that when using LIMIT and/or OFFSET, whether using the older
439or newer SQL Server syntaxes, the statement must have an ORDER BY as well,
440else a :class:`.CompileError` is raised.
442.. _mssql_comment_support:
444DDL Comment Support
445--------------------
447Comment support, which includes DDL rendering for attributes such as
448:paramref:`_schema.Table.comment` and :paramref:`_schema.Column.comment`, as
449well as the ability to reflect these comments, is supported assuming a
450supported version of SQL Server is in use. If a non-supported version such as
451Azure Synapse is detected at first-connect time (based on the presence
452of the ``fn_listextendedproperty`` SQL function), comment support including
453rendering and table-comment reflection is disabled, as both features rely upon
454SQL Server stored procedures and functions that are not available on all
455backend types.
457To force comment support to be on or off, bypassing autodetection, set the
458parameter ``supports_comments`` within :func:`_sa.create_engine`::
460 e = create_engine("mssql+pyodbc://u:p@dsn", supports_comments=False)
462.. versionadded:: 2.0 Added support for table and column comments for
463 the SQL Server dialect, including DDL generation and reflection.
465.. _mssql_isolation_level:
467Transaction Isolation Level
468---------------------------
470All SQL Server dialects support setting of transaction isolation level
471both via a dialect-specific parameter
472:paramref:`_sa.create_engine.isolation_level`
473accepted by :func:`_sa.create_engine`,
474as well as the :paramref:`.Connection.execution_options.isolation_level`
475argument as passed to
476:meth:`_engine.Connection.execution_options`.
477This feature works by issuing the
478command ``SET TRANSACTION ISOLATION LEVEL <level>`` for
479each new connection.
481To set isolation level using :func:`_sa.create_engine`::
483 engine = create_engine(
484 "mssql+pyodbc://scott:tiger@ms_2008", isolation_level="REPEATABLE READ"
485 )
487To set using per-connection execution options::
489 connection = engine.connect()
490 connection = connection.execution_options(isolation_level="READ COMMITTED")
492Valid values for ``isolation_level`` include:
494* ``AUTOCOMMIT`` - pyodbc / pymssql-specific
495* ``READ COMMITTED``
496* ``READ UNCOMMITTED``
497* ``REPEATABLE READ``
498* ``SERIALIZABLE``
499* ``SNAPSHOT`` - specific to SQL Server
501There are also more options for isolation level configurations, such as
502"sub-engine" objects linked to a main :class:`_engine.Engine` which each apply
503different isolation level settings. See the discussion at
504:ref:`dbapi_autocommit` for background.
506.. seealso::
508 :ref:`dbapi_autocommit`
510.. _mssql_reset_on_return:
512Temporary Table / Resource Reset for Connection Pooling
513-------------------------------------------------------
515The :class:`.QueuePool` connection pool implementation used
516by the SQLAlchemy :class:`.Engine` object includes
517:ref:`reset on return <pool_reset_on_return>` behavior that will invoke
518the DBAPI ``.rollback()`` method when connections are returned to the pool.
519While this rollback will clear out the immediate state used by the previous
520transaction, it does not cover a wider range of session-level state, including
521temporary tables as well as other server state such as prepared statement
522handles and statement caches. An undocumented SQL Server procedure known
523as ``sp_reset_connection`` is known to be a workaround for this issue which
524will reset most of the session state that builds up on a connection, including
525temporary tables.
527To install ``sp_reset_connection`` as the means of performing reset-on-return,
528the :meth:`.PoolEvents.reset` event hook may be used, as demonstrated in the
529example below. The :paramref:`_sa.create_engine.pool_reset_on_return` parameter
530is set to ``None`` so that the custom scheme can replace the default behavior
531completely. The custom hook implementation calls ``.rollback()`` in any case,
532as it's usually important that the DBAPI's own tracking of commit/rollback
533will remain consistent with the state of the transaction::
535 from sqlalchemy import create_engine
536 from sqlalchemy import event
538 mssql_engine = create_engine(
539 "mssql+pyodbc://scott:tiger^5HHH@mssql2017:1433/test?driver=ODBC+Driver+17+for+SQL+Server",
540 # disable default reset-on-return scheme
541 pool_reset_on_return=None,
542 )
545 @event.listens_for(mssql_engine, "reset")
546 def _reset_mssql(dbapi_connection, connection_record, reset_state):
547 if not reset_state.terminate_only:
548 dbapi_connection.execute("{call sys.sp_reset_connection}")
550 # so that the DBAPI itself knows that the connection has been
551 # reset
552 dbapi_connection.rollback()
554.. versionchanged:: 2.0.0b3 Added additional state arguments to
555 the :meth:`.PoolEvents.reset` event and additionally ensured the event
556 is invoked for all "reset" occurrences, so that it's appropriate
557 as a place for custom "reset" handlers. Previous schemes which
558 use the :meth:`.PoolEvents.checkin` handler remain usable as well.
560.. seealso::
562 :ref:`pool_reset_on_return` - in the :ref:`pooling_toplevel` documentation
564Nullability
565-----------
566MSSQL has support for three levels of column nullability. The default
567nullability allows nulls and is explicit in the CREATE TABLE
568construct:
570.. sourcecode:: sql
572 name VARCHAR(20) NULL
574If ``nullable=None`` is specified then no specification is made. In
575other words the database's configured default is used. This will
576render:
578.. sourcecode:: sql
580 name VARCHAR(20)
582If ``nullable`` is ``True`` or ``False`` then the column will be
583``NULL`` or ``NOT NULL`` respectively.
585Date / Time Handling
586--------------------
587DATE and TIME are supported. Bind parameters are converted
588to datetime.datetime() objects as required by most MSSQL drivers,
589and results are processed from strings if needed.
590The DATE and TIME types are not available for MSSQL 2005 and
591previous - if a server version below 2008 is detected, DDL
592for these types will be issued as DATETIME.
594.. _mssql_large_type_deprecation:
596Large Text/Binary Type Deprecation
597----------------------------------
599Per
600`SQL Server 2012/2014 Documentation <https://technet.microsoft.com/en-us/library/ms187993.aspx>`_,
601the ``NTEXT``, ``TEXT`` and ``IMAGE`` datatypes are to be removed from SQL
602Server in a future release. SQLAlchemy normally relates these types to the
603:class:`.UnicodeText`, :class:`_expression.TextClause` and
604:class:`.LargeBinary` datatypes.
606In order to accommodate this change, a new flag ``deprecate_large_types``
607is added to the dialect, which will be automatically set based on detection
608of the server version in use, if not otherwise set by the user. The
609behavior of this flag is as follows:
611* When this flag is ``True``, the :class:`.UnicodeText`,
612 :class:`_expression.TextClause` and
613 :class:`.LargeBinary` datatypes, when used to render DDL, will render the
614 types ``NVARCHAR(max)``, ``VARCHAR(max)``, and ``VARBINARY(max)``,
615 respectively. This is a new behavior as of the addition of this flag.
617* When this flag is ``False``, the :class:`.UnicodeText`,
618 :class:`_expression.TextClause` and
619 :class:`.LargeBinary` datatypes, when used to render DDL, will render the
620 types ``NTEXT``, ``TEXT``, and ``IMAGE``,
621 respectively. This is the long-standing behavior of these types.
623* The flag begins with the value ``None``, before a database connection is
624 established. If the dialect is used to render DDL without the flag being
625 set, it is interpreted the same as ``False``.
627* On first connection, the dialect detects if SQL Server version 2012 or
628 greater is in use; if the flag is still at ``None``, it sets it to ``True``
629 or ``False`` based on whether 2012 or greater is detected.
631* The flag can be set to either ``True`` or ``False`` when the dialect
632 is created, typically via :func:`_sa.create_engine`::
634 eng = create_engine(
635 "mssql+pymssql://user:pass@host/db", deprecate_large_types=True
636 )
638* Complete control over whether the "old" or "new" types are rendered is
639 available in all SQLAlchemy versions by using the UPPERCASE type objects
640 instead: :class:`_types.NVARCHAR`, :class:`_types.VARCHAR`,
641 :class:`_types.VARBINARY`, :class:`_types.TEXT`, :class:`_mssql.NTEXT`,
642 :class:`_mssql.IMAGE`
643 will always remain fixed and always output exactly that
644 type.
646.. _multipart_schema_names:
648Multipart Schema Names
649----------------------
651SQL Server schemas sometimes require multiple parts to their "schema"
652qualifier, that is, including the database name and owner name as separate
653tokens, such as ``mydatabase.dbo.some_table``. These multipart names can be set
654at once using the :paramref:`_schema.Table.schema` argument of
655:class:`_schema.Table`::
657 Table(
658 "some_table",
659 metadata,
660 Column("q", String(50)),
661 schema="mydatabase.dbo",
662 )
664When performing operations such as table or component reflection, a schema
665argument that contains a dot will be split into separate
666"database" and "owner" components in order to correctly query the SQL
667Server information schema tables, as these two values are stored separately.
668Additionally, when rendering the schema name for DDL or SQL, the two
669components will be quoted separately for case sensitive names and other
670special characters. Given an argument as below::
672 Table(
673 "some_table",
674 metadata,
675 Column("q", String(50)),
676 schema="MyDataBase.dbo",
677 )
679The above schema would be rendered as ``[MyDataBase].dbo``, and also in
680reflection, would be reflected using "dbo" as the owner and "MyDataBase"
681as the database name.
683To control how the schema name is broken into database / owner,
684specify brackets (which in SQL Server are quoting characters) in the name.
685Below, the "owner" will be considered as ``MyDataBase.dbo`` and the
686"database" will be None::
688 Table(
689 "some_table",
690 metadata,
691 Column("q", String(50)),
692 schema="[MyDataBase.dbo]",
693 )
695To individually specify both database and owner name with special characters
696or embedded dots, use two sets of brackets::
698 Table(
699 "some_table",
700 metadata,
701 Column("q", String(50)),
702 schema="[MyDataBase.Period].[MyOwner.Dot]",
703 )
705.. _legacy_schema_rendering:
707Legacy Schema Mode
708------------------
710Very old versions of the MSSQL dialect introduced the behavior such that a
711schema-qualified table would be auto-aliased when used in a
712SELECT statement; given a table::
714 account_table = Table(
715 "account",
716 metadata,
717 Column("id", Integer, primary_key=True),
718 Column("info", String(100)),
719 schema="customer_schema",
720 )
722this legacy mode of rendering would assume that "customer_schema.account"
723would not be accepted by all parts of the SQL statement, as illustrated
724below:
726.. sourcecode:: pycon+sql
728 >>> eng = create_engine("mssql+pymssql://mydsn", legacy_schema_aliasing=True)
729 >>> print(account_table.select().compile(eng))
730 {printsql}SELECT account_1.id, account_1.info
731 FROM customer_schema.account AS account_1
733This mode of behavior is now off by default, as it appears to have served
734no purpose; however in the case that legacy applications rely upon it,
735it is available using the ``legacy_schema_aliasing`` argument to
736:func:`_sa.create_engine` as illustrated above.
738.. deprecated:: 1.4
740 The ``legacy_schema_aliasing`` flag is now
741 deprecated and will be removed in a future release.
743.. _mssql_indexes:
745Clustered Index Support
746-----------------------
748The MSSQL dialect supports clustered indexes (and primary keys) via the
749``mssql_clustered`` option. This option is available to :class:`.Index`,
750:class:`.UniqueConstraint`. and :class:`.PrimaryKeyConstraint`.
751For indexes this option can be combined with the ``mssql_columnstore`` one
752to create a clustered columnstore index.
754To generate a clustered index::
756 Index("my_index", table.c.x, mssql_clustered=True)
758which renders the index as ``CREATE CLUSTERED INDEX my_index ON table (x)``.
760To generate a clustered primary key use::
762 Table(
763 "my_table",
764 metadata,
765 Column("x", ...),
766 Column("y", ...),
767 PrimaryKeyConstraint("x", "y", mssql_clustered=True),
768 )
770which will render the table, for example, as:
772.. sourcecode:: sql
774 CREATE TABLE my_table (
775 x INTEGER NOT NULL,
776 y INTEGER NOT NULL,
777 PRIMARY KEY CLUSTERED (x, y)
778 )
780Similarly, we can generate a clustered unique constraint using::
782 Table(
783 "my_table",
784 metadata,
785 Column("x", ...),
786 Column("y", ...),
787 PrimaryKeyConstraint("x"),
788 UniqueConstraint("y", mssql_clustered=True),
789 )
791To explicitly request a non-clustered primary key (for example, when
792a separate clustered index is desired), use::
794 Table(
795 "my_table",
796 metadata,
797 Column("x", ...),
798 Column("y", ...),
799 PrimaryKeyConstraint("x", "y", mssql_clustered=False),
800 )
802which will render the table, for example, as:
804.. sourcecode:: sql
806 CREATE TABLE my_table (
807 x INTEGER NOT NULL,
808 y INTEGER NOT NULL,
809 PRIMARY KEY NONCLUSTERED (x, y)
810 )
812Columnstore Index Support
813-------------------------
815The MSSQL dialect supports columnstore indexes via the ``mssql_columnstore``
816option. This option is available to :class:`.Index`. It be combined with
817the ``mssql_clustered`` option to create a clustered columnstore index.
819To generate a columnstore index::
821 Index("my_index", table.c.x, mssql_columnstore=True)
823which renders the index as ``CREATE COLUMNSTORE INDEX my_index ON table (x)``.
825To generate a clustered columnstore index provide no columns::
827 idx = Index("my_index", mssql_clustered=True, mssql_columnstore=True)
828 # required to associate the index with the table
829 table.append_constraint(idx)
831the above renders the index as
832``CREATE CLUSTERED COLUMNSTORE INDEX my_index ON table``.
834.. versionadded:: 2.0.18
836MSSQL-Specific Index Options
837-----------------------------
839In addition to clustering, the MSSQL dialect supports other special options
840for :class:`.Index`.
842INCLUDE
843^^^^^^^
845The ``mssql_include`` option renders INCLUDE(colname) for the given string
846names::
848 Index("my_index", table.c.x, mssql_include=["y"])
850would render the index as ``CREATE INDEX my_index ON table (x) INCLUDE (y)``
852.. _mssql_index_where:
854Filtered Indexes
855^^^^^^^^^^^^^^^^
857The ``mssql_where`` option renders WHERE(condition) for the given string
858names::
860 Index("my_index", table.c.x, mssql_where=table.c.x > 10)
862would render the index as ``CREATE INDEX my_index ON table (x) WHERE x > 10``.
864Index ordering
865^^^^^^^^^^^^^^
867Index ordering is available via functional expressions, such as::
869 Index("my_index", table.c.x.desc())
871would render the index as ``CREATE INDEX my_index ON table (x DESC)``
873.. seealso::
875 :ref:`schema_indexes_functional`
877Compatibility Levels
878--------------------
879MSSQL supports the notion of setting compatibility levels at the
880database level. This allows, for instance, to run a database that
881is compatible with SQL2000 while running on a SQL2005 database
882server. ``server_version_info`` will always return the database
883server version information (in this case SQL2005) and not the
884compatibility level information. Because of this, if running under
885a backwards compatibility mode SQLAlchemy may attempt to use T-SQL
886statements that are unable to be parsed by the database server.
888.. _mssql_triggers:
890Triggers
891--------
893SQLAlchemy by default uses OUTPUT INSERTED to get at newly
894generated primary key values via IDENTITY columns or other
895server side defaults. MS-SQL does not
896allow the usage of OUTPUT INSERTED on tables that have triggers.
897To disable the usage of OUTPUT INSERTED on a per-table basis,
898specify ``implicit_returning=False`` for each :class:`_schema.Table`
899which has triggers::
901 Table(
902 "mytable",
903 metadata,
904 Column("id", Integer, primary_key=True),
905 # ...,
906 implicit_returning=False,
907 )
909Declarative form::
911 class MyClass(Base):
912 # ...
913 __table_args__ = {"implicit_returning": False}
915.. _mssql_rowcount_versioning:
917Rowcount Support / ORM Versioning
918---------------------------------
920The SQL Server drivers may have limited ability to return the number
921of rows updated from an UPDATE or DELETE statement.
923As of this writing, the PyODBC driver is not able to return a rowcount when
924OUTPUT INSERTED is used. Previous versions of SQLAlchemy therefore had
925limitations for features such as the "ORM Versioning" feature that relies upon
926accurate rowcounts in order to match version numbers with matched rows.
928SQLAlchemy 2.0 now retrieves the "rowcount" manually for these particular use
929cases based on counting the rows that arrived back within RETURNING; so while
930the driver still has this limitation, the ORM Versioning feature is no longer
931impacted by it. As of SQLAlchemy 2.0.5, ORM versioning has been fully
932re-enabled for the pyodbc driver.
934.. versionchanged:: 2.0.5 ORM versioning support is restored for the pyodbc
935 driver. Previously, a warning would be emitted during ORM flush that
936 versioning was not supported.
939Enabling Snapshot Isolation
940---------------------------
942SQL Server has a default transaction
943isolation mode that locks entire tables, and causes even mildly concurrent
944applications to have long held locks and frequent deadlocks.
945Enabling snapshot isolation for the database as a whole is recommended
946for modern levels of concurrency support. This is accomplished via the
947following ALTER DATABASE commands executed at the SQL prompt:
949.. sourcecode:: sql
951 ALTER DATABASE MyDatabase SET ALLOW_SNAPSHOT_ISOLATION ON
953 ALTER DATABASE MyDatabase SET READ_COMMITTED_SNAPSHOT ON
955Background on SQL Server snapshot isolation is available at
956https://msdn.microsoft.com/en-us/library/ms175095.aspx.
958""" # noqa
960from __future__ import annotations
962import codecs
963import datetime
964from functools import lru_cache
965import operator
966import re
967from typing import Any
968from typing import Literal
969from typing import overload
970from typing import TYPE_CHECKING
971from uuid import UUID as _python_UUID
973from . import information_schema as ischema
974from .json import JSON
975from .json import JSONIndexType
976from .json import JSONPathType
977from ... import exc
978from ... import Identity
979from ... import schema as sa_schema
980from ... import Sequence
981from ... import sql
982from ... import text
983from ... import util
984from ...engine import cursor as _cursor
985from ...engine import default
986from ...engine import ObjectKind
987from ...engine import ObjectScope
988from ...engine import reflection
989from ...engine.reflection import ReflectionDefaults
990from ...sql import coercions
991from ...sql import compiler
992from ...sql import elements
993from ...sql import expression
994from ...sql import quoted_name
995from ...sql import roles
996from ...sql import sqltypes
997from ...sql import try_cast as try_cast # noqa: F401
998from ...sql import util as sql_util
999from ...sql._typing import is_sql_compiler
1000from ...sql.compiler import AggregateOrderByStyle
1001from ...sql.compiler import InsertmanyvaluesSentinelOpts
1002from ...sql.elements import TryCast as TryCast # noqa: F401
1003from ...types import BIGINT
1004from ...types import BINARY
1005from ...types import CHAR
1006from ...types import DATE
1007from ...types import DATETIME
1008from ...types import DECIMAL
1009from ...types import FLOAT
1010from ...types import INTEGER
1011from ...types import NCHAR
1012from ...types import NUMERIC
1013from ...types import NVARCHAR
1014from ...types import SMALLINT
1015from ...types import TEXT
1016from ...types import VARCHAR
1017from ...util import update_wrapper
1019if TYPE_CHECKING:
1020 from ...sql.ddl import DropIndex
1021 from ...sql.dml import DMLState
1022 from ...sql.selectable import TableClause
1024# https://sqlserverbuilds.blogspot.com/
1025MS_2017_VERSION = (14,)
1026MS_2016_VERSION = (13,)
1027MS_2014_VERSION = (12,)
1028MS_2012_VERSION = (11,)
1029MS_2008_VERSION = (10,)
1030MS_2005_VERSION = (9,)
1031MS_2000_VERSION = (8,)
1033RESERVED_WORDS = {
1034 "add",
1035 "all",
1036 "alter",
1037 "and",
1038 "any",
1039 "as",
1040 "asc",
1041 "authorization",
1042 "backup",
1043 "begin",
1044 "between",
1045 "break",
1046 "browse",
1047 "bulk",
1048 "by",
1049 "cascade",
1050 "case",
1051 "check",
1052 "checkpoint",
1053 "close",
1054 "clustered",
1055 "coalesce",
1056 "collate",
1057 "column",
1058 "commit",
1059 "compute",
1060 "constraint",
1061 "contains",
1062 "containstable",
1063 "continue",
1064 "convert",
1065 "create",
1066 "cross",
1067 "current",
1068 "current_date",
1069 "current_time",
1070 "current_timestamp",
1071 "current_user",
1072 "cursor",
1073 "database",
1074 "dbcc",
1075 "deallocate",
1076 "declare",
1077 "default",
1078 "delete",
1079 "deny",
1080 "desc",
1081 "disk",
1082 "distinct",
1083 "distributed",
1084 "double",
1085 "drop",
1086 "dump",
1087 "else",
1088 "end",
1089 "errlvl",
1090 "escape",
1091 "except",
1092 "exec",
1093 "execute",
1094 "exists",
1095 "exit",
1096 "external",
1097 "fetch",
1098 "file",
1099 "fillfactor",
1100 "for",
1101 "foreign",
1102 "freetext",
1103 "freetexttable",
1104 "from",
1105 "full",
1106 "function",
1107 "goto",
1108 "grant",
1109 "group",
1110 "having",
1111 "holdlock",
1112 "identity",
1113 "identity_insert",
1114 "identitycol",
1115 "if",
1116 "in",
1117 "index",
1118 "inner",
1119 "insert",
1120 "intersect",
1121 "into",
1122 "is",
1123 "join",
1124 "key",
1125 "kill",
1126 "left",
1127 "like",
1128 "lineno",
1129 "load",
1130 "merge",
1131 "national",
1132 "nocheck",
1133 "nonclustered",
1134 "not",
1135 "null",
1136 "nullif",
1137 "of",
1138 "off",
1139 "offsets",
1140 "on",
1141 "open",
1142 "opendatasource",
1143 "openquery",
1144 "openrowset",
1145 "openxml",
1146 "option",
1147 "or",
1148 "order",
1149 "outer",
1150 "over",
1151 "percent",
1152 "pivot",
1153 "plan",
1154 "precision",
1155 "primary",
1156 "print",
1157 "proc",
1158 "procedure",
1159 "public",
1160 "raiserror",
1161 "read",
1162 "readtext",
1163 "reconfigure",
1164 "references",
1165 "replication",
1166 "restore",
1167 "restrict",
1168 "return",
1169 "revert",
1170 "revoke",
1171 "right",
1172 "rollback",
1173 "rowcount",
1174 "rowguidcol",
1175 "rule",
1176 "save",
1177 "schema",
1178 "securityaudit",
1179 "select",
1180 "session_user",
1181 "set",
1182 "setuser",
1183 "shutdown",
1184 "some",
1185 "statistics",
1186 "system_user",
1187 "table",
1188 "tablesample",
1189 "textsize",
1190 "then",
1191 "to",
1192 "top",
1193 "tran",
1194 "transaction",
1195 "trigger",
1196 "truncate",
1197 "tsequal",
1198 "union",
1199 "unique",
1200 "unpivot",
1201 "update",
1202 "updatetext",
1203 "use",
1204 "user",
1205 "values",
1206 "varying",
1207 "view",
1208 "waitfor",
1209 "when",
1210 "where",
1211 "while",
1212 "with",
1213 "writetext",
1214}
1217class REAL(sqltypes.REAL):
1218 """the SQL Server REAL datatype."""
1220 def __init__(self, **kw):
1221 # REAL is a synonym for FLOAT(24) on SQL server.
1222 # it is only accepted as the word "REAL" in DDL, the numeric
1223 # precision value is not allowed to be present
1224 kw.setdefault("precision", 24)
1225 super().__init__(**kw)
1228class DOUBLE_PRECISION(sqltypes.DOUBLE_PRECISION):
1229 """the SQL Server DOUBLE PRECISION datatype.
1231 .. versionadded:: 2.0.11
1233 """
1235 def __init__(self, **kw):
1236 # DOUBLE PRECISION is a synonym for FLOAT(53) on SQL server.
1237 # it is only accepted as the word "DOUBLE PRECISION" in DDL,
1238 # the numeric precision value is not allowed to be present
1239 kw.setdefault("precision", 53)
1240 super().__init__(**kw)
1243class TINYINT(sqltypes.Integer):
1244 __visit_name__ = "TINYINT"
1247# MSSQL DATE/TIME types have varied behavior, sometimes returning
1248# strings. MSDate/TIME check for everything, and always
1249# filter bind parameters into datetime objects (required by pyodbc,
1250# not sure about other dialects).
1253class _MSDate(sqltypes.Date):
1254 def bind_processor(self, dialect):
1255 def process(value):
1256 if type(value) == datetime.date:
1257 return datetime.datetime(value.year, value.month, value.day)
1258 else:
1259 return value
1261 return process
1263 _reg = re.compile(r"(\d+)-(\d+)-(\d+)")
1265 def result_processor(self, dialect, coltype):
1266 def process(value):
1267 if isinstance(value, datetime.datetime):
1268 return value.date()
1269 elif isinstance(value, str):
1270 m = self._reg.match(value)
1271 if not m:
1272 raise ValueError(
1273 "could not parse %r as a date value" % (value,)
1274 )
1275 return datetime.date(*[int(x or 0) for x in m.groups()])
1276 else:
1277 return value
1279 return process
1282class TIME(sqltypes.TIME):
1283 def __init__(self, precision=None, **kwargs):
1284 self.precision = precision
1285 super().__init__()
1287 __zero_date = datetime.date(1900, 1, 1)
1289 def bind_processor(self, dialect):
1290 def process(value):
1291 if isinstance(value, datetime.datetime):
1292 value = datetime.datetime.combine(
1293 self.__zero_date, value.time()
1294 )
1295 elif isinstance(value, datetime.time):
1296 """issue #5339
1297 per: https://github.com/mkleehammer/pyodbc/wiki/Tips-and-Tricks-by-Database-Platform#time-columns
1298 pass TIME value as string
1299 """ # noqa
1300 value = str(value)
1301 return value
1303 return process
1305 _reg = re.compile(r"(\d+):(\d+):(\d+)(?:\.(\d{0,6}))?")
1307 def result_processor(self, dialect, coltype):
1308 def process(value):
1309 if isinstance(value, datetime.datetime):
1310 return value.time()
1311 elif isinstance(value, str):
1312 m = self._reg.match(value)
1313 if not m:
1314 raise ValueError(
1315 "could not parse %r as a time value" % (value,)
1316 )
1317 return datetime.time(*[int(x or 0) for x in m.groups()])
1318 else:
1319 return value
1321 return process
1324_MSTime = TIME
1327class _BASETIMEIMPL(TIME):
1328 __visit_name__ = "_BASETIMEIMPL"
1331class _DateTimeBase:
1332 def bind_processor(self, dialect):
1333 def process(value):
1334 if type(value) == datetime.date:
1335 return datetime.datetime(value.year, value.month, value.day)
1336 else:
1337 return value
1339 return process
1342class _MSDateTime(_DateTimeBase, sqltypes.DateTime):
1343 pass
1346class SMALLDATETIME(_DateTimeBase, sqltypes.DateTime):
1347 __visit_name__ = "SMALLDATETIME"
1350class DATETIME2(_DateTimeBase, sqltypes.DateTime):
1351 __visit_name__ = "DATETIME2"
1353 def __init__(self, precision=None, **kw):
1354 super().__init__(**kw)
1355 self.precision = precision
1358class DATETIMEOFFSET(_DateTimeBase, sqltypes.DateTime):
1359 __visit_name__ = "DATETIMEOFFSET"
1361 def __init__(self, precision=None, **kw):
1362 super().__init__(**kw)
1363 self.precision = precision
1366class _UnicodeLiteral:
1367 def literal_processor(self, dialect):
1368 def process(value):
1369 value = value.replace("'", "''")
1371 if dialect.identifier_preparer._double_percents:
1372 value = value.replace("%", "%%")
1374 return "N'%s'" % value
1376 return process
1379class _MSUnicode(_UnicodeLiteral, sqltypes.Unicode):
1380 pass
1383class _MSUnicodeText(_UnicodeLiteral, sqltypes.UnicodeText):
1384 pass
1387class TIMESTAMP(sqltypes._Binary):
1388 """Implement the SQL Server TIMESTAMP type.
1390 Note this is **completely different** than the SQL Standard
1391 TIMESTAMP type, which is not supported by SQL Server. It
1392 is a read-only datatype that does not support INSERT of values.
1394 .. seealso::
1396 :class:`_mssql.ROWVERSION`
1398 """
1400 __visit_name__ = "TIMESTAMP"
1402 # expected by _Binary to be present
1403 length = None
1405 def __init__(self, convert_int=False):
1406 """Construct a TIMESTAMP or ROWVERSION type.
1408 :param convert_int: if True, binary integer values will
1409 be converted to integers on read.
1411 """
1412 self.convert_int = convert_int
1414 def result_processor(self, dialect, coltype):
1415 super_ = super().result_processor(dialect, coltype)
1416 if self.convert_int:
1418 def process(value):
1419 if super_:
1420 value = super_(value)
1421 if value is not None:
1422 # https://stackoverflow.com/a/30403242/34549
1423 value = int(codecs.encode(value, "hex"), 16)
1424 return value
1426 return process
1427 else:
1428 return super_
1431class ROWVERSION(TIMESTAMP):
1432 """Implement the SQL Server ROWVERSION type.
1434 The ROWVERSION datatype is a SQL Server synonym for the TIMESTAMP
1435 datatype, however current SQL Server documentation suggests using
1436 ROWVERSION for new datatypes going forward.
1438 The ROWVERSION datatype does **not** reflect (e.g. introspect) from the
1439 database as itself; the returned datatype will be
1440 :class:`_mssql.TIMESTAMP`.
1442 This is a read-only datatype that does not support INSERT of values.
1444 .. seealso::
1446 :class:`_mssql.TIMESTAMP`
1448 """
1450 __visit_name__ = "ROWVERSION"
1453class NTEXT(sqltypes.UnicodeText):
1454 """MSSQL NTEXT type, for variable-length unicode text up to 2^30
1455 characters."""
1457 __visit_name__ = "NTEXT"
1460class VARBINARY(sqltypes.VARBINARY, sqltypes.LargeBinary):
1461 """The MSSQL VARBINARY type.
1463 This type adds additional features to the core :class:`_types.VARBINARY`
1464 type, including "deprecate_large_types" mode where
1465 either ``VARBINARY(max)`` or IMAGE is rendered, as well as the SQL
1466 Server ``FILESTREAM`` option.
1468 .. seealso::
1470 :ref:`mssql_large_type_deprecation`
1472 """
1474 __visit_name__ = "VARBINARY"
1476 def __init__(self, length=None, filestream=False):
1477 """
1478 Construct a VARBINARY type.
1480 :param length: optional, a length for the column for use in
1481 DDL statements, for those binary types that accept a length,
1482 such as the MySQL BLOB type.
1484 :param filestream=False: if True, renders the ``FILESTREAM`` keyword
1485 in the table definition. In this case ``length`` must be ``None``
1486 or ``'max'``.
1488 .. versionadded:: 1.4.31
1490 """
1492 self.filestream = filestream
1493 if self.filestream and length not in (None, "max"):
1494 raise ValueError(
1495 "length must be None or 'max' when setting filestream"
1496 )
1497 super().__init__(length=length)
1500class IMAGE(sqltypes.LargeBinary):
1501 __visit_name__ = "IMAGE"
1504class XML(sqltypes.Text):
1505 """MSSQL XML type.
1507 This is a placeholder type for reflection purposes that does not include
1508 any Python-side datatype support. It also does not currently support
1509 additional arguments, such as "CONTENT", "DOCUMENT",
1510 "xml_schema_collection".
1512 """
1514 __visit_name__ = "XML"
1517class BIT(sqltypes.Boolean):
1518 """MSSQL BIT type.
1520 Both pyodbc and pymssql return values from BIT columns as
1521 Python <class 'bool'> so just subclass Boolean.
1523 """
1525 __visit_name__ = "BIT"
1528class MONEY(sqltypes.TypeEngine):
1529 __visit_name__ = "MONEY"
1532class SMALLMONEY(sqltypes.TypeEngine):
1533 __visit_name__ = "SMALLMONEY"
1536class MSUUid(sqltypes.Uuid):
1537 def bind_processor(self, dialect):
1538 if self.native_uuid:
1539 # this is currently assuming pyodbc; might not work for
1540 # some other mssql driver
1541 return None
1542 else:
1543 if self.as_uuid:
1545 def process(value):
1546 if value is not None:
1547 value = value.hex
1548 return value
1550 return process
1551 else:
1553 def process(value):
1554 if value is not None:
1555 value = value.replace("-", "").replace("''", "'")
1556 return value
1558 return process
1560 def literal_processor(self, dialect):
1561 if self.native_uuid:
1563 def process(value):
1564 return f"""'{str(value).replace("'", "''")}'"""
1566 return process
1567 else:
1568 if self.as_uuid:
1570 def process(value):
1571 return f"""'{value.hex}'"""
1573 return process
1574 else:
1576 def process(value):
1577 return f"""'{
1578 value.replace("-", "").replace("'", "''")
1579 }'"""
1581 return process
1584class UNIQUEIDENTIFIER(sqltypes.Uuid[sqltypes._UUID_RETURN]):
1585 __visit_name__ = "UNIQUEIDENTIFIER"
1587 @overload
1588 def __init__(
1589 self: UNIQUEIDENTIFIER[_python_UUID], as_uuid: Literal[True] = ...
1590 ): ...
1592 @overload
1593 def __init__(
1594 self: UNIQUEIDENTIFIER[str], as_uuid: Literal[False] = ...
1595 ): ...
1597 def __init__(self, as_uuid: bool = True):
1598 """Construct a :class:`_mssql.UNIQUEIDENTIFIER` type.
1601 :param as_uuid=True: if True, values will be interpreted
1602 as Python uuid objects, converting to/from string via the
1603 DBAPI.
1605 .. versionchanged:: 2.0 Added direct "uuid" support to the
1606 :class:`_mssql.UNIQUEIDENTIFIER` datatype; uuid interpretation
1607 defaults to ``True``.
1609 """
1610 self.as_uuid = as_uuid
1611 self.native_uuid = True
1614class SQL_VARIANT(sqltypes.TypeEngine):
1615 __visit_name__ = "SQL_VARIANT"
1618# old names.
1619MSDateTime = _MSDateTime
1620MSDate = _MSDate
1621MSReal = REAL
1622MSTinyInteger = TINYINT
1623MSTime = TIME
1624MSSmallDateTime = SMALLDATETIME
1625MSDateTime2 = DATETIME2
1626MSDateTimeOffset = DATETIMEOFFSET
1627MSText = TEXT
1628MSNText = NTEXT
1629MSString = VARCHAR
1630MSNVarchar = NVARCHAR
1631MSChar = CHAR
1632MSNChar = NCHAR
1633MSBinary = BINARY
1634MSVarBinary = VARBINARY
1635MSImage = IMAGE
1636MSBit = BIT
1637MSMoney = MONEY
1638MSSmallMoney = SMALLMONEY
1639MSUniqueIdentifier = UNIQUEIDENTIFIER
1640MSVariant = SQL_VARIANT
1642ischema_names = {
1643 "int": INTEGER,
1644 "bigint": BIGINT,
1645 "smallint": SMALLINT,
1646 "tinyint": TINYINT,
1647 "varchar": VARCHAR,
1648 "nvarchar": NVARCHAR,
1649 "char": CHAR,
1650 "nchar": NCHAR,
1651 "text": TEXT,
1652 "ntext": NTEXT,
1653 "decimal": DECIMAL,
1654 "numeric": NUMERIC,
1655 "float": FLOAT,
1656 "datetime": DATETIME,
1657 "datetime2": DATETIME2,
1658 "datetimeoffset": DATETIMEOFFSET,
1659 "date": DATE,
1660 "time": TIME,
1661 "smalldatetime": SMALLDATETIME,
1662 "binary": BINARY,
1663 "varbinary": VARBINARY,
1664 "bit": BIT,
1665 "real": REAL,
1666 "double precision": DOUBLE_PRECISION,
1667 "image": IMAGE,
1668 "xml": XML,
1669 "timestamp": TIMESTAMP,
1670 "money": MONEY,
1671 "smallmoney": SMALLMONEY,
1672 "uniqueidentifier": UNIQUEIDENTIFIER,
1673 "sql_variant": SQL_VARIANT,
1674}
1677class MSTypeCompiler(compiler.GenericTypeCompiler):
1678 def _extend(self, spec, type_, length=None):
1679 """Extend a string-type declaration with standard SQL
1680 COLLATE annotations.
1682 """
1684 if getattr(type_, "collation", None):
1685 collation = "COLLATE %s" % type_.collation
1686 else:
1687 collation = None
1689 if not length:
1690 length = type_.length
1692 if length:
1693 spec = spec + "(%s)" % length
1695 return " ".join([c for c in (spec, collation) if c is not None])
1697 def visit_double(self, type_, **kw):
1698 return self.visit_DOUBLE_PRECISION(type_, **kw)
1700 def visit_FLOAT(self, type_, **kw):
1701 precision = getattr(type_, "precision", None)
1702 if precision is None:
1703 return "FLOAT"
1704 else:
1705 return "FLOAT(%(precision)s)" % {"precision": precision}
1707 def visit_TINYINT(self, type_, **kw):
1708 return "TINYINT"
1710 def visit_TIME(self, type_, **kw):
1711 precision = getattr(type_, "precision", None)
1712 if precision is not None:
1713 return "TIME(%s)" % precision
1714 else:
1715 return "TIME"
1717 def visit_TIMESTAMP(self, type_, **kw):
1718 return "TIMESTAMP"
1720 def visit_ROWVERSION(self, type_, **kw):
1721 return "ROWVERSION"
1723 def visit_datetime(self, type_, **kw):
1724 if type_.timezone:
1725 return self.visit_DATETIMEOFFSET(type_, **kw)
1726 else:
1727 return self.visit_DATETIME(type_, **kw)
1729 def visit_DATETIMEOFFSET(self, type_, **kw):
1730 precision = getattr(type_, "precision", None)
1731 if precision is not None:
1732 return "DATETIMEOFFSET(%s)" % type_.precision
1733 else:
1734 return "DATETIMEOFFSET"
1736 def visit_DATETIME2(self, type_, **kw):
1737 precision = getattr(type_, "precision", None)
1738 if precision is not None:
1739 return "DATETIME2(%s)" % precision
1740 else:
1741 return "DATETIME2"
1743 def visit_SMALLDATETIME(self, type_, **kw):
1744 return "SMALLDATETIME"
1746 def visit_unicode(self, type_, **kw):
1747 return self.visit_NVARCHAR(type_, **kw)
1749 def visit_text(self, type_, **kw):
1750 if self.dialect.deprecate_large_types:
1751 return self.visit_VARCHAR(type_, **kw)
1752 else:
1753 return self.visit_TEXT(type_, **kw)
1755 def visit_unicode_text(self, type_, **kw):
1756 if self.dialect.deprecate_large_types:
1757 return self.visit_NVARCHAR(type_, **kw)
1758 else:
1759 return self.visit_NTEXT(type_, **kw)
1761 def visit_NTEXT(self, type_, **kw):
1762 return self._extend("NTEXT", type_)
1764 def visit_TEXT(self, type_, **kw):
1765 return self._extend("TEXT", type_)
1767 def visit_VARCHAR(self, type_, **kw):
1768 return self._extend("VARCHAR", type_, length=type_.length or "max")
1770 def visit_CHAR(self, type_, **kw):
1771 return self._extend("CHAR", type_)
1773 def visit_NCHAR(self, type_, **kw):
1774 return self._extend("NCHAR", type_)
1776 def visit_NVARCHAR(self, type_, **kw):
1777 return self._extend("NVARCHAR", type_, length=type_.length or "max")
1779 def visit_date(self, type_, **kw):
1780 if self.dialect.server_version_info < MS_2008_VERSION:
1781 return self.visit_DATETIME(type_, **kw)
1782 else:
1783 return self.visit_DATE(type_, **kw)
1785 def visit__BASETIMEIMPL(self, type_, **kw):
1786 return self.visit_time(type_, **kw)
1788 def visit_time(self, type_, **kw):
1789 if self.dialect.server_version_info < MS_2008_VERSION:
1790 return self.visit_DATETIME(type_, **kw)
1791 else:
1792 return self.visit_TIME(type_, **kw)
1794 def visit_large_binary(self, type_, **kw):
1795 if self.dialect.deprecate_large_types:
1796 return self.visit_VARBINARY(type_, **kw)
1797 else:
1798 return self.visit_IMAGE(type_, **kw)
1800 def visit_IMAGE(self, type_, **kw):
1801 return "IMAGE"
1803 def visit_XML(self, type_, **kw):
1804 return "XML"
1806 def visit_VARBINARY(self, type_, **kw):
1807 text = self._extend("VARBINARY", type_, length=type_.length or "max")
1808 if getattr(type_, "filestream", False):
1809 text += " FILESTREAM"
1810 return text
1812 def visit_boolean(self, type_, **kw):
1813 return self.visit_BIT(type_)
1815 def visit_BIT(self, type_, **kw):
1816 return "BIT"
1818 def visit_JSON(self, type_, **kw):
1819 # this is a bit of a break with SQLAlchemy's convention of
1820 # "UPPERCASE name goes to UPPERCASE type name with no modification"
1821 return self._extend("NVARCHAR", type_, length="max")
1823 def visit_MONEY(self, type_, **kw):
1824 return "MONEY"
1826 def visit_SMALLMONEY(self, type_, **kw):
1827 return "SMALLMONEY"
1829 def visit_uuid(self, type_, **kw):
1830 if type_.native_uuid:
1831 return self.visit_UNIQUEIDENTIFIER(type_, **kw)
1832 else:
1833 return super().visit_uuid(type_, **kw)
1835 def visit_UNIQUEIDENTIFIER(self, type_, **kw):
1836 return "UNIQUEIDENTIFIER"
1838 def visit_SQL_VARIANT(self, type_, **kw):
1839 return "SQL_VARIANT"
1842class MSExecutionContext(default.DefaultExecutionContext):
1843 _embedded_scope_identity = False
1844 _enable_identity_insert = False
1845 _select_lastrowid = False
1846 _lastrowid = None
1848 dialect: MSDialect
1850 def _opt_encode(self, statement):
1851 if self.compiled and self.compiled.schema_translate_map:
1852 rst = self.compiled.preparer._render_schema_translates
1853 statement = rst(statement, self.compiled.schema_translate_map)
1855 return statement
1857 def pre_exec(self):
1858 """Activate IDENTITY_INSERT if needed."""
1860 if self.isinsert:
1861 if TYPE_CHECKING:
1862 assert is_sql_compiler(self.compiled)
1863 assert isinstance(self.compiled.compile_state, DMLState)
1864 assert isinstance(
1865 self.compiled.compile_state.dml_table, TableClause
1866 )
1868 tbl = self.compiled.compile_state.dml_table
1869 id_column = tbl._autoincrement_column
1871 if id_column is not None and (
1872 not isinstance(id_column.default, Sequence)
1873 ):
1874 insert_has_identity = True
1875 compile_state = self.compiled.dml_compile_state
1876 self._enable_identity_insert = (
1877 id_column.key in self.compiled_parameters[0]
1878 ) or (
1879 compile_state._dict_parameters
1880 and (id_column.key in compile_state._insert_col_keys)
1881 )
1883 else:
1884 insert_has_identity = False
1885 self._enable_identity_insert = False
1887 self._select_lastrowid = (
1888 not self.compiled.inline
1889 and insert_has_identity
1890 and not self.compiled.effective_returning
1891 and not self._enable_identity_insert
1892 and not self.executemany
1893 )
1895 if self._enable_identity_insert:
1896 self.root_connection._cursor_execute(
1897 self.cursor,
1898 self._opt_encode(
1899 "SET IDENTITY_INSERT %s ON"
1900 % self.identifier_preparer.format_table(tbl)
1901 ),
1902 (),
1903 self,
1904 )
1906 # don't embed the scope_identity select into an
1907 # "INSERT .. DEFAULT VALUES"
1908 if (
1909 self._select_lastrowid
1910 and self.dialect.scope_identity_must_be_embedded
1911 and self.dialect.use_scope_identity
1912 and len(self.parameters[0])
1913 ):
1914 self._embedded_scope_identity = True
1916 self.statement += "; select scope_identity()"
1918 def post_exec(self):
1920 conn = self.root_connection
1922 if self.isinsert or self.isupdate or self.isdelete:
1923 self._rowcount = self.cursor.rowcount
1925 # handle INSERT with embedded SELECT SCOPE_IDENTITY() call
1926 if self._embedded_scope_identity:
1927 # Fetch the last inserted id from the manipulated statement
1928 # We may have to skip over a number of result sets with
1929 # no data (due to triggers, etc.) so run up to three times
1931 row = None
1932 for _ in range(3):
1933 if self.cursor.description:
1934 rows = self.cursor.fetchall()
1935 if rows:
1936 row = rows[0]
1937 break
1938 else:
1939 self.cursor.nextset()
1941 self._lastrowid = int(row[0]) if row else None
1943 self.cursor_fetch_strategy = _cursor._NO_CURSOR_DML
1945 elif self._select_lastrowid:
1946 if self.dialect.use_scope_identity:
1947 conn._cursor_execute(
1948 self.cursor,
1949 "SELECT scope_identity() AS lastrowid",
1950 (),
1951 self,
1952 )
1953 else:
1954 conn._cursor_execute(
1955 self.cursor, "SELECT @@identity AS lastrowid", (), self
1956 )
1957 # fetchall() ensures the cursor is consumed without closing it
1958 row = self.cursor.fetchall()[0]
1959 self._lastrowid = int(row[0])
1961 self.cursor_fetch_strategy = _cursor._NO_CURSOR_DML
1962 elif (
1963 self.compiled is not None
1964 and is_sql_compiler(self.compiled)
1965 and self.compiled.effective_returning
1966 ):
1967 self.cursor_fetch_strategy = (
1968 _cursor.FullyBufferedCursorFetchStrategy(
1969 self.cursor,
1970 self.cursor.description,
1971 self.cursor.fetchall(),
1972 )
1973 )
1975 if self._enable_identity_insert:
1976 # Disable IDENTITY_INSERT if enabled.
1977 if TYPE_CHECKING:
1978 assert is_sql_compiler(self.compiled)
1979 assert isinstance(self.compiled.compile_state, DMLState)
1980 assert isinstance(
1981 self.compiled.compile_state.dml_table, TableClause
1982 )
1983 conn._cursor_execute(
1984 self.cursor,
1985 self._opt_encode(
1986 "SET IDENTITY_INSERT %s OFF"
1987 % self.identifier_preparer.format_table(
1988 self.compiled.compile_state.dml_table
1989 )
1990 ),
1991 (),
1992 self,
1993 )
1995 def get_lastrowid(self):
1996 return self._lastrowid
1998 def handle_dbapi_exception(self, e):
1999 if self._enable_identity_insert:
2000 try:
2001 self.cursor.execute(
2002 self._opt_encode(
2003 "SET IDENTITY_INSERT %s OFF"
2004 % self.identifier_preparer.format_table(
2005 self.compiled.compile_state.dml_table
2006 )
2007 )
2008 )
2009 except Exception:
2010 pass
2012 def fire_sequence(self, seq, type_):
2013 return self._execute_scalar(
2014 (
2015 "SELECT NEXT VALUE FOR %s"
2016 % self.identifier_preparer.format_sequence(seq)
2017 ),
2018 type_,
2019 )
2021 def get_insert_default(self, column):
2022 if (
2023 isinstance(column, sa_schema.Column)
2024 and column is column.table._autoincrement_column
2025 and isinstance(column.default, sa_schema.Sequence)
2026 and column.default.optional
2027 ):
2028 return None
2029 return super().get_insert_default(column)
2032class MSSQLCompiler(compiler.SQLCompiler):
2033 returning_precedes_values = True
2035 extract_map = util.update_copy(
2036 compiler.SQLCompiler.extract_map,
2037 {
2038 "doy": "dayofyear",
2039 "dow": "weekday",
2040 "milliseconds": "millisecond",
2041 "microseconds": "microsecond",
2042 },
2043 )
2045 def __init__(self, *args, **kwargs):
2046 self.tablealiases = {}
2047 super().__init__(*args, **kwargs)
2049 def visit_frame_clause(self, frameclause, **kw):
2050 kw["literal_execute"] = True
2051 return super().visit_frame_clause(frameclause, **kw)
2053 def _with_legacy_schema_aliasing(fn):
2054 def decorate(self, *arg, **kw):
2055 if self.dialect.legacy_schema_aliasing:
2056 return fn(self, *arg, **kw)
2057 else:
2058 super_ = getattr(super(MSSQLCompiler, self), fn.__name__)
2059 return super_(*arg, **kw)
2061 return decorate
2063 def visit_now_func(self, fn, **kw):
2064 return "CURRENT_TIMESTAMP"
2066 def visit_current_date_func(self, fn, **kw):
2067 return "GETDATE()"
2069 def visit_length_func(self, fn, **kw):
2070 return "LEN%s" % self.function_argspec(fn, **kw)
2072 def visit_char_length_func(self, fn, **kw):
2073 return "LEN%s" % self.function_argspec(fn, **kw)
2075 def visit_aggregate_strings_func(self, fn, **kw):
2076 cl = list(fn.clauses)
2077 expr, delimiter = cl[0:2]
2079 literal_exec = dict(kw)
2080 literal_exec["literal_execute"] = True
2082 return (
2083 f"string_agg({expr._compiler_dispatch(self, **kw)}, "
2084 f"{delimiter._compiler_dispatch(self, **literal_exec)})"
2085 )
2087 def visit_pow_func(self, fn, **kw):
2088 return f"POWER{self.function_argspec(fn)}"
2090 def visit_concat_op_expression_clauselist(
2091 self, clauselist, operator, **kw
2092 ):
2093 return " + ".join(self.process(elem, **kw) for elem in clauselist)
2095 def visit_concat_op_binary(self, binary, operator, **kw):
2096 return "%s + %s" % (
2097 self.process(binary.left, **kw),
2098 self.process(binary.right, **kw),
2099 )
2101 def visit_truediv_binary(self, binary, operator, **kw):
2102 return (
2103 self.process(binary.left, **kw)
2104 + " / "
2105 + f"({self.process(binary.right, **kw)} + 0.0)"
2106 )
2108 def visit_true(self, expr, **kw):
2109 return "1"
2111 def visit_false(self, expr, **kw):
2112 return "0"
2114 def visit_match_op_binary(self, binary, operator, **kw):
2115 return "CONTAINS (%s, %s)" % (
2116 self.process(binary.left, **kw),
2117 self.process(binary.right, **kw),
2118 )
2120 def get_select_precolumns(self, select, **kw):
2121 """MS-SQL puts TOP, it's version of LIMIT here"""
2123 s = super().get_select_precolumns(select, **kw)
2125 if select._has_row_limiting_clause and self._use_top(select):
2126 # ODBC drivers and possibly others
2127 # don't support bind params in the SELECT clause on SQL Server.
2128 # so have to use literal here.
2129 kw["literal_execute"] = True
2130 s += "TOP %s " % self.process(
2131 self._get_limit_or_fetch(select), **kw
2132 )
2133 if select._fetch_clause is not None:
2134 if select._fetch_clause_options["percent"]:
2135 s += "PERCENT "
2136 if select._fetch_clause_options["with_ties"]:
2137 s += "WITH TIES "
2139 return s
2141 def get_from_hint_text(self, table, text):
2142 return text
2144 def get_crud_hint_text(self, table, text):
2145 return text
2147 def _get_limit_or_fetch(self, select):
2148 if select._fetch_clause is None:
2149 return select._limit_clause
2150 else:
2151 return select._fetch_clause
2153 def _use_top(self, select):
2154 return (select._offset_clause is None) and (
2155 select._simple_int_clause(select._limit_clause)
2156 or (
2157 # limit can use TOP with is by itself. fetch only uses TOP
2158 # when it needs to because of PERCENT and/or WITH TIES
2159 # TODO: Why? shouldn't we use TOP always ?
2160 select._simple_int_clause(select._fetch_clause)
2161 and (
2162 select._fetch_clause_options["percent"]
2163 or select._fetch_clause_options["with_ties"]
2164 )
2165 )
2166 )
2168 def limit_clause(self, cs, **kwargs):
2169 return ""
2171 def _check_can_use_fetch_limit(self, select):
2172 # to use ROW_NUMBER(), an ORDER BY is required.
2173 # OFFSET are FETCH are options of the ORDER BY clause
2174 if not select._order_by_clause.clauses:
2175 raise exc.CompileError(
2176 "MSSQL requires an order_by when "
2177 "using an OFFSET or a non-simple "
2178 "LIMIT clause"
2179 )
2181 if select._fetch_clause_options is not None and (
2182 select._fetch_clause_options["percent"]
2183 or select._fetch_clause_options["with_ties"]
2184 ):
2185 raise exc.CompileError(
2186 "MSSQL needs TOP to use PERCENT and/or WITH TIES. "
2187 "Only simple fetch without offset can be used."
2188 )
2190 def _row_limit_clause(self, select, **kw):
2191 """MSSQL 2012 supports OFFSET/FETCH operators
2192 Use it instead subquery with row_number
2194 """
2196 if self.dialect._supports_offset_fetch and not self._use_top(select):
2197 self._check_can_use_fetch_limit(select)
2199 return self.fetch_clause(
2200 select,
2201 fetch_clause=self._get_limit_or_fetch(select),
2202 require_offset=True,
2203 **kw,
2204 )
2206 else:
2207 return ""
2209 def visit_try_cast(self, element, **kw):
2210 return "TRY_CAST (%s AS %s)" % (
2211 self.process(element.clause, **kw),
2212 self.process(element.typeclause, **kw),
2213 )
2215 def translate_select_structure(self, select_stmt, **kwargs):
2216 """Look for ``LIMIT`` and OFFSET in a select statement, and if
2217 so tries to wrap it in a subquery with ``row_number()`` criterion.
2218 MSSQL 2012 and above are excluded
2220 """
2221 select = select_stmt
2223 if (
2224 select._has_row_limiting_clause
2225 and not self.dialect._supports_offset_fetch
2226 and not self._use_top(select)
2227 and not getattr(select, "_mssql_visit", None)
2228 ):
2229 self._check_can_use_fetch_limit(select)
2231 _order_by_clauses = [
2232 sql_util.unwrap_label_reference(elem)
2233 for elem in select._order_by_clause.clauses
2234 ]
2236 limit_clause = self._get_limit_or_fetch(select)
2237 offset_clause = select._offset_clause
2239 select = select._generate()
2240 select._mssql_visit = True
2241 select = (
2242 select.add_columns(
2243 sql.func.ROW_NUMBER()
2244 .over(order_by=_order_by_clauses)
2245 .label("mssql_rn")
2246 )
2247 .order_by(None)
2248 .alias()
2249 )
2251 mssql_rn = sql.column("mssql_rn")
2252 limitselect = sql.select(
2253 *[c for c in select.c if c.key != "mssql_rn"]
2254 )
2255 if offset_clause is not None:
2256 limitselect = limitselect.where(mssql_rn > offset_clause)
2257 if limit_clause is not None:
2258 limitselect = limitselect.where(
2259 mssql_rn <= (limit_clause + offset_clause)
2260 )
2261 else:
2262 limitselect = limitselect.where(mssql_rn <= (limit_clause))
2263 return limitselect
2264 else:
2265 return select
2267 @_with_legacy_schema_aliasing
2268 def visit_table(self, table, mssql_aliased=False, iscrud=False, **kwargs):
2269 if mssql_aliased is table or iscrud:
2270 return super().visit_table(table, **kwargs)
2272 # alias schema-qualified tables
2273 alias = self._schema_aliased_table(table)
2274 if alias is not None:
2275 return self.process(alias, mssql_aliased=table, **kwargs)
2276 else:
2277 return super().visit_table(table, **kwargs)
2279 @_with_legacy_schema_aliasing
2280 def visit_alias(self, alias, **kw):
2281 # translate for schema-qualified table aliases
2282 kw["mssql_aliased"] = alias.element
2283 return super().visit_alias(alias, **kw)
2285 @_with_legacy_schema_aliasing
2286 def visit_column(self, column, add_to_result_map=None, **kw):
2287 if (
2288 column.table is not None
2289 and (not self.isupdate and not self.isdelete)
2290 or self.is_subquery()
2291 ):
2292 # translate for schema-qualified table aliases
2293 t = self._schema_aliased_table(column.table)
2294 if t is not None:
2295 converted = elements._corresponding_column_or_error(t, column)
2296 if add_to_result_map is not None:
2297 add_to_result_map(
2298 column.name,
2299 column.name,
2300 (column, column.name, column.key),
2301 column.type,
2302 )
2304 return super().visit_column(converted, **kw)
2306 return super().visit_column(
2307 column, add_to_result_map=add_to_result_map, **kw
2308 )
2310 def _schema_aliased_table(self, table):
2311 if getattr(table, "schema", None) is not None:
2312 if table not in self.tablealiases:
2313 self.tablealiases[table] = table.alias()
2314 return self.tablealiases[table]
2315 else:
2316 return None
2318 def visit_extract(self, extract, **kw):
2319 field = self.extract_map.get(extract.field, extract.field)
2320 return "DATEPART(%s, %s)" % (field, self.process(extract.expr, **kw))
2322 def visit_savepoint(self, savepoint_stmt, **kw):
2323 return "SAVE TRANSACTION %s" % self.preparer.format_savepoint(
2324 savepoint_stmt
2325 )
2327 def visit_rollback_to_savepoint(self, savepoint_stmt, **kw):
2328 return "ROLLBACK TRANSACTION %s" % self.preparer.format_savepoint(
2329 savepoint_stmt
2330 )
2332 def visit_binary(self, binary, **kwargs):
2333 """Move bind parameters to the right-hand side of an operator, where
2334 possible.
2336 """
2337 if (
2338 isinstance(binary.left, expression.BindParameter)
2339 and binary.operator == operator.eq
2340 and not isinstance(binary.right, expression.BindParameter)
2341 ):
2342 return self.process(
2343 expression.BinaryExpression(
2344 binary.right, binary.left, binary.operator
2345 ),
2346 **kwargs,
2347 )
2348 return super().visit_binary(binary, **kwargs)
2350 def returning_clause(
2351 self, stmt, returning_cols, *, populate_result_map, **kw
2352 ):
2353 # SQL server returning clause requires that the columns refer to
2354 # the virtual table names "inserted" or "deleted". Here, we make
2355 # a simple alias of our table with that name, and then adapt the
2356 # columns we have from the list of RETURNING columns to that new name
2357 # so that they render as "inserted.<colname>" / "deleted.<colname>".
2359 if stmt.is_insert or stmt.is_update:
2360 target = stmt.table.alias("inserted")
2361 elif stmt.is_delete:
2362 target = stmt.table.alias("deleted")
2363 else:
2364 assert False, "expected Insert, Update or Delete statement"
2366 adapter = sql_util.ClauseAdapter(target)
2368 # adapter.traverse() takes a column from our target table and returns
2369 # the one that is linked to the "inserted" / "deleted" tables. So in
2370 # order to retrieve these values back from the result (e.g. like
2371 # row[column]), tell the compiler to also add the original unadapted
2372 # column to the result map. Before #4877, these were (unknowingly)
2373 # falling back using string name matching in the result set which
2374 # necessarily used an expensive KeyError in order to match.
2376 columns = [
2377 self._label_returning_column(
2378 stmt,
2379 adapter.traverse(column),
2380 populate_result_map,
2381 {"result_map_targets": (column,)},
2382 fallback_label_name=fallback_label_name,
2383 column_is_repeated=repeated,
2384 name=name,
2385 proxy_name=proxy_name,
2386 **kw,
2387 )
2388 for (
2389 name,
2390 proxy_name,
2391 fallback_label_name,
2392 column,
2393 repeated,
2394 ) in stmt._generate_columns_plus_names(
2395 True, cols=expression._select_iterables(returning_cols)
2396 )
2397 ]
2399 return "OUTPUT " + ", ".join(columns)
2401 def get_cte_preamble(self, recursive):
2402 # SQL Server finds it too inconvenient to accept
2403 # an entirely optional, SQL standard specified,
2404 # "RECURSIVE" word with their "WITH",
2405 # so here we go
2406 return "WITH"
2408 def label_select_column(self, select, column, asfrom):
2409 if isinstance(column, expression.Function):
2410 return column.label(None)
2411 else:
2412 return super().label_select_column(select, column, asfrom)
2414 def for_update_clause(self, select, **kw):
2415 # "FOR UPDATE" is only allowed on "DECLARE CURSOR" which
2416 # SQLAlchemy doesn't use
2417 return ""
2419 def order_by_clause(self, select, **kw):
2420 # MSSQL only allows ORDER BY in subqueries if there is a LIMIT:
2421 # "The ORDER BY clause is invalid in views, inline functions,
2422 # derived tables, subqueries, and common table expressions,
2423 # unless TOP, OFFSET or FOR XML is also specified."
2424 if (
2425 self.is_subquery()
2426 and not self._use_top(select)
2427 and (
2428 select._offset is None
2429 or not self.dialect._supports_offset_fetch
2430 )
2431 ):
2432 # avoid processing the order by clause if we won't end up
2433 # using it, because we don't want all the bind params tacked
2434 # onto the positional list if that is what the dbapi requires
2435 return ""
2437 order_by = self.process(select._order_by_clause, **kw)
2439 if order_by:
2440 return " ORDER BY " + order_by
2441 else:
2442 return ""
2444 def update_from_clause(
2445 self, update_stmt, from_table, extra_froms, from_hints, **kw
2446 ):
2447 """Render the UPDATE..FROM clause specific to MSSQL.
2449 In MSSQL, if the UPDATE statement involves an alias of the table to
2450 be updated, then the table itself must be added to the FROM list as
2451 well. Otherwise, it is optional. Here, we add it regardless.
2453 """
2454 return "FROM " + ", ".join(
2455 t._compiler_dispatch(self, asfrom=True, fromhints=from_hints, **kw)
2456 for t in [from_table] + extra_froms
2457 )
2459 def delete_table_clause(self, delete_stmt, from_table, extra_froms, **kw):
2460 """If we have extra froms make sure we render any alias as hint."""
2461 ashint = False
2462 if extra_froms:
2463 ashint = True
2464 return from_table._compiler_dispatch(
2465 self, asfrom=True, iscrud=True, ashint=ashint, **kw
2466 )
2468 def delete_extra_from_clause(
2469 self, delete_stmt, from_table, extra_froms, from_hints, **kw
2470 ):
2471 """Render the DELETE .. FROM clause specific to MSSQL.
2473 Yes, it has the FROM keyword twice.
2475 """
2476 return "FROM " + ", ".join(
2477 t._compiler_dispatch(self, asfrom=True, fromhints=from_hints, **kw)
2478 for t in [from_table] + extra_froms
2479 )
2481 def visit_empty_set_expr(self, type_, **kw):
2482 return "SELECT 1 WHERE 1!=1"
2484 def visit_is_distinct_from_binary(self, binary, operator, **kw):
2485 return "NOT EXISTS (SELECT %s INTERSECT SELECT %s)" % (
2486 self.process(binary.left),
2487 self.process(binary.right),
2488 )
2490 def visit_is_not_distinct_from_binary(self, binary, operator, **kw):
2491 return "EXISTS (SELECT %s INTERSECT SELECT %s)" % (
2492 self.process(binary.left),
2493 self.process(binary.right),
2494 )
2496 def _render_json_extract_from_binary(self, binary, operator, **kw):
2497 # note we are intentionally calling upon the process() calls in the
2498 # order in which they appear in the SQL String as this is used
2499 # by positional parameter rendering
2501 if binary.type._type_affinity is sqltypes.JSON:
2502 return "JSON_QUERY(%s, %s)" % (
2503 self.process(binary.left, **kw),
2504 self.process(binary.right, **kw),
2505 )
2507 # as with other dialects, start with an explicit test for NULL
2508 case_expression = "CASE JSON_VALUE(%s, %s) WHEN NULL THEN NULL" % (
2509 self.process(binary.left, **kw),
2510 self.process(binary.right, **kw),
2511 )
2513 if binary.type._type_affinity is sqltypes.Integer:
2514 type_expression = "ELSE CAST(JSON_VALUE(%s, %s) AS INTEGER)" % (
2515 self.process(binary.left, **kw),
2516 self.process(binary.right, **kw),
2517 )
2518 elif binary.type._type_affinity in (sqltypes.Numeric, sqltypes.Float):
2519 type_expression = "ELSE CAST(JSON_VALUE(%s, %s) AS %s)" % (
2520 self.process(binary.left, **kw),
2521 self.process(binary.right, **kw),
2522 (
2523 "FLOAT"
2524 if isinstance(binary.type, sqltypes.Float)
2525 else "NUMERIC(%s, %s)"
2526 % (binary.type.precision, binary.type.scale)
2527 ),
2528 )
2529 elif binary.type._type_affinity is sqltypes.Boolean:
2530 # the NULL handling is particularly weird with boolean, so
2531 # explicitly return numeric (BIT) constants
2532 type_expression = (
2533 "WHEN 'true' THEN 1 WHEN 'false' THEN 0 ELSE "
2534 "CAST(JSON_VALUE(%s, %s) AS BIT)"
2535 % (
2536 self.process(binary.left, **kw),
2537 self.process(binary.right, **kw),
2538 )
2539 )
2540 elif binary.type._type_affinity is sqltypes.String:
2541 # TODO: does this comment (from mysql) apply to here, too?
2542 # this fails with a JSON value that's a four byte unicode
2543 # string. SQLite has the same problem at the moment
2544 type_expression = "ELSE JSON_VALUE(%s, %s)" % (
2545 self.process(binary.left, **kw),
2546 self.process(binary.right, **kw),
2547 )
2548 else:
2549 # other affinity....this is not expected right now
2550 type_expression = "ELSE JSON_QUERY(%s, %s)" % (
2551 self.process(binary.left, **kw),
2552 self.process(binary.right, **kw),
2553 )
2555 return case_expression + " " + type_expression + " END"
2557 def visit_json_getitem_op_binary(self, binary, operator, **kw):
2558 return self._render_json_extract_from_binary(binary, operator, **kw)
2560 def visit_json_path_getitem_op_binary(self, binary, operator, **kw):
2561 return self._render_json_extract_from_binary(binary, operator, **kw)
2563 def visit_sequence(self, seq, **kw):
2564 return "NEXT VALUE FOR %s" % self.preparer.format_sequence(seq)
2567class MSSQLStrictCompiler(MSSQLCompiler):
2568 """A subclass of MSSQLCompiler which disables the usage of bind
2569 parameters where not allowed natively by MS-SQL.
2571 A dialect may use this compiler on a platform where native
2572 binds are used.
2574 """
2576 ansi_bind_rules = True
2578 def visit_in_op_binary(self, binary, operator, **kw):
2579 kw["literal_execute"] = True
2580 return "%s IN %s" % (
2581 self.process(binary.left, **kw),
2582 self.process(binary.right, **kw),
2583 )
2585 def visit_not_in_op_binary(self, binary, operator, **kw):
2586 kw["literal_execute"] = True
2587 return "%s NOT IN %s" % (
2588 self.process(binary.left, **kw),
2589 self.process(binary.right, **kw),
2590 )
2592 def render_literal_value(self, value, type_):
2593 """
2594 For date and datetime values, convert to a string
2595 format acceptable to MSSQL. That seems to be the
2596 so-called ODBC canonical date format which looks
2597 like this:
2599 yyyy-mm-dd hh:mi:ss.mmm(24h)
2601 For other data types, call the base class implementation.
2602 """
2603 # datetime and date are both subclasses of datetime.date
2604 if issubclass(type(value), datetime.date):
2605 # SQL Server wants single quotes around the date string.
2606 return "'" + str(value) + "'"
2607 else:
2608 return super().render_literal_value(value, type_)
2611class MSDDLCompiler(compiler.DDLCompiler):
2612 def get_column_specification(self, column, **kwargs):
2613 colspec = self.preparer.format_column(column)
2615 # type is not accepted in a computed column
2616 if column.computed is not None:
2617 colspec += " " + self.process(column.computed)
2618 else:
2619 colspec += " " + self.dialect.type_compiler_instance.process(
2620 column.type, type_expression=column
2621 )
2623 if column.nullable is not None:
2624 if (
2625 not column.nullable
2626 or column.primary_key
2627 or isinstance(column.default, sa_schema.Sequence)
2628 or column.autoincrement is True
2629 or column.identity
2630 ):
2631 colspec += " NOT NULL"
2632 elif column.computed is None:
2633 # don't specify "NULL" for computed columns
2634 colspec += " NULL"
2636 if column.table is None:
2637 raise exc.CompileError(
2638 "mssql requires Table-bound columns "
2639 "in order to generate DDL"
2640 )
2642 d_opt = column.dialect_options["mssql"]
2643 start = d_opt["identity_start"]
2644 increment = d_opt["identity_increment"]
2645 if start is not None or increment is not None:
2646 if column.identity:
2647 raise exc.CompileError(
2648 "Cannot specify options 'mssql_identity_start' and/or "
2649 "'mssql_identity_increment' while also using the "
2650 "'Identity' construct."
2651 )
2652 util.warn_deprecated(
2653 "The dialect options 'mssql_identity_start' and "
2654 "'mssql_identity_increment' are deprecated. "
2655 "Use the 'Identity' object instead.",
2656 "1.4",
2657 )
2659 if column.identity:
2660 colspec += self.process(column.identity, **kwargs)
2661 elif (
2662 column is column.table._autoincrement_column
2663 or column.autoincrement is True
2664 ) and (
2665 not isinstance(column.default, Sequence) or column.default.optional
2666 ):
2667 colspec += self.process(Identity(start=start, increment=increment))
2668 else:
2669 default = self.get_column_default_string(column)
2670 if default is not None:
2671 colspec += " DEFAULT " + default
2673 return colspec
2675 def visit_create_index(self, create, include_schema=False, **kw):
2676 index = create.element
2677 self._verify_index_table(index)
2678 preparer = self.preparer
2679 text = "CREATE "
2680 if index.unique:
2681 text += "UNIQUE "
2683 # handle clustering option
2684 clustered = index.dialect_options["mssql"]["clustered"]
2685 if clustered is not None:
2686 if clustered:
2687 text += "CLUSTERED "
2688 else:
2689 text += "NONCLUSTERED "
2691 # handle columnstore option (has no negative value)
2692 columnstore = index.dialect_options["mssql"]["columnstore"]
2693 if columnstore:
2694 text += "COLUMNSTORE "
2696 text += "INDEX %s ON %s" % (
2697 self._prepared_index_name(index, include_schema=include_schema),
2698 preparer.format_table(index.table),
2699 )
2701 # in some case mssql allows indexes with no columns defined
2702 if len(index.expressions) > 0:
2703 text += " (%s)" % ", ".join(
2704 self.sql_compiler.process(
2705 expr, include_table=False, literal_binds=True
2706 )
2707 for expr in index.expressions
2708 )
2710 # handle other included columns
2711 if index.dialect_options["mssql"]["include"]:
2712 inclusions = [
2713 index.table.c[col] if isinstance(col, str) else col
2714 for col in index.dialect_options["mssql"]["include"]
2715 ]
2717 text += " INCLUDE (%s)" % ", ".join(
2718 [preparer.quote(c.name) for c in inclusions]
2719 )
2721 whereclause = index.dialect_options["mssql"]["where"]
2723 if whereclause is not None:
2724 whereclause = coercions.expect(
2725 roles.DDLExpressionRole, whereclause
2726 )
2728 where_compiled = self.sql_compiler.process(
2729 whereclause, include_table=False, literal_binds=True
2730 )
2731 text += " WHERE " + where_compiled
2733 return text
2735 def visit_drop_index(self, drop: DropIndex, **kw: Any) -> str:
2736 index_name = self._prepared_index_name(
2737 drop.element, include_schema=False
2738 )
2739 table_name = self.preparer.format_table(drop.element.table)
2740 if_exists = " IF EXISTS" if drop.if_exists else ""
2741 return f"\nDROP INDEX{if_exists} {index_name} ON {table_name}"
2743 def visit_create_table_as(self, element, **kw):
2744 prep = self.preparer
2746 # SQL Server doesn't support CREATE TABLE AS, use SELECT INTO instead
2747 # Format: SELECT columns INTO new_table FROM source WHERE ...
2749 qualified = prep.format_table(element.table)
2751 # Get the inner SELECT SQL
2752 inner_kw = dict(kw)
2753 inner_kw["literal_binds"] = True
2754 select_sql = self.sql_compiler.process(element.selectable, **inner_kw)
2756 # Inject INTO clause before FROM keyword
2757 # Find FROM position (case-insensitive)
2758 select_upper = select_sql.upper()
2759 from_idx = select_upper.find(" FROM ")
2760 if from_idx == -1:
2761 from_idx = select_upper.find("\nFROM ")
2763 if from_idx == -1:
2764 raise exc.CompileError(
2765 "Could not find FROM keyword in selectable for CREATE TABLE AS"
2766 )
2768 # Insert INTO clause before FROM
2769 result = (
2770 select_sql[:from_idx]
2771 + f"INTO {qualified} "
2772 + select_sql[from_idx:]
2773 )
2775 return result
2777 def visit_create_view(self, create, **kw):
2778 # SQL Server uses CREATE OR ALTER instead of CREATE OR REPLACE
2779 result = super().visit_create_view(create, **kw)
2780 if create.or_replace:
2781 result = result.replace("CREATE OR REPLACE", "CREATE OR ALTER")
2782 return result
2784 def visit_primary_key_constraint(self, constraint, **kw):
2785 if len(constraint) == 0:
2786 return ""
2787 text = ""
2788 if constraint.name is not None:
2789 text += "CONSTRAINT %s " % self.preparer.format_constraint(
2790 constraint
2791 )
2792 text += "PRIMARY KEY "
2794 clustered = constraint.dialect_options["mssql"]["clustered"]
2795 if clustered is not None:
2796 if clustered:
2797 text += "CLUSTERED "
2798 else:
2799 text += "NONCLUSTERED "
2801 text += "(%s)" % ", ".join(
2802 self.preparer.quote(c.name) for c in constraint
2803 )
2804 text += self.define_constraint_deferrability(constraint)
2805 return text
2807 def visit_unique_constraint(self, constraint, **kw):
2808 if len(constraint) == 0:
2809 return ""
2810 text = ""
2811 if constraint.name is not None:
2812 formatted_name = self.preparer.format_constraint(constraint)
2813 if formatted_name is not None:
2814 text += "CONSTRAINT %s " % formatted_name
2815 text += "UNIQUE %s" % self.define_unique_constraint_distinct(
2816 constraint, **kw
2817 )
2818 clustered = constraint.dialect_options["mssql"]["clustered"]
2819 if clustered is not None:
2820 if clustered:
2821 text += "CLUSTERED "
2822 else:
2823 text += "NONCLUSTERED "
2825 text += "(%s)" % ", ".join(
2826 self.preparer.quote(c.name) for c in constraint
2827 )
2828 text += self.define_constraint_deferrability(constraint)
2829 return text
2831 def visit_computed_column(self, generated, **kw):
2832 text = "AS (%s)" % self.sql_compiler.process(
2833 generated.sqltext, include_table=False, literal_binds=True
2834 )
2835 # explicitly check for True|False since None means server default
2836 if generated.persisted is True:
2837 text += " PERSISTED"
2838 return text
2840 def visit_set_table_comment(self, create, **kw):
2841 schema = self.preparer.schema_for_object(create.element)
2842 schema_name = schema if schema else self.dialect.default_schema_name
2843 return (
2844 "execute sp_addextendedproperty 'MS_Description', "
2845 "{}, 'schema', {}, 'table', {}".format(
2846 self.sql_compiler.render_literal_value(
2847 create.element.comment, sqltypes.NVARCHAR()
2848 ),
2849 self.preparer.quote_schema(schema_name),
2850 self.preparer.format_table(create.element, use_schema=False),
2851 )
2852 )
2854 def visit_drop_table_comment(self, drop, **kw):
2855 schema = self.preparer.schema_for_object(drop.element)
2856 schema_name = schema if schema else self.dialect.default_schema_name
2857 return (
2858 "execute sp_dropextendedproperty 'MS_Description', 'schema', "
2859 "{}, 'table', {}".format(
2860 self.preparer.quote_schema(schema_name),
2861 self.preparer.format_table(drop.element, use_schema=False),
2862 )
2863 )
2865 def visit_set_column_comment(self, create, **kw):
2866 schema = self.preparer.schema_for_object(create.element.table)
2867 schema_name = schema if schema else self.dialect.default_schema_name
2868 return (
2869 "execute sp_addextendedproperty 'MS_Description', "
2870 "{}, 'schema', {}, 'table', {}, 'column', {}".format(
2871 self.sql_compiler.render_literal_value(
2872 create.element.comment, sqltypes.NVARCHAR()
2873 ),
2874 self.preparer.quote_schema(schema_name),
2875 self.preparer.format_table(
2876 create.element.table, use_schema=False
2877 ),
2878 self.preparer.format_column(create.element),
2879 )
2880 )
2882 def visit_drop_column_comment(self, drop, **kw):
2883 schema = self.preparer.schema_for_object(drop.element.table)
2884 schema_name = schema if schema else self.dialect.default_schema_name
2885 return (
2886 "execute sp_dropextendedproperty 'MS_Description', 'schema', "
2887 "{}, 'table', {}, 'column', {}".format(
2888 self.preparer.quote_schema(schema_name),
2889 self.preparer.format_table(
2890 drop.element.table, use_schema=False
2891 ),
2892 self.preparer.format_column(drop.element),
2893 )
2894 )
2896 def visit_create_sequence(self, create, **kw):
2897 prefix = None
2898 if create.element.data_type is not None:
2899 data_type = create.element.data_type
2900 prefix = " AS %s" % self.type_compiler.process(data_type)
2901 return super().visit_create_sequence(create, prefix=prefix, **kw)
2903 def visit_identity_column(self, identity, **kw):
2904 text = " IDENTITY"
2905 if identity.start is not None or identity.increment is not None:
2906 start = 1 if identity.start is None else identity.start
2907 increment = 1 if identity.increment is None else identity.increment
2908 text += "(%s,%s)" % (start, increment)
2909 return text
2912class MSIdentifierPreparer(compiler.IdentifierPreparer):
2913 reserved_words = RESERVED_WORDS
2915 def __init__(self, dialect):
2916 super().__init__(
2917 dialect,
2918 initial_quote="[",
2919 final_quote="]",
2920 quote_case_sensitive_collations=False,
2921 )
2923 def _escape_identifier(self, value):
2924 return value.replace("]", "]]")
2926 def _unescape_identifier(self, value):
2927 return value.replace("]]", "]")
2929 def quote_schema(self, schema):
2930 """Prepare a quoted table and schema name."""
2932 dbname, owner = _schema_elements(schema)
2933 if dbname:
2934 result = "%s.%s" % (self.quote(dbname), self.quote(owner))
2935 elif owner:
2936 result = self.quote(owner)
2937 else:
2938 result = ""
2939 return result
2942def _db_plus_owner_listing(fn):
2943 def wrap(dialect, connection, schema=None, **kw):
2944 dbname, owner = _owner_plus_db(dialect, schema)
2945 return _switch_db(
2946 dbname,
2947 connection,
2948 fn,
2949 dialect,
2950 connection,
2951 dbname,
2952 owner,
2953 schema,
2954 **kw,
2955 )
2957 return update_wrapper(wrap, fn)
2960def _db_plus_owner(fn):
2961 def wrap(dialect, connection, tablename, schema=None, **kw):
2962 dbname, owner = _owner_plus_db(dialect, schema)
2963 return _switch_db(
2964 dbname,
2965 connection,
2966 fn,
2967 dialect,
2968 connection,
2969 tablename,
2970 dbname,
2971 owner,
2972 schema,
2973 **kw,
2974 )
2976 return update_wrapper(wrap, fn)
2979def _db_plus_owner_multi(fn):
2980 def wrap(dialect, connection, schema=None, **kw):
2981 dbname, owner = _owner_plus_db(dialect, schema)
2982 return _switch_db(
2983 dbname,
2984 connection,
2985 fn,
2986 dialect,
2987 connection,
2988 dbname,
2989 owner,
2990 schema,
2991 **kw,
2992 )
2994 return update_wrapper(wrap, fn)
2997def _switch_db(dbname, connection, fn, *arg, **kw):
2998 if dbname:
2999 current_db = connection.exec_driver_sql("select db_name()").scalar()
3000 if current_db != dbname:
3001 connection.exec_driver_sql(
3002 "use %s" % connection.dialect.identifier_preparer.quote(dbname)
3003 )
3004 try:
3005 return fn(*arg, **kw)
3006 finally:
3007 if dbname and current_db != dbname:
3008 connection.exec_driver_sql(
3009 "use %s"
3010 % connection.dialect.identifier_preparer.quote(current_db)
3011 )
3014def _owner_plus_db(dialect, schema):
3015 if not schema:
3016 return None, dialect.default_schema_name
3017 else:
3018 return _schema_elements(schema)
3021_memoized_schema = util.LRUCache()
3024def _schema_elements(schema):
3025 if isinstance(schema, quoted_name) and schema.quote:
3026 return None, schema
3028 if schema in _memoized_schema:
3029 return _memoized_schema[schema]
3031 # tests for this function are in:
3032 # test/dialect/mssql/test_reflection.py ->
3033 # OwnerPlusDBTest.test_owner_database_pairs
3034 # test/dialect/mssql/test_compiler.py -> test_force_schema_*
3035 # test/dialect/mssql/test_compiler.py -> test_schema_many_tokens_*
3036 #
3038 if schema.startswith("__[SCHEMA_"):
3039 return None, schema
3041 push = []
3042 symbol = ""
3043 bracket = False
3044 has_brackets = False
3045 for token in re.split(r"(\[|\]|\.)", schema):
3046 if not token:
3047 continue
3048 if token == "[":
3049 bracket = True
3050 has_brackets = True
3051 elif token == "]":
3052 bracket = False
3053 elif not bracket and token == ".":
3054 if has_brackets:
3055 push.append("[%s]" % symbol)
3056 else:
3057 push.append(symbol)
3058 symbol = ""
3059 has_brackets = False
3060 else:
3061 symbol += token
3062 if symbol:
3063 push.append(symbol)
3064 if len(push) > 1:
3065 dbname, owner = ".".join(push[0:-1]), push[-1]
3067 # test for internal brackets
3068 if re.match(r".*\].*\[.*", dbname[1:-1]):
3069 dbname = quoted_name(dbname, quote=False)
3070 else:
3071 dbname = dbname.lstrip("[").rstrip("]")
3073 elif len(push):
3074 dbname, owner = None, push[0]
3075 else:
3076 dbname, owner = None, None
3078 _memoized_schema[schema] = dbname, owner
3079 return dbname, owner
3082class MSDialect(default._BackendsMultiReflection, default.DefaultDialect):
3083 # will assume it's at least mssql2005
3084 name = "mssql"
3085 supports_statement_cache = True
3086 supports_default_values = True
3087 supports_empty_insert = False
3088 favor_returning_over_lastrowid = True
3089 scope_identity_must_be_embedded = False
3091 returns_native_bytes = True
3093 supports_comments = True
3094 supports_default_metavalue = False
3095 """dialect supports INSERT... VALUES (DEFAULT) syntax -
3096 SQL Server **does** support this, but **not** for the IDENTITY column,
3097 so we can't turn this on.
3099 """
3101 aggregate_order_by_style = AggregateOrderByStyle.WITHIN_GROUP
3103 # supports_native_uuid is partial here, so we implement our
3104 # own impl type
3106 execution_ctx_cls = MSExecutionContext
3107 use_scope_identity = True
3108 max_identifier_length = 128
3109 schema_name = "dbo"
3111 insert_returning = True
3112 update_returning = True
3113 delete_returning = True
3114 update_returning_multifrom = True
3115 delete_returning_multifrom = True
3117 colspecs = {
3118 sqltypes.DateTime: _MSDateTime,
3119 sqltypes.Date: _MSDate,
3120 sqltypes.JSON: JSON,
3121 sqltypes.JSON.JSONIndexType: JSONIndexType,
3122 sqltypes.JSON.JSONPathType: JSONPathType,
3123 sqltypes.Time: _BASETIMEIMPL,
3124 sqltypes.Unicode: _MSUnicode,
3125 sqltypes.UnicodeText: _MSUnicodeText,
3126 DATETIMEOFFSET: DATETIMEOFFSET,
3127 DATETIME2: DATETIME2,
3128 SMALLDATETIME: SMALLDATETIME,
3129 DATETIME: DATETIME,
3130 sqltypes.Uuid: MSUUid,
3131 }
3133 engine_config_types = default.DefaultDialect.engine_config_types.union(
3134 {"legacy_schema_aliasing": util.asbool}
3135 )
3137 ischema_names = ischema_names
3139 supports_sequences = True
3140 sequences_optional = True
3141 # This is actually used for autoincrement, where itentity is used that
3142 # starts with 1.
3143 # for sequences T-SQL's actual default is -9223372036854775808
3144 default_sequence_base = 1
3146 supports_native_boolean = False
3147 non_native_boolean_check_constraint = False
3148 supports_unicode_binds = True
3149 postfetch_lastrowid = True
3151 # may be changed at server inspection time for older SQL server versions
3152 supports_multivalues_insert = True
3154 use_insertmanyvalues = True
3156 # note pyodbc will set this to False if fast_executemany is set,
3157 # as of SQLAlchemy 2.0.9
3158 use_insertmanyvalues_wo_returning = True
3160 insertmanyvalues_implicit_sentinel = (
3161 InsertmanyvaluesSentinelOpts.AUTOINCREMENT
3162 | InsertmanyvaluesSentinelOpts.IDENTITY
3163 | InsertmanyvaluesSentinelOpts.USE_INSERT_FROM_SELECT
3164 )
3166 # "The incoming request has too many parameters. The server supports a "
3167 # "maximum of 2100 parameters."
3168 # in fact you can have 2099 parameters.
3169 insertmanyvalues_max_parameters = 2099
3171 _supports_offset_fetch = False
3172 _supports_nvarchar_max = False
3174 legacy_schema_aliasing = False
3176 server_version_info = ()
3178 statement_compiler = MSSQLCompiler
3179 ddl_compiler = MSDDLCompiler
3180 type_compiler_cls = MSTypeCompiler
3181 preparer = MSIdentifierPreparer
3183 construct_arguments = [
3184 (sa_schema.PrimaryKeyConstraint, {"clustered": None}),
3185 (sa_schema.UniqueConstraint, {"clustered": None}),
3186 (
3187 sa_schema.Index,
3188 {
3189 "clustered": None,
3190 "include": None,
3191 "where": None,
3192 "columnstore": None,
3193 },
3194 ),
3195 (
3196 sa_schema.Column,
3197 {"identity_start": None, "identity_increment": None},
3198 ),
3199 ]
3201 def __init__(
3202 self,
3203 query_timeout=None,
3204 use_scope_identity=True,
3205 schema_name="dbo",
3206 deprecate_large_types=None,
3207 supports_comments=None,
3208 json_serializer=None,
3209 json_deserializer=None,
3210 legacy_schema_aliasing=None,
3211 ignore_no_transaction_on_rollback=False,
3212 **opts,
3213 ):
3214 self.query_timeout = int(query_timeout or 0)
3215 self.schema_name = schema_name
3217 self.use_scope_identity = use_scope_identity
3218 self.deprecate_large_types = deprecate_large_types
3219 self.ignore_no_transaction_on_rollback = (
3220 ignore_no_transaction_on_rollback
3221 )
3222 self._user_defined_supports_comments = uds = supports_comments
3223 if uds is not None:
3224 self.supports_comments = uds
3226 if legacy_schema_aliasing is not None:
3227 util.warn_deprecated(
3228 "The legacy_schema_aliasing parameter is "
3229 "deprecated and will be removed in a future release.",
3230 "1.4",
3231 )
3232 self.legacy_schema_aliasing = legacy_schema_aliasing
3234 super().__init__(**opts)
3236 self._json_serializer = json_serializer
3237 self._json_deserializer = json_deserializer
3239 def do_savepoint(self, connection, name):
3240 # give the DBAPI a push
3241 connection.exec_driver_sql("IF @@TRANCOUNT = 0 BEGIN TRANSACTION")
3242 super().do_savepoint(connection, name)
3244 def do_release_savepoint(self, connection, name):
3245 # SQL Server does not support RELEASE SAVEPOINT
3246 pass
3248 def do_rollback(self, dbapi_connection):
3249 try:
3250 super().do_rollback(dbapi_connection)
3251 except self.dbapi.ProgrammingError as e:
3252 if self.ignore_no_transaction_on_rollback and re.match(
3253 r".*\b111214\b", str(e)
3254 ):
3255 util.warn(
3256 "ProgrammingError 111214 "
3257 "'No corresponding transaction found.' "
3258 "has been suppressed via "
3259 "ignore_no_transaction_on_rollback=True"
3260 )
3261 else:
3262 raise
3264 _isolation_lookup = {
3265 "SERIALIZABLE",
3266 "READ UNCOMMITTED",
3267 "READ COMMITTED",
3268 "REPEATABLE READ",
3269 "SNAPSHOT",
3270 }
3272 def get_isolation_level_values(self, dbapi_connection):
3273 return list(self._isolation_lookup)
3275 def set_isolation_level(self, dbapi_connection, level):
3276 cursor = dbapi_connection.cursor()
3277 cursor.execute(f"SET TRANSACTION ISOLATION LEVEL {level}")
3278 cursor.close()
3279 if level == "SNAPSHOT":
3280 dbapi_connection.commit()
3282 def get_isolation_level(self, dbapi_connection):
3283 cursor = dbapi_connection.cursor()
3284 view_name = "sys.system_views"
3285 try:
3286 cursor.execute(
3287 (
3288 "SELECT name FROM {} WHERE name IN "
3289 "('dm_exec_sessions', 'dm_pdw_nodes_exec_sessions')"
3290 ).format(view_name)
3291 )
3292 row = cursor.fetchone()
3293 if not row:
3294 raise NotImplementedError(
3295 "Can't fetch isolation level on this particular "
3296 "SQL Server version."
3297 )
3299 view_name = f"sys.{row[0]}"
3301 cursor.execute("""
3302 SELECT CASE transaction_isolation_level
3303 WHEN 0 THEN NULL
3304 WHEN 1 THEN 'READ UNCOMMITTED'
3305 WHEN 2 THEN 'READ COMMITTED'
3306 WHEN 3 THEN 'REPEATABLE READ'
3307 WHEN 4 THEN 'SERIALIZABLE'
3308 WHEN 5 THEN 'SNAPSHOT' END
3309 AS TRANSACTION_ISOLATION_LEVEL
3310 FROM {}
3311 where session_id = @@SPID
3312 """.format(view_name))
3313 except self.dbapi.Error as err:
3314 raise NotImplementedError(
3315 "Can't fetch isolation level; encountered error {} when "
3316 'attempting to query the "{}" view.'.format(err, view_name)
3317 ) from err
3318 else:
3320 row = cursor.fetchone()
3321 return row[0].upper()
3322 finally:
3323 cursor.close()
3325 def min_year_for_type(self, typeobj):
3326 typeobj = sqltypes.to_instance(typeobj)
3328 if isinstance(typeobj, SMALLDATETIME):
3329 return 1900
3330 elif isinstance(typeobj, (DATETIME2, DATETIMEOFFSET)):
3331 return datetime.MINYEAR
3332 elif isinstance(typeobj, sqltypes.DateTime):
3333 # DateTime renders DATETIMEOFFSET with timezone=True,
3334 # else DATETIME
3335 return datetime.MINYEAR if typeobj.timezone else 1753
3336 elif (
3337 isinstance(typeobj, sqltypes.Date)
3338 and not isinstance(typeobj, sqltypes.DATE)
3339 and self.server_version_info < MS_2008_VERSION
3340 ):
3341 # Date renders DATETIME before SQL Server 2008
3342 return 1753
3343 else:
3344 return datetime.MINYEAR
3346 def initialize(self, connection):
3347 super().initialize(connection)
3348 self._setup_version_attributes()
3349 self._setup_supports_nvarchar_max(connection)
3350 self._setup_supports_comments(connection)
3352 def _setup_version_attributes(self):
3353 if self.server_version_info >= MS_2008_VERSION:
3354 self.supports_multivalues_insert = True
3355 else:
3356 self.supports_multivalues_insert = False
3358 if self.deprecate_large_types is None:
3359 self.deprecate_large_types = (
3360 self.server_version_info >= MS_2012_VERSION
3361 )
3363 self._supports_offset_fetch = (
3364 self.server_version_info and self.server_version_info[0] >= 11
3365 )
3367 def _setup_supports_nvarchar_max(self, connection):
3368 try:
3369 connection.scalar(
3370 sql.text("SELECT CAST('test max support' AS NVARCHAR(max))")
3371 )
3372 except exc.DBAPIError:
3373 self._supports_nvarchar_max = False
3374 else:
3375 self._supports_nvarchar_max = True
3377 def _setup_supports_comments(self, connection):
3378 if self._user_defined_supports_comments is not None:
3379 return
3381 try:
3382 connection.scalar(
3383 sql.text(
3384 "SELECT 1 FROM fn_listextendedproperty"
3385 "(default, default, default, default, "
3386 "default, default, default)"
3387 )
3388 )
3389 except exc.DBAPIError:
3390 self.supports_comments = False
3391 else:
3392 self.supports_comments = True
3394 def _get_default_schema_name(self, connection):
3395 query = sql.text("SELECT schema_name()")
3396 default_schema_name = connection.scalar(query)
3397 if default_schema_name is not None:
3398 # guard against the case where the default_schema_name is being
3399 # fed back into a table reflection function.
3400 return quoted_name(default_schema_name, quote=True)
3401 else:
3402 return self.schema_name
3404 @reflection.cache
3405 @_db_plus_owner
3406 def has_table(self, connection, tablename, dbname, owner, schema, **kw):
3407 self._ensure_has_table_connection(connection)
3409 return self._internal_has_multi_table(
3410 connection, [tablename], owner, schema, **kw
3411 )[schema, tablename]
3413 def has_multi_table(self, connection, table_names, schema=None, **kw):
3414 # this does not really matche the signature of _db_plus_owner_multi
3415 # so manually do the unpacking here
3416 dbname, owner = _owner_plus_db(self, schema)
3417 return _switch_db(
3418 dbname,
3419 connection,
3420 self._internal_has_multi_table,
3421 connection,
3422 table_names,
3423 owner,
3424 schema,
3425 **kw,
3426 ).items()
3428 @reflection.cache
3429 @_db_plus_owner
3430 def has_sequence(
3431 self, connection, sequencename, dbname, owner, schema, **kw
3432 ):
3433 sequences = ischema.sequences
3435 s = sql.select(sequences.c.sequence_name).where(
3436 sequences.c.sequence_name == sequencename
3437 )
3439 if owner:
3440 s = s.where(sequences.c.sequence_schema == owner)
3442 c = connection.execute(s)
3444 return c.first() is not None
3446 @reflection.cache
3447 @_db_plus_owner_listing
3448 def get_sequence_names(self, connection, dbname, owner, schema, **kw):
3449 sequences = ischema.sequences
3451 s = sql.select(sequences.c.sequence_name)
3452 if owner:
3453 s = s.where(sequences.c.sequence_schema == owner)
3455 c = connection.execute(s)
3457 return [row[0] for row in c]
3459 @reflection.cache
3460 def get_schema_names(self, connection, **kw):
3461 s = sql.select(ischema.schemata.c.schema_name).order_by(
3462 ischema.schemata.c.schema_name
3463 )
3464 schema_names = [r[0] for r in connection.execute(s)]
3465 return schema_names
3467 @reflection.cache
3468 @_db_plus_owner_listing
3469 def get_table_names(self, connection, dbname, owner, schema, **kw):
3470 tables = ischema.tables
3471 s = (
3472 sql.select(tables.c.table_name)
3473 .where(
3474 sql.and_(
3475 tables.c.table_schema == owner,
3476 tables.c.table_type == "BASE TABLE",
3477 )
3478 )
3479 .order_by(tables.c.table_name)
3480 )
3481 table_names = [r[0] for r in connection.execute(s)]
3482 return table_names
3484 @reflection.cache
3485 @_db_plus_owner_listing
3486 def get_view_names(self, connection, dbname, owner, schema, **kw):
3487 tables = ischema.tables
3488 s = (
3489 sql.select(tables.c.table_name)
3490 .where(
3491 sql.and_(
3492 tables.c.table_schema == owner,
3493 tables.c.table_type == "VIEW",
3494 )
3495 )
3496 .order_by(tables.c.table_name)
3497 )
3498 view_names = [r[0] for r in connection.execute(s)]
3499 return view_names
3501 def _internal_has_multi_table(
3502 self, connection, table_names, owner, schema, **kw
3503 ):
3504 regular_names, multi_object_names, temp_names = (
3505 self._partition_filter_names(
3506 connection, owner, table_names, ObjectScope.ANY, ObjectKind.ANY
3507 )
3508 )
3510 result = {}
3511 if multi_object_names:
3512 # in multi_object_names we already have the names that exist in
3513 # the database, no need to do another query here
3514 name_map = self._multi_name_map(regular_names)
3515 for server_name in multi_object_names:
3516 result[(schema, name_map(server_name))] = True
3518 if temp_names:
3519 # mssql does not support temporary views
3520 # U filters on user tables only.
3521 query = text("SELECT object_id(:table_name, 'U')")
3522 for temp_name in temp_names:
3523 result[(schema, temp_name)] = bool(
3524 connection.scalar(
3525 query,
3526 {"table_name": f"tempdb..[{temp_name}]"},
3527 )
3528 )
3530 for name in table_names:
3531 result.setdefault((schema, name), False)
3532 return result
3534 @reflection.cache
3535 @_db_plus_owner
3536 def get_view_definition(
3537 self, connection, viewname, dbname, owner, schema, **kw
3538 ):
3539 view_def = connection.execute(
3540 sql.text(
3541 "select mod.definition "
3542 "from sys.sql_modules as mod "
3543 "join sys.views as views on mod.object_id = views.object_id "
3544 "join sys.schemas as sch on views.schema_id = sch.schema_id "
3545 "where views.name=:viewname and sch.name=:schname"
3546 ).bindparams(
3547 sql.bindparam("viewname", viewname, ischema.CoerceUnicode()),
3548 sql.bindparam("schname", owner, ischema.CoerceUnicode()),
3549 )
3550 ).scalar()
3551 if view_def:
3552 return view_def
3553 else:
3554 raise exc.NoSuchTableError(f"{owner}.{viewname}")
3556 def _temp_table_name_like_pattern(self, tablename):
3557 # LIKE uses '%' to match zero or more characters and '_' to match any
3558 # single character. We want to match literal underscores, so T-SQL
3559 # requires that we enclose them in square brackets.
3560 return tablename + (
3561 ("[_][_][_]%") if not tablename.startswith("##") else ""
3562 )
3564 def _get_internal_temp_table_name(self, connection, tablename):
3565 # it's likely that schema is always "dbo", but since we can
3566 # get it here, let's get it.
3567 # see https://stackoverflow.com/questions/8311959/
3568 # specifying-schema-for-temporary-tables
3570 try:
3571 return connection.execute(
3572 sql.text(
3573 "select table_schema, table_name "
3574 "from tempdb.information_schema.tables "
3575 "where table_name like :p1"
3576 ),
3577 {"p1": self._temp_table_name_like_pattern(tablename)},
3578 ).one()
3579 except exc.MultipleResultsFound as me:
3580 raise exc.UnreflectableTableError(
3581 "Found more than one temporary table named '%s' in tempdb "
3582 "at this time. Cannot reliably resolve that name to its "
3583 "internal table name." % tablename
3584 ) from me
3585 except exc.NoResultFound as ne:
3586 raise exc.NoSuchTableError(
3587 "Unable to find a temporary table named '%s' in tempdb."
3588 % tablename
3589 ) from ne
3591 def _parse_column_info(
3592 self,
3593 name,
3594 type_,
3595 nullable,
3596 maxlen,
3597 numericprec,
3598 numericscale,
3599 default,
3600 collation,
3601 definition,
3602 is_persisted,
3603 is_identity,
3604 identity_start,
3605 identity_increment,
3606 comment,
3607 base_type=None,
3608 ):
3609 # Try to resolve the user type first (e.g., "sysname"),
3610 # then fall back to the base type (e.g., "nvarchar").
3611 # base_type may be None for CLR types (geography, geometry,
3612 # hierarchyid) which have no corresponding base type.
3613 coltype = self.ischema_names.get(type_, None)
3614 if coltype is None and base_type is not None and base_type != type_:
3615 coltype = self.ischema_names.get(base_type, None)
3616 kwargs = {}
3618 if coltype in (MSBinary, MSVarBinary, sqltypes.LargeBinary):
3619 kwargs["length"] = maxlen if maxlen != -1 else None
3620 elif coltype in (MSString, MSChar):
3621 kwargs["length"] = maxlen if maxlen != -1 else None
3622 if collation:
3623 kwargs["collation"] = collation
3624 elif coltype in (MSNVarchar, MSNChar):
3625 kwargs["length"] = maxlen // 2 if maxlen != -1 else None
3626 if collation:
3627 kwargs["collation"] = collation
3628 elif coltype in (MSText, MSNText):
3629 # TEXT / NTEXT are unlengthed LOB types. sys.columns.max_length
3630 # reports 16 for these, which is the size of the in-row LOB
3631 # pointer and not a character length, so no length is applied.
3632 if collation:
3633 kwargs["collation"] = collation
3635 if coltype is None:
3636 if base_type is not None and base_type != type_:
3637 util.warn(
3638 "Did not recognize type '%s' (user type) or '%s' "
3639 "(base type) of column '%s'" % (type_, base_type, name)
3640 )
3641 else:
3642 util.warn(
3643 "Did not recognize type '%s' of column '%s'"
3644 % (type_, name)
3645 )
3646 coltype = sqltypes.NULLTYPE
3647 else:
3648 if issubclass(coltype, sqltypes.NumericCommon):
3649 kwargs["precision"] = numericprec
3650 if not issubclass(coltype, sqltypes.Float):
3651 kwargs["scale"] = numericscale
3652 coltype = coltype(**kwargs)
3654 cdict = {
3655 "name": name,
3656 "type": coltype,
3657 "nullable": nullable,
3658 "default": default,
3659 "autoincrement": is_identity is not None,
3660 "comment": comment,
3661 }
3663 if definition is not None and is_persisted is not None:
3664 cdict["computed"] = {
3665 "sqltext": definition,
3666 "persisted": is_persisted,
3667 }
3669 if is_identity is not None:
3670 if identity_start is None or identity_increment is None:
3671 cdict["identity"] = {}
3672 else:
3673 if isinstance(
3674 coltype, (sqltypes.BigInteger, sqltypes.Integer)
3675 ):
3676 start = int(identity_start)
3677 increment = int(identity_increment)
3678 else:
3679 start = identity_start
3680 increment = identity_increment
3681 cdict["identity"] = {
3682 "start": start,
3683 "increment": increment,
3684 }
3686 return cdict
3688 @lru_cache()
3689 def _columns_select(self):
3690 """Build the unified Core sys.* select for column reflection.
3692 Returns a ``select()`` that includes ``table_name`` and ``owner``
3693 in the result so callers can group rows by table.
3695 Used by :meth:`.get_multi_columns` (and indirectly by
3696 :meth:`.get_columns` via the multi delegation).
3697 """
3698 sys_columns = ischema.sys_columns
3699 sys_types = ischema.sys_types
3700 sys_base_types = ischema.sys_types.alias("base_types")
3701 sys_default_constraints = ischema.sys_default_constraints
3702 computed_cols = ischema.computed_columns
3703 identity_cols = ischema.identity_columns
3704 extended_properties = ischema.extended_properties
3705 sys_objects = ischema.sys_objects
3706 sys_schemas = ischema.sys_schemas
3708 if self._supports_nvarchar_max:
3709 computed_definition = computed_cols.c.definition
3710 else:
3711 # tds_version 4.2 does not support NVARCHAR(MAX)
3712 computed_definition = sql.cast(
3713 computed_cols.c.definition, NVARCHAR(4000)
3714 )
3716 s = (
3717 sql.select(
3718 sys_objects.c.name.label("table_name"),
3719 sys_schemas.c.name.label("owner"),
3720 sys_columns.c.name.label("column_name"),
3721 sys_types.c.name.label("type_name"),
3722 sys_base_types.c.name.label("base_type"),
3723 sys_columns.c.is_nullable,
3724 sys_columns.c.max_length,
3725 sys_columns.c.precision,
3726 sys_columns.c.scale,
3727 sys_default_constraints.c.definition.label("default_value"),
3728 sys_columns.c.collation_name,
3729 computed_definition.label("computed_definition"),
3730 computed_cols.c.is_persisted,
3731 identity_cols.c.is_identity,
3732 identity_cols.c.seed_value,
3733 identity_cols.c.increment_value,
3734 extended_properties.c.value.label("comment"),
3735 sys_columns.c.column_id,
3736 )
3737 .select_from(sys_columns)
3738 .join(
3739 sys_objects,
3740 onclause=sys_columns.c.object_id == sys_objects.c.object_id,
3741 )
3742 .join(
3743 sys_schemas,
3744 onclause=sys_objects.c.schema_id == sys_schemas.c.schema_id,
3745 )
3746 .join(
3747 sys_types,
3748 onclause=sys_columns.c.user_type_id
3749 == sys_types.c.user_type_id,
3750 )
3751 .outerjoin(
3752 sys_base_types,
3753 onclause=sql.and_(
3754 sys_types.c.system_type_id
3755 == sys_base_types.c.system_type_id,
3756 sys_base_types.c.user_type_id
3757 == sys_base_types.c.system_type_id,
3758 ),
3759 )
3760 .outerjoin(
3761 sys_default_constraints,
3762 sql.and_(
3763 sys_default_constraints.c.object_id
3764 == sys_columns.c.default_object_id,
3765 sys_default_constraints.c.parent_column_id
3766 == sys_columns.c.column_id,
3767 ),
3768 )
3769 .outerjoin(
3770 computed_cols,
3771 onclause=sql.and_(
3772 computed_cols.c.object_id == sys_columns.c.object_id,
3773 computed_cols.c.column_id == sys_columns.c.column_id,
3774 ),
3775 )
3776 .outerjoin(
3777 identity_cols,
3778 onclause=sql.and_(
3779 identity_cols.c.object_id == sys_columns.c.object_id,
3780 identity_cols.c.column_id == sys_columns.c.column_id,
3781 ),
3782 )
3783 .outerjoin(
3784 extended_properties,
3785 onclause=sql.and_(
3786 extended_properties.c["class"] == 1,
3787 extended_properties.c.name == "MS_Description",
3788 sys_columns.c.object_id == extended_properties.c.major_id,
3789 sys_columns.c.column_id == extended_properties.c.minor_id,
3790 ),
3791 )
3792 .where(
3793 ischema.sys_schemas.c.name == sql.bindparam("owner"),
3794 ischema.sys_objects.c.name.in_(
3795 sql.bindparam("filter_names", expanding=True)
3796 ),
3797 )
3798 .order_by(sys_objects.c.name, sys_columns.c.column_id)
3799 )
3800 return s
3802 @staticmethod
3803 def _fk_query_sql(fk_info_where, extra_cols=""):
3804 return """\
3805WITH fk_info AS (
3806 SELECT
3807 ischema_ref_con.constraint_schema,
3808 ischema_ref_con.constraint_name,
3809 ischema_key_col.ordinal_position,
3810 ischema_key_col.table_schema,
3811 ischema_key_col.table_name,
3812 ischema_ref_con.unique_constraint_schema,
3813 ischema_ref_con.unique_constraint_name,
3814 ischema_ref_con.match_option,
3815 ischema_ref_con.update_rule,
3816 ischema_ref_con.delete_rule,
3817 ischema_key_col.column_name AS constrained_column
3818 FROM
3819 INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS ischema_ref_con
3820 INNER JOIN
3821 INFORMATION_SCHEMA.KEY_COLUMN_USAGE ischema_key_col ON
3822 ischema_key_col.table_schema =
3823 ischema_ref_con.constraint_schema
3824 AND ischema_key_col.constraint_name =
3825 ischema_ref_con.constraint_name
3826 WHERE %(fk_info_where)s
3827),
3828constraint_info AS (
3829 SELECT
3830 ischema_key_col.constraint_schema,
3831 ischema_key_col.constraint_name,
3832 ischema_key_col.ordinal_position,
3833 ischema_key_col.table_schema,
3834 ischema_key_col.table_name,
3835 ischema_key_col.column_name
3836 FROM
3837 INFORMATION_SCHEMA.KEY_COLUMN_USAGE ischema_key_col
3838),
3839index_info AS (
3840 SELECT
3841 sys.schemas.name AS index_schema,
3842 sys.indexes.name AS index_name,
3843 sys.index_columns.key_ordinal AS ordinal_position,
3844 sys.schemas.name AS table_schema,
3845 sys.objects.name AS table_name,
3846 sys.columns.name AS column_name
3847 FROM
3848 sys.indexes
3849 INNER JOIN sys.objects
3850 ON sys.objects.object_id = sys.indexes.object_id
3851 INNER JOIN sys.schemas
3852 ON sys.schemas.schema_id = sys.objects.schema_id
3853 INNER JOIN sys.index_columns
3854 ON sys.index_columns.object_id = sys.objects.object_id
3855 AND sys.index_columns.index_id = sys.indexes.index_id
3856 INNER JOIN sys.columns
3857 ON sys.columns.object_id = sys.indexes.object_id
3858 AND sys.columns.column_id = sys.index_columns.column_id
3859)
3860 SELECT
3861 %(extra_cols)s
3862 fk_info.constraint_schema,
3863 fk_info.constraint_name,
3864 fk_info.ordinal_position,
3865 fk_info.constrained_column,
3866 constraint_info.table_schema AS referred_table_schema,
3867 constraint_info.table_name AS referred_table_name,
3868 constraint_info.column_name AS referred_column,
3869 fk_info.match_option,
3870 fk_info.update_rule,
3871 fk_info.delete_rule
3872 FROM
3873 fk_info INNER JOIN constraint_info ON
3874 constraint_info.constraint_schema =
3875 fk_info.unique_constraint_schema
3876 AND constraint_info.constraint_name =
3877 fk_info.unique_constraint_name
3878 AND constraint_info.ordinal_position = fk_info.ordinal_position
3879 UNION
3880 SELECT
3881 %(extra_cols)s
3882 fk_info.constraint_schema,
3883 fk_info.constraint_name,
3884 fk_info.ordinal_position,
3885 fk_info.constrained_column,
3886 index_info.table_schema AS referred_table_schema,
3887 index_info.table_name AS referred_table_name,
3888 index_info.column_name AS referred_column,
3889 fk_info.match_option,
3890 fk_info.update_rule,
3891 fk_info.delete_rule
3892 FROM
3893 fk_info INNER JOIN index_info ON
3894 index_info.index_schema = fk_info.unique_constraint_schema
3895 AND index_info.index_name = fk_info.unique_constraint_name
3896 AND index_info.ordinal_position = fk_info.ordinal_position
3897 AND NOT (index_info.table_schema = fk_info.table_schema
3898 AND index_info.table_name = fk_info.table_name)
3900 ORDER BY fk_info.constraint_schema, fk_info.constraint_name,
3901 fk_info.ordinal_position
3902""" % {
3903 "fk_info_where": fk_info_where,
3904 "extra_cols": extra_cols,
3905 }
3907 # --- multi-reflection API ---
3909 @lru_cache
3910 def _partition_filter_names_query(self, add_filter_names: bool):
3911 sys_objects = ischema.sys_objects
3912 sys_schemas = ischema.sys_schemas
3914 s = (
3915 sql.select(sys_objects.c.name)
3916 .select_from(sys_objects)
3917 .join(
3918 sys_schemas,
3919 onclause=sys_objects.c.schema_id == sys_schemas.c.schema_id,
3920 )
3921 .where(
3922 sys_schemas.c.name == sql.bindparam("schema"),
3923 sys_objects.c.type.in_(
3924 sql.bindparam("type_filter", expanding=True)
3925 ),
3926 )
3927 )
3928 if add_filter_names:
3929 s = s.where(
3930 sys_objects.c.name.in_(
3931 sql.bindparam("filter_names", expanding=True)
3932 )
3933 )
3935 return s
3937 def _partition_filter_names(
3938 self, connection, owner, filter_names, scope, kind
3939 ):
3940 """Split caller-supplied ``filter_names`` between the bulk
3941 ``sys.*`` query path (regular objects in the user database) and
3942 the temp object path (objects living in ``tempdb``), and resolve
3943 the existing-in-catalog names for the bulk path in one shot.
3945 Returns the tuple
3946 ``(regular_names, multi_object_names, temp_names)``:
3948 * ``regular_names`` -- list of caller-supplied names (with the
3949 caller's casing) for the bulk path, used to build the
3950 case-insensitive name map. Empty list when no regular names
3951 were requested.
3952 * ``multi_object_names`` -- list of names (with server-side
3953 casing) that actually exist in the catalog and match the
3954 requested ``kind``. This is what the bulk SQL's IN clause
3955 gets filtered with. Empty list when no matches, disabling
3956 the bulk query.
3957 * ``temp_names`` -- list of ``#``-prefixed temp object names
3958 that the caller asked for and that the requested
3959 ``scope``/``kind`` allows.
3961 ``scope=ObjectScope.DEFAULT`` excludes temp objects entirely.
3962 ``scope=ObjectScope.TEMPORARY`` excludes regular objects.
3963 ``ObjectKind.VIEW`` (without TABLE) excludes temp because SQL
3964 Server does not have temp views in the same form. SQL Server
3965 also has no ``get_temp_table_names`` implementation, so temp
3966 objects can only be reflected when the caller supplies their
3967 names explicitly.
3968 """
3969 include_default = scope is not ObjectScope.TEMPORARY
3970 include_temp = (
3971 scope is not ObjectScope.DEFAULT and ObjectKind.TABLE in kind
3972 )
3974 if filter_names:
3975 if include_temp:
3976 temp_names = [n for n in filter_names if n.startswith("#")]
3977 else:
3978 temp_names = []
3980 if include_default:
3981 regular_names = [
3982 n for n in filter_names if not n.startswith("#")
3983 ]
3984 run_bulk = bool(regular_names)
3985 else:
3986 regular_names = []
3987 run_bulk = False
3988 else:
3989 temp_names = []
3990 regular_names = []
3991 run_bulk = include_default
3993 if not run_bulk:
3994 return (regular_names, [], temp_names)
3996 type_filter = []
3997 if ObjectKind.TABLE in kind:
3998 type_filter.append("U")
3999 if ObjectKind.VIEW in kind:
4000 type_filter.append("V")
4001 # SQL Server does not support materialized views, so ignore them
4002 if not type_filter:
4003 return (regular_names, [], temp_names)
4005 has_regular_names = bool(regular_names)
4006 query = self._partition_filter_names_query(has_regular_names)
4007 params = {"schema": owner, "type_filter": type_filter}
4008 if has_regular_names:
4009 params["filter_names"] = regular_names
4011 multi_object_names = connection.scalars(query, params).all()
4013 return (regular_names, multi_object_names, temp_names)
4015 @staticmethod
4016 def _multi_name_map(filter_names):
4017 """Build a case-insensitive ``server_name -> user_name`` mapper.
4019 MSSQL object names are case-insensitive under the default
4020 collation, so a user may pass ``"sOmEtAbLe"`` for a table
4021 physically stored as ``"SomeTable"``. The bulk SQL returns the
4022 server-side casing, but result keys must use the user-supplied
4023 casing so that downstream code (e.g.
4024 :meth:`.Inspector.reflect_table`) can find the entry.
4026 Returns a callable ``(server_name) -> user_name``. When
4027 ``filter_names`` is empty/None, returns identity.
4028 """
4029 if not filter_names:
4030 return lambda n: n
4031 lookup = {n.lower(): n for n in filter_names}
4032 return lambda n: lookup.get(n.lower(), n)
4034 @_db_plus_owner_multi
4035 def get_multi_columns(
4036 self,
4037 connection,
4038 dbname,
4039 owner,
4040 schema,
4041 filter_names,
4042 scope,
4043 kind,
4044 **kw,
4045 ):
4046 regular_names, multi_object_names, temp_names = (
4047 self._partition_filter_names(
4048 connection, owner, filter_names, scope, kind
4049 )
4050 )
4052 result = {}
4054 if multi_object_names:
4055 name_map = self._multi_name_map(regular_names)
4056 self._fetch_multi_columns(
4057 connection,
4058 owner=owner,
4059 names=multi_object_names,
4060 schema=schema,
4061 name_map=name_map,
4062 result=result,
4063 exec_opts={"schema_translate_map": {}},
4064 )
4066 if temp_names:
4067 self._fetch_multi_columns_temp(
4068 connection,
4069 temp_names=temp_names,
4070 schema=schema,
4071 result=result,
4072 )
4074 return result.items()
4076 def _fetch_multi_columns(
4077 self, connection, owner, names, schema, name_map, result, exec_opts
4078 ):
4079 """Execute the unified columns select for a single catalog pass.
4081 Used by :meth:`.get_multi_columns` for both the main-database pass
4082 (no ``schema_translate_map``) and the tempdb pass (with
4083 ``schema_translate_map={"sys": "tempdb.sys"}`` applied).
4084 """
4085 s = self._columns_select()
4087 rp = connection.execute(
4088 s,
4089 {"owner": owner, "filter_names": names},
4090 execution_options=exec_opts,
4091 )
4093 for row in rp.mappings():
4094 table_name = name_map(row["table_name"])
4095 cdict = self._parse_column_info(
4096 name=row["column_name"],
4097 type_=row["type_name"],
4098 base_type=row["base_type"],
4099 nullable=row["is_nullable"] == 1,
4100 maxlen=row["max_length"],
4101 numericprec=row["precision"],
4102 numericscale=row["scale"],
4103 default=row["default_value"],
4104 collation=row["collation_name"],
4105 definition=row["computed_definition"],
4106 is_persisted=row["is_persisted"],
4107 is_identity=row["is_identity"],
4108 identity_start=row["seed_value"],
4109 identity_increment=row["increment_value"],
4110 comment=row["comment"],
4111 )
4112 result.setdefault((schema, table_name), []).append(cdict)
4114 for n in names:
4115 key = (schema, name_map(n))
4116 if key not in result:
4117 result[key] = ReflectionDefaults.columns()
4119 def _fetch_multi_columns_temp(
4120 self, connection, temp_names, schema, result
4121 ):
4122 """Run the unified columns select against tempdb for temp names.
4124 Resolves each ``#name`` to its mangled tempdb name (one round
4125 trip per temp via :meth:`._get_internal_temp_table_name`), then
4126 runs a single bulk select against ``tempdb.sys.*`` via
4127 ``schema_translate_map``. The result rows (keyed by mangled
4128 name) are mapped back to the original ``#name`` for the caller.
4129 """
4130 resolved_by_owner, temp_name_map = self._resolve_temp_names(
4131 connection, temp_names
4132 )
4133 for temp_owner, mangled_names in resolved_by_owner.items():
4134 self._fetch_multi_columns(
4135 connection,
4136 owner=temp_owner,
4137 names=mangled_names,
4138 schema=schema,
4139 name_map=temp_name_map,
4140 result=result,
4141 exec_opts={"schema_translate_map": {"sys": "tempdb.sys"}},
4142 )
4144 def _resolve_temp_names(self, connection, temp_names):
4145 """Resolve user-facing temp names (``#foo``) to their mangled
4146 tempdb names, grouped by owner.
4148 Returns ``(resolved_by_owner, temp_name_map)``:
4150 * ``resolved_by_owner`` -- ``{owner: [mangled_name, ...]}``,
4151 one bulk query per owner (typically just ``'dbo'``).
4152 * ``temp_name_map`` -- a function that maps mangled names back
4153 to the user-supplied name.
4155 Names that don't resolve (no matching temp table in tempdb) are
4156 silently dropped. The single-table reflection wrappers handle
4157 the resulting absence by raising :exc:`.NoSuchTableError` via
4158 :meth:`._value_or_raise`.
4159 """
4160 resolved_by_owner = {}
4161 original_by_mangled = {}
4162 for name in temp_names:
4163 try:
4164 temp_owner, mangled = self._get_internal_temp_table_name(
4165 connection, name
4166 )
4167 except exc.NoSuchTableError:
4168 continue
4169 resolved_by_owner.setdefault(temp_owner, []).append(mangled)
4170 original_by_mangled[mangled] = name
4172 def temp_name_map(server_name):
4173 return original_by_mangled.get(server_name, server_name)
4175 return resolved_by_owner, temp_name_map
4177 @_db_plus_owner_multi
4178 def get_multi_pk_constraint(
4179 self,
4180 connection,
4181 dbname,
4182 owner,
4183 schema,
4184 filter_names,
4185 scope,
4186 kind,
4187 **kw,
4188 ):
4189 regular_names, multi_object_names, temp_names = (
4190 self._partition_filter_names(
4191 connection, owner, filter_names, scope, kind
4192 )
4193 )
4195 result = {}
4197 if multi_object_names:
4198 name_map = self._multi_name_map(regular_names)
4199 self._fetch_multi_pk_constraint(
4200 connection,
4201 owner=owner,
4202 names=multi_object_names,
4203 schema=schema,
4204 name_map=name_map,
4205 result=result,
4206 exec_opts={"schema_translate_map": {}},
4207 )
4209 if temp_names:
4210 self._fetch_multi_pk_constraint_temp(
4211 connection,
4212 temp_names=temp_names,
4213 schema=schema,
4214 result=result,
4215 )
4217 return result.items()
4219 @lru_cache()
4220 def _pk_constraint_select(self):
4221 """Build the unified Core sys.* select for PK constraint reflection.
4223 Returns a ``select()`` that includes ``table_name`` and ``owner``
4224 in the result so callers can group rows by table. The clustered
4225 flag comes directly from ``sys.indexes.type`` (1 = clustered).
4226 """
4227 sys_key_constraints = ischema.sys_key_constraints
4228 sys_indexes = ischema.sys_indexes
4229 sys_index_columns = ischema.sys_index_columns
4230 sys_columns = ischema.sys_columns
4231 sys_objects = ischema.sys_objects
4232 sys_schemas = ischema.sys_schemas
4234 s = (
4235 sql.select(
4236 sys_objects.c.name.label("table_name"),
4237 sys_schemas.c.name.label("owner"),
4238 sys_key_constraints.c.name.label("constraint_name"),
4239 sys_columns.c.name.label("column_name"),
4240 sys_index_columns.c.key_ordinal,
4241 sys_indexes.c.type.label("index_type"),
4242 )
4243 .select_from(sys_key_constraints)
4244 .join(
4245 sys_objects,
4246 onclause=sys_key_constraints.c.parent_object_id
4247 == sys_objects.c.object_id,
4248 )
4249 .join(
4250 sys_schemas,
4251 onclause=sys_objects.c.schema_id == sys_schemas.c.schema_id,
4252 )
4253 .join(
4254 sys_index_columns,
4255 onclause=sql.and_(
4256 sys_index_columns.c.object_id
4257 == sys_key_constraints.c.parent_object_id,
4258 sys_index_columns.c.index_id
4259 == sys_key_constraints.c.unique_index_id,
4260 ),
4261 )
4262 .join(
4263 sys_columns,
4264 onclause=sql.and_(
4265 sys_columns.c.object_id == sys_index_columns.c.object_id,
4266 sys_columns.c.column_id == sys_index_columns.c.column_id,
4267 ),
4268 )
4269 .join(
4270 sys_indexes,
4271 onclause=sql.and_(
4272 sys_indexes.c.object_id
4273 == sys_key_constraints.c.parent_object_id,
4274 sys_indexes.c.index_id
4275 == sys_key_constraints.c.unique_index_id,
4276 ),
4277 )
4278 .where(
4279 sys_key_constraints.c.type == "PK",
4280 ischema.sys_schemas.c.name == sql.bindparam("owner"),
4281 ischema.sys_objects.c.name.in_(
4282 sql.bindparam("filter_names", expanding=True)
4283 ),
4284 )
4285 .order_by(sys_objects.c.name, sys_index_columns.c.key_ordinal)
4286 )
4287 return s
4289 def _fetch_multi_pk_constraint(
4290 self, connection, owner, names, schema, name_map, result, exec_opts
4291 ):
4292 """Execute the unified pk_constraint select for one catalog pass.
4294 Used by :meth:`.get_multi_pk_constraint` for both the main DB
4295 pass and the tempdb pass (with ``schema_translate_map`` set).
4296 """
4297 s = self._pk_constraint_select()
4299 rp = connection.execute(
4300 s,
4301 {"owner": owner, "filter_names": names},
4302 execution_options=exec_opts,
4303 )
4305 for row in rp.mappings():
4306 table_name = name_map(row["table_name"])
4307 key = (schema, table_name)
4308 if key not in result:
4309 result[key] = {
4310 "constrained_columns": [],
4311 "name": row["constraint_name"],
4312 "dialect_options": {
4313 "mssql_clustered": row["index_type"] == 1
4314 },
4315 }
4316 result[key]["constrained_columns"].append(row["column_name"])
4318 for n in names:
4319 key = (schema, name_map(n))
4320 if key not in result:
4321 result[key] = ReflectionDefaults.pk_constraint()
4323 def _fetch_multi_pk_constraint_temp(
4324 self, connection, temp_names, schema, result
4325 ):
4326 """Run the unified pk_constraint select against tempdb."""
4327 resolved_by_owner, temp_name_map = self._resolve_temp_names(
4328 connection, temp_names
4329 )
4331 for temp_owner, mangled_names in resolved_by_owner.items():
4332 self._fetch_multi_pk_constraint(
4333 connection,
4334 owner=temp_owner,
4335 names=mangled_names,
4336 schema=schema,
4337 name_map=temp_name_map,
4338 result=result,
4339 exec_opts={"schema_translate_map": {"sys": "tempdb.sys"}},
4340 )
4342 @_db_plus_owner_multi
4343 def get_multi_foreign_keys(
4344 self,
4345 connection,
4346 dbname,
4347 owner,
4348 schema,
4349 filter_names,
4350 scope,
4351 kind,
4352 **kw,
4353 ):
4354 regular_names, multi_object_names, temp_names = (
4355 self._partition_filter_names(
4356 connection, owner, filter_names, scope, kind
4357 )
4358 )
4360 final = {}
4362 if multi_object_names:
4363 name_map = self._multi_name_map(regular_names)
4364 self._fetch_multi_foreign_keys(
4365 connection,
4366 dbname=dbname,
4367 owner=owner,
4368 names=multi_object_names,
4369 schema=schema,
4370 name_map=name_map,
4371 final=final,
4372 )
4374 # FK reflection for temp tables: the INFORMATION_SCHEMA query in
4375 # _fk_query_sql does not see tempdb objects (it queries the main
4376 # DB's catalog). The pre-existing single-table behavior was to
4377 # return empty foreign_keys for temp tables that exist. We
4378 # preserve that here by checking existence and returning the
4379 # default (empty) reflection for each temp that is reachable.
4380 has_temp_tables = self._internal_has_multi_table(
4381 connection, temp_names, owner="dbo", schema=schema, **kw
4382 )
4383 for name in temp_names:
4384 key = (schema, name)
4385 if has_temp_tables.get(key):
4386 final[key] = ReflectionDefaults.foreign_keys()
4388 return final.items()
4390 def _fetch_multi_foreign_keys(
4391 self, connection, dbname, owner, names, schema, name_map, final
4392 ):
4393 """Execute the unified FK query for the main-DB pass.
4395 Uses the _fk_query_sql sql text, restricted to the provided owner and
4396 IN-list of names. Groups result rows by table and constraint
4397 name into the final dict.
4398 """
4399 rp = connection.execute(
4400 sql.text(
4401 self._fk_query_sql(
4402 fk_info_where=(
4403 "ischema_key_col.table_schema = :owner"
4404 "\n AND ischema_key_col.table_name"
4405 " IN :filter_names"
4406 ),
4407 extra_cols="fk_info.table_name,",
4408 )
4409 )
4410 .bindparams(
4411 sql.bindparam("owner", owner, ischema.CoerceUnicode()),
4412 sql.bindparam("filter_names", names, expanding=True),
4413 )
4414 .columns(
4415 constraint_schema=sqltypes.Unicode(),
4416 constraint_name=sqltypes.Unicode(),
4417 table_name=sqltypes.Unicode(),
4418 constrained_column=sqltypes.Unicode(),
4419 referred_table_schema=sqltypes.Unicode(),
4420 referred_table_name=sqltypes.Unicode(),
4421 referred_column=sqltypes.Unicode(),
4422 )
4423 )
4425 grouped = {}
4426 for r in rp.all():
4427 (
4428 table_name,
4429 _, # constraint_schema
4430 rfknm,
4431 _, # ordinal
4432 scol,
4433 rschema,
4434 rtbl,
4435 rcol,
4436 _, # match
4437 fkuprule,
4438 fkdelrule,
4439 ) = r
4441 key = (schema, name_map(table_name))
4442 if key not in grouped:
4443 grouped[key] = util.defaultdict(
4444 lambda: {
4445 "name": None,
4446 "constrained_columns": [],
4447 "referred_schema": None,
4448 "referred_table": None,
4449 "referred_columns": [],
4450 "options": {},
4451 }
4452 )
4454 rec = grouped[key][rfknm]
4455 rec["name"] = rfknm
4457 if fkuprule != "NO ACTION":
4458 rec["options"]["onupdate"] = fkuprule
4459 if fkdelrule != "NO ACTION":
4460 rec["options"]["ondelete"] = fkdelrule
4462 if not rec["referred_table"]:
4463 rec["referred_table"] = rtbl
4464 if schema is not None or owner != rschema:
4465 if dbname:
4466 rschema = dbname + "." + rschema
4467 rec["referred_schema"] = rschema
4469 rec["constrained_columns"].append(scol)
4470 rec["referred_columns"].append(rcol)
4472 for key, fk_dict in grouped.items():
4473 final[key] = list(fk_dict.values())
4475 for n in names:
4476 key = (schema, name_map(n))
4477 if key not in final:
4478 final[key] = ReflectionDefaults.foreign_keys()
4480 @_db_plus_owner_multi
4481 def get_multi_indexes(
4482 self,
4483 connection,
4484 dbname,
4485 owner,
4486 schema,
4487 filter_names,
4488 scope,
4489 kind,
4490 **kw,
4491 ):
4492 regular_names, multi_object_names, temp_names = (
4493 self._partition_filter_names(
4494 connection, owner, filter_names, scope, kind
4495 )
4496 )
4498 result = {}
4500 if multi_object_names:
4501 name_map = self._multi_name_map(regular_names)
4502 self._fetch_multi_indexes(
4503 connection,
4504 owner=owner,
4505 names=multi_object_names,
4506 schema=schema,
4507 name_map=name_map,
4508 result=result,
4509 exec_opts={"schema_translate_map": {}},
4510 )
4512 if temp_names:
4513 self._fetch_multi_indexes_temp(
4514 connection,
4515 temp_names=temp_names,
4516 schema=schema,
4517 result=result,
4518 )
4520 return result.items()
4522 @lru_cache()
4523 def _indexes_metadata_select(self):
4524 """Build the Core sys.* select for index metadata (one row per
4525 index per table).
4527 Used by :meth:`._fetch_multi_indexes` for both the main-DB pass
4528 and the tempdb pass (via ``schema_translate_map``). Replaces a
4529 previous ``sql.text()`` body that did not honor
4530 ``schema_translate_map`` (which only rewrites Core
4531 schema-bearing objects, not literal SQL text).
4532 """
4533 sys_indexes = ischema.sys_indexes
4534 sys_objects = ischema.sys_objects
4535 sys_schemas = ischema.sys_schemas
4537 if self.server_version_info >= MS_2008_VERSION:
4538 filter_definition = sys_indexes.c.filter_definition
4539 else:
4540 filter_definition = sql.null()
4542 s = (
4543 sql.select(
4544 sys_objects.c.name.label("table_name"),
4545 sys_indexes.c.index_id,
4546 sys_indexes.c.is_unique,
4547 sys_indexes.c.name,
4548 sys_indexes.c.type,
4549 filter_definition.label("filter_definition"),
4550 )
4551 .select_from(sys_indexes)
4552 .join(
4553 sys_objects,
4554 onclause=sys_indexes.c.object_id == sys_objects.c.object_id,
4555 )
4556 .join(
4557 sys_schemas,
4558 onclause=sys_schemas.c.schema_id == sys_objects.c.schema_id,
4559 )
4560 .where(
4561 sys_indexes.c.is_primary_key == 0,
4562 sys_indexes.c.type != 0,
4563 ischema.sys_schemas.c.name == sql.bindparam("owner"),
4564 ischema.sys_objects.c.name.in_(
4565 sql.bindparam("filter_names", expanding=True)
4566 ),
4567 )
4568 .order_by(sys_objects.c.name, sys_indexes.c.name)
4569 )
4570 return s
4572 @lru_cache()
4573 def _indexes_columns_select(self):
4574 """Build the Core sys.* select for index columns (one row per
4575 index column per index per table)."""
4576 sys_index_columns = ischema.sys_index_columns
4577 sys_columns = ischema.sys_columns
4578 sys_objects = ischema.sys_objects
4579 sys_schemas = ischema.sys_schemas
4581 s = (
4582 sql.select(
4583 sys_objects.c.name.label("table_name"),
4584 sys_index_columns.c.index_id,
4585 sys_columns.c.name,
4586 sys_index_columns.c.is_included_column,
4587 )
4588 .select_from(sys_index_columns)
4589 .join(
4590 sys_columns,
4591 onclause=sql.and_(
4592 sys_columns.c.object_id == sys_index_columns.c.object_id,
4593 sys_columns.c.column_id == sys_index_columns.c.column_id,
4594 ),
4595 )
4596 .join(
4597 sys_objects,
4598 onclause=sys_objects.c.object_id
4599 == sys_index_columns.c.object_id,
4600 )
4601 .join(
4602 sys_schemas,
4603 onclause=sys_schemas.c.schema_id == sys_objects.c.schema_id,
4604 )
4605 .where(
4606 ischema.sys_schemas.c.name == sql.bindparam("owner"),
4607 ischema.sys_objects.c.name.in_(
4608 sql.bindparam("filter_names", expanding=True)
4609 ),
4610 )
4611 .order_by(
4612 sys_objects.c.name,
4613 sys_index_columns.c.index_id,
4614 sys_index_columns.c.key_ordinal,
4615 )
4616 )
4617 return s
4619 def _fetch_multi_indexes(
4620 self, connection, owner, names, schema, name_map, result, exec_opts
4621 ):
4622 """Execute the unified indexes queries for one catalog pass.
4624 Runs two Core sys.* selects (index metadata + index columns),
4625 grouped by table. Used for both the main-DB pass and the
4626 tempdb pass (with ``schema_translate_map={"sys": "tempdb.sys"}``
4627 applied).
4628 """
4629 meta_q = self._indexes_metadata_select()
4631 rp = connection.execute(
4632 meta_q,
4633 {"owner": owner, "filter_names": names},
4634 execution_options=exec_opts,
4635 )
4637 # {table_name: {index_id: index_dict}}
4638 indexes_by_table = {}
4639 for row in rp.mappings():
4640 tname = name_map(row["table_name"])
4641 if tname not in indexes_by_table:
4642 indexes_by_table[tname] = {}
4644 current = {
4645 "name": row["name"],
4646 "unique": row["is_unique"] == 1,
4647 "column_names": [],
4648 "dialect_options": {"mssql_include": []},
4649 }
4650 do = current["dialect_options"]
4651 index_type = row["type"]
4652 if index_type in {1, 2}:
4653 do["mssql_clustered"] = index_type == 1
4654 if index_type in {5, 6}:
4655 do["mssql_clustered"] = index_type == 5
4656 do["mssql_columnstore"] = True
4657 if row["filter_definition"] is not None:
4658 do["mssql_where"] = row["filter_definition"]
4660 indexes_by_table[tname][row["index_id"]] = current
4662 cols_q = self._indexes_columns_select()
4664 rp2 = connection.execute(
4665 cols_q,
4666 {"owner": owner, "filter_names": names},
4667 execution_options=exec_opts,
4668 )
4670 for row in rp2.mappings():
4671 tname = name_map(row["table_name"])
4672 idx_id = row["index_id"]
4673 if tname not in indexes_by_table:
4674 continue
4675 if idx_id not in indexes_by_table[tname]:
4676 continue
4677 index_def = indexes_by_table[tname][idx_id]
4678 do = index_def["dialect_options"]
4679 is_colstore = do.get("mssql_columnstore")
4680 is_clustered = do.get("mssql_clustered")
4681 if not (is_colstore and is_clustered):
4682 # a clustered columnstore index includes all columns but does
4683 # not want them in the index definition
4684 if row["is_included_column"] and not is_colstore:
4685 # a noncludsted columnstore index reports that includes
4686 # columns but requires that are listed as normal columns
4687 do["mssql_include"].append(row["name"])
4688 else:
4689 index_def["column_names"].append(row["name"])
4691 for tname, idx_dict in indexes_by_table.items():
4692 result[(schema, tname)] = list(idx_dict.values())
4694 for n in names:
4695 key = (schema, name_map(n))
4696 if key not in result:
4697 result[key] = ReflectionDefaults.indexes()
4699 def _fetch_multi_indexes_temp(
4700 self, connection, temp_names, schema, result
4701 ):
4702 """Run the unified indexes queries against tempdb."""
4703 resolved_by_owner, temp_name_map = self._resolve_temp_names(
4704 connection, temp_names
4705 )
4707 for temp_owner, mangled_names in resolved_by_owner.items():
4708 self._fetch_multi_indexes(
4709 connection,
4710 owner=temp_owner,
4711 names=mangled_names,
4712 schema=schema,
4713 name_map=temp_name_map,
4714 result=result,
4715 exec_opts={"schema_translate_map": {"sys": "tempdb.sys"}},
4716 )
4718 @_db_plus_owner_multi
4719 def get_multi_table_comment(
4720 self,
4721 connection,
4722 dbname,
4723 owner,
4724 schema,
4725 filter_names,
4726 scope,
4727 kind,
4728 **kw,
4729 ):
4730 if not self.supports_comments:
4731 raise NotImplementedError(
4732 "Can't get table comments on current SQL Server "
4733 "version in use"
4734 )
4736 regular_names, multi_object_names, temp_names = (
4737 self._partition_filter_names(
4738 connection, owner, filter_names, scope, kind
4739 )
4740 )
4742 result = {}
4744 if multi_object_names:
4745 name_map = self._multi_name_map(regular_names)
4746 self._fetch_multi_table_comment(
4747 connection,
4748 owner=owner,
4749 names=multi_object_names,
4750 schema=schema,
4751 name_map=name_map,
4752 result=result,
4753 exec_opts={"schema_translate_map": {}},
4754 )
4756 if temp_names:
4757 self._fetch_multi_table_comment_temp(
4758 connection,
4759 temp_names=temp_names,
4760 schema=schema,
4761 result=result,
4762 )
4764 return result.items()
4766 @lru_cache()
4767 def _table_comment_select(self):
4768 """Build the Core sys.* select for table comment reflection."""
4769 sys_objects = ischema.sys_objects
4770 sys_schemas = ischema.sys_schemas
4771 extended_properties = ischema.extended_properties
4773 s = (
4774 sql.select(
4775 sys_objects.c.name.label("table_name"),
4776 extended_properties.c.value.label("comment"),
4777 )
4778 .select_from(sys_objects)
4779 .join(
4780 sys_schemas,
4781 onclause=sys_objects.c.schema_id == sys_schemas.c.schema_id,
4782 )
4783 .outerjoin(
4784 extended_properties,
4785 onclause=sql.and_(
4786 extended_properties.c["class"] == 1,
4787 extended_properties.c.name == "MS_Description",
4788 extended_properties.c.major_id == sys_objects.c.object_id,
4789 extended_properties.c.minor_id == 0,
4790 ),
4791 )
4792 .where(
4793 sys_objects.c.type.in_(("U", "V")),
4794 ischema.sys_schemas.c.name == sql.bindparam("owner"),
4795 ischema.sys_objects.c.name.in_(
4796 sql.bindparam("filter_names", expanding=True)
4797 ),
4798 )
4799 )
4800 return s
4802 def _fetch_multi_table_comment(
4803 self, connection, owner, names, schema, name_map, result, exec_opts
4804 ):
4805 """Execute the unified table comment query for one catalog pass.
4807 Core sys.* select so ``schema_translate_map`` actually rewrites
4808 the catalog reference for the tempdb pass.
4809 """
4810 q = self._table_comment_select()
4812 rp = connection.execute(
4813 q,
4814 {"owner": owner, "filter_names": names},
4815 execution_options=exec_opts,
4816 )
4818 for row in rp.mappings():
4819 table_name = name_map(row["table_name"])
4820 comment = row["comment"]
4821 result[(schema, table_name)] = (
4822 {"text": comment} if comment else {"text": None}
4823 )
4825 for n in names:
4826 key = (schema, name_map(n))
4827 if key not in result:
4828 result[key] = ReflectionDefaults.table_comment()
4830 def _fetch_multi_table_comment_temp(
4831 self, connection, temp_names, schema, result
4832 ):
4833 """Run the unified table comment query against tempdb."""
4834 resolved_by_owner, temp_name_map = self._resolve_temp_names(
4835 connection, temp_names
4836 )
4837 for temp_owner, mangled_names in resolved_by_owner.items():
4838 self._fetch_multi_table_comment(
4839 connection,
4840 owner=temp_owner,
4841 names=mangled_names,
4842 schema=schema,
4843 name_map=temp_name_map,
4844 result=result,
4845 exec_opts={"schema_translate_map": {"sys": "tempdb.sys"}},
4846 )
4848 # override mixin methods for which this dialect has no native
4849 # get_multi_* implementation, preventing infinite recursion
4850 # through DefaultDialect._default_multi_reflect
4852 def get_unique_constraints(
4853 self, connection, table_name, schema=None, **kw
4854 ):
4855 raise NotImplementedError()
4857 def get_check_constraints(self, connection, table_name, schema=None, **kw):
4858 raise NotImplementedError()
4860 def get_table_options(self, connection, table_name, schema=None, **kw):
4861 raise NotImplementedError()