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 _get_internal_temp_table_name(self, connection, tablename):
3557 # it's likely that schema is always "dbo", but since we can
3558 # get it here, let's get it.
3559 # see https://stackoverflow.com/questions/8311959/
3560 # specifying-schema-for-temporary-tables
3562 # object_id() resolves a local temp table name only against the
3563 # current session's tables (plus any ##global one), so the lookup
3564 # neither sees nor locks other sessions' same-named temp tables in
3565 # tempdb, as a scan of tempdb's catalog by name would. See #13754.
3566 try:
3567 return connection.execute(
3568 sql.text(
3569 "select s.name, o.name "
3570 "from tempdb.sys.objects as o "
3571 "join tempdb.sys.schemas as s "
3572 "on s.schema_id = o.schema_id "
3573 "where o.object_id = object_id(:qualified, 'U')"
3574 ),
3575 {"qualified": f"tempdb..[{tablename}]"},
3576 ).one()
3577 except exc.NoResultFound as ne:
3578 raise exc.NoSuchTableError(
3579 "Unable to find a temporary table named '%s' in tempdb."
3580 % tablename
3581 ) from ne
3583 def _parse_column_info(
3584 self,
3585 name,
3586 type_,
3587 nullable,
3588 maxlen,
3589 numericprec,
3590 numericscale,
3591 default,
3592 collation,
3593 definition,
3594 is_persisted,
3595 is_identity,
3596 identity_start,
3597 identity_increment,
3598 comment,
3599 base_type=None,
3600 ):
3601 # Try to resolve the user type first (e.g., "sysname"),
3602 # then fall back to the base type (e.g., "nvarchar").
3603 # base_type may be None for CLR types (geography, geometry,
3604 # hierarchyid) which have no corresponding base type.
3605 coltype = self.ischema_names.get(type_, None)
3606 if coltype is None and base_type is not None and base_type != type_:
3607 coltype = self.ischema_names.get(base_type, None)
3608 kwargs = {}
3610 if coltype in (MSBinary, MSVarBinary, sqltypes.LargeBinary):
3611 kwargs["length"] = maxlen if maxlen != -1 else None
3612 elif coltype in (MSString, MSChar):
3613 kwargs["length"] = maxlen if maxlen != -1 else None
3614 if collation:
3615 kwargs["collation"] = collation
3616 elif coltype in (MSNVarchar, MSNChar):
3617 kwargs["length"] = maxlen // 2 if maxlen != -1 else None
3618 if collation:
3619 kwargs["collation"] = collation
3620 elif coltype in (MSText, MSNText):
3621 # TEXT / NTEXT are unlengthed LOB types. sys.columns.max_length
3622 # reports 16 for these, which is the size of the in-row LOB
3623 # pointer and not a character length, so no length is applied.
3624 if collation:
3625 kwargs["collation"] = collation
3627 if coltype is None:
3628 if base_type is not None and base_type != type_:
3629 util.warn(
3630 "Did not recognize type '%s' (user type) or '%s' "
3631 "(base type) of column '%s'" % (type_, base_type, name)
3632 )
3633 else:
3634 util.warn(
3635 "Did not recognize type '%s' of column '%s'"
3636 % (type_, name)
3637 )
3638 coltype = sqltypes.NULLTYPE
3639 else:
3640 if issubclass(coltype, sqltypes.NumericCommon):
3641 kwargs["precision"] = numericprec
3642 if not issubclass(coltype, sqltypes.Float):
3643 kwargs["scale"] = numericscale
3644 coltype = coltype(**kwargs)
3646 cdict = {
3647 "name": name,
3648 "type": coltype,
3649 "nullable": nullable,
3650 "default": default,
3651 "autoincrement": is_identity is not None,
3652 "comment": comment,
3653 }
3655 if definition is not None and is_persisted is not None:
3656 cdict["computed"] = {
3657 "sqltext": definition,
3658 "persisted": is_persisted,
3659 }
3661 if is_identity is not None:
3662 if identity_start is None or identity_increment is None:
3663 cdict["identity"] = {}
3664 else:
3665 if isinstance(
3666 coltype, (sqltypes.BigInteger, sqltypes.Integer)
3667 ):
3668 start = int(identity_start)
3669 increment = int(identity_increment)
3670 else:
3671 start = identity_start
3672 increment = identity_increment
3673 cdict["identity"] = {
3674 "start": start,
3675 "increment": increment,
3676 }
3678 return cdict
3680 @lru_cache()
3681 def _columns_select(self):
3682 """Build the unified Core sys.* select for column reflection.
3684 Returns a ``select()`` that includes ``table_name`` and ``owner``
3685 in the result so callers can group rows by table.
3687 Used by :meth:`.get_multi_columns` (and indirectly by
3688 :meth:`.get_columns` via the multi delegation).
3689 """
3690 sys_columns = ischema.sys_columns
3691 sys_types = ischema.sys_types
3692 sys_base_types = ischema.sys_types.alias("base_types")
3693 sys_default_constraints = ischema.sys_default_constraints
3694 computed_cols = ischema.computed_columns
3695 identity_cols = ischema.identity_columns
3696 extended_properties = ischema.extended_properties
3697 sys_objects = ischema.sys_objects
3698 sys_schemas = ischema.sys_schemas
3700 if self._supports_nvarchar_max:
3701 computed_definition = computed_cols.c.definition
3702 else:
3703 # tds_version 4.2 does not support NVARCHAR(MAX)
3704 computed_definition = sql.cast(
3705 computed_cols.c.definition, NVARCHAR(4000)
3706 )
3708 s = (
3709 sql.select(
3710 sys_objects.c.name.label("table_name"),
3711 sys_schemas.c.name.label("owner"),
3712 sys_columns.c.name.label("column_name"),
3713 sys_types.c.name.label("type_name"),
3714 sys_base_types.c.name.label("base_type"),
3715 sys_columns.c.is_nullable,
3716 sys_columns.c.max_length,
3717 sys_columns.c.precision,
3718 sys_columns.c.scale,
3719 sys_default_constraints.c.definition.label("default_value"),
3720 sys_columns.c.collation_name,
3721 computed_definition.label("computed_definition"),
3722 computed_cols.c.is_persisted,
3723 identity_cols.c.is_identity,
3724 identity_cols.c.seed_value,
3725 identity_cols.c.increment_value,
3726 extended_properties.c.value.label("comment"),
3727 sys_columns.c.column_id,
3728 )
3729 .select_from(sys_columns)
3730 .join(
3731 sys_objects,
3732 onclause=sys_columns.c.object_id == sys_objects.c.object_id,
3733 )
3734 .join(
3735 sys_schemas,
3736 onclause=sys_objects.c.schema_id == sys_schemas.c.schema_id,
3737 )
3738 .join(
3739 sys_types,
3740 onclause=sys_columns.c.user_type_id
3741 == sys_types.c.user_type_id,
3742 )
3743 .outerjoin(
3744 sys_base_types,
3745 onclause=sql.and_(
3746 sys_types.c.system_type_id
3747 == sys_base_types.c.system_type_id,
3748 sys_base_types.c.user_type_id
3749 == sys_base_types.c.system_type_id,
3750 ),
3751 )
3752 .outerjoin(
3753 sys_default_constraints,
3754 sql.and_(
3755 sys_default_constraints.c.object_id
3756 == sys_columns.c.default_object_id,
3757 sys_default_constraints.c.parent_column_id
3758 == sys_columns.c.column_id,
3759 ),
3760 )
3761 .outerjoin(
3762 computed_cols,
3763 onclause=sql.and_(
3764 computed_cols.c.object_id == sys_columns.c.object_id,
3765 computed_cols.c.column_id == sys_columns.c.column_id,
3766 ),
3767 )
3768 .outerjoin(
3769 identity_cols,
3770 onclause=sql.and_(
3771 identity_cols.c.object_id == sys_columns.c.object_id,
3772 identity_cols.c.column_id == sys_columns.c.column_id,
3773 ),
3774 )
3775 .outerjoin(
3776 extended_properties,
3777 onclause=sql.and_(
3778 extended_properties.c["class"] == 1,
3779 extended_properties.c.name == "MS_Description",
3780 sys_columns.c.object_id == extended_properties.c.major_id,
3781 sys_columns.c.column_id == extended_properties.c.minor_id,
3782 ),
3783 )
3784 .where(
3785 ischema.sys_schemas.c.name == sql.bindparam("owner"),
3786 ischema.sys_objects.c.name.in_(
3787 sql.bindparam("filter_names", expanding=True)
3788 ),
3789 )
3790 .order_by(sys_objects.c.name, sys_columns.c.column_id)
3791 )
3792 return s
3794 @staticmethod
3795 def _fk_query_sql(fk_info_where, extra_cols=""):
3796 return """\
3797WITH fk_info AS (
3798 SELECT
3799 ischema_ref_con.constraint_schema,
3800 ischema_ref_con.constraint_name,
3801 ischema_key_col.ordinal_position,
3802 ischema_key_col.table_schema,
3803 ischema_key_col.table_name,
3804 ischema_ref_con.unique_constraint_schema,
3805 ischema_ref_con.unique_constraint_name,
3806 ischema_ref_con.match_option,
3807 ischema_ref_con.update_rule,
3808 ischema_ref_con.delete_rule,
3809 ischema_key_col.column_name AS constrained_column
3810 FROM
3811 INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS ischema_ref_con
3812 INNER JOIN
3813 INFORMATION_SCHEMA.KEY_COLUMN_USAGE ischema_key_col ON
3814 ischema_key_col.table_schema =
3815 ischema_ref_con.constraint_schema
3816 AND ischema_key_col.constraint_name =
3817 ischema_ref_con.constraint_name
3818 WHERE %(fk_info_where)s
3819),
3820constraint_info AS (
3821 SELECT
3822 ischema_key_col.constraint_schema,
3823 ischema_key_col.constraint_name,
3824 ischema_key_col.ordinal_position,
3825 ischema_key_col.table_schema,
3826 ischema_key_col.table_name,
3827 ischema_key_col.column_name
3828 FROM
3829 INFORMATION_SCHEMA.KEY_COLUMN_USAGE ischema_key_col
3830),
3831index_info AS (
3832 SELECT
3833 sys.schemas.name AS index_schema,
3834 sys.indexes.name AS index_name,
3835 sys.index_columns.key_ordinal AS ordinal_position,
3836 sys.schemas.name AS table_schema,
3837 sys.objects.name AS table_name,
3838 sys.columns.name AS column_name
3839 FROM
3840 sys.indexes
3841 INNER JOIN sys.objects
3842 ON sys.objects.object_id = sys.indexes.object_id
3843 INNER JOIN sys.schemas
3844 ON sys.schemas.schema_id = sys.objects.schema_id
3845 INNER JOIN sys.index_columns
3846 ON sys.index_columns.object_id = sys.objects.object_id
3847 AND sys.index_columns.index_id = sys.indexes.index_id
3848 INNER JOIN sys.columns
3849 ON sys.columns.object_id = sys.indexes.object_id
3850 AND sys.columns.column_id = sys.index_columns.column_id
3851)
3852 SELECT
3853 %(extra_cols)s
3854 fk_info.constraint_schema,
3855 fk_info.constraint_name,
3856 fk_info.ordinal_position,
3857 fk_info.constrained_column,
3858 constraint_info.table_schema AS referred_table_schema,
3859 constraint_info.table_name AS referred_table_name,
3860 constraint_info.column_name AS referred_column,
3861 fk_info.match_option,
3862 fk_info.update_rule,
3863 fk_info.delete_rule
3864 FROM
3865 fk_info INNER JOIN constraint_info ON
3866 constraint_info.constraint_schema =
3867 fk_info.unique_constraint_schema
3868 AND constraint_info.constraint_name =
3869 fk_info.unique_constraint_name
3870 AND constraint_info.ordinal_position = fk_info.ordinal_position
3871 UNION
3872 SELECT
3873 %(extra_cols)s
3874 fk_info.constraint_schema,
3875 fk_info.constraint_name,
3876 fk_info.ordinal_position,
3877 fk_info.constrained_column,
3878 index_info.table_schema AS referred_table_schema,
3879 index_info.table_name AS referred_table_name,
3880 index_info.column_name AS referred_column,
3881 fk_info.match_option,
3882 fk_info.update_rule,
3883 fk_info.delete_rule
3884 FROM
3885 fk_info INNER JOIN index_info ON
3886 index_info.index_schema = fk_info.unique_constraint_schema
3887 AND index_info.index_name = fk_info.unique_constraint_name
3888 AND index_info.ordinal_position = fk_info.ordinal_position
3889 AND NOT (index_info.table_schema = fk_info.table_schema
3890 AND index_info.table_name = fk_info.table_name)
3892 ORDER BY fk_info.constraint_schema, fk_info.constraint_name,
3893 fk_info.ordinal_position
3894""" % {
3895 "fk_info_where": fk_info_where,
3896 "extra_cols": extra_cols,
3897 }
3899 # --- multi-reflection API ---
3901 @lru_cache
3902 def _partition_filter_names_query(self, add_filter_names: bool):
3903 sys_objects = ischema.sys_objects
3904 sys_schemas = ischema.sys_schemas
3906 s = (
3907 sql.select(sys_objects.c.name)
3908 .select_from(sys_objects)
3909 .join(
3910 sys_schemas,
3911 onclause=sys_objects.c.schema_id == sys_schemas.c.schema_id,
3912 )
3913 .where(
3914 sys_schemas.c.name == sql.bindparam("schema"),
3915 sys_objects.c.type.in_(
3916 sql.bindparam("type_filter", expanding=True)
3917 ),
3918 )
3919 )
3920 if add_filter_names:
3921 s = s.where(
3922 sys_objects.c.name.in_(
3923 sql.bindparam("filter_names", expanding=True)
3924 )
3925 )
3927 return s
3929 def _partition_filter_names(
3930 self, connection, owner, filter_names, scope, kind
3931 ):
3932 """Split caller-supplied ``filter_names`` between the bulk
3933 ``sys.*`` query path (regular objects in the user database) and
3934 the temp object path (objects living in ``tempdb``), and resolve
3935 the existing-in-catalog names for the bulk path in one shot.
3937 Returns the tuple
3938 ``(regular_names, multi_object_names, temp_names)``:
3940 * ``regular_names`` -- list of caller-supplied names (with the
3941 caller's casing) for the bulk path, used to build the
3942 case-insensitive name map. Empty list when no regular names
3943 were requested.
3944 * ``multi_object_names`` -- list of names (with server-side
3945 casing) that actually exist in the catalog and match the
3946 requested ``kind``. This is what the bulk SQL's IN clause
3947 gets filtered with. Empty list when no matches, disabling
3948 the bulk query.
3949 * ``temp_names`` -- list of ``#``-prefixed temp object names
3950 that the caller asked for and that the requested
3951 ``scope``/``kind`` allows.
3953 ``scope=ObjectScope.DEFAULT`` excludes temp objects entirely.
3954 ``scope=ObjectScope.TEMPORARY`` excludes regular objects.
3955 ``ObjectKind.VIEW`` (without TABLE) excludes temp because SQL
3956 Server does not have temp views in the same form. SQL Server
3957 also has no ``get_temp_table_names`` implementation, so temp
3958 objects can only be reflected when the caller supplies their
3959 names explicitly.
3960 """
3961 include_default = scope is not ObjectScope.TEMPORARY
3962 include_temp = (
3963 scope is not ObjectScope.DEFAULT and ObjectKind.TABLE in kind
3964 )
3966 if filter_names:
3967 if include_temp:
3968 temp_names = [n for n in filter_names if n.startswith("#")]
3969 else:
3970 temp_names = []
3972 if include_default:
3973 regular_names = [
3974 n for n in filter_names if not n.startswith("#")
3975 ]
3976 run_bulk = bool(regular_names)
3977 else:
3978 regular_names = []
3979 run_bulk = False
3980 else:
3981 temp_names = []
3982 regular_names = []
3983 run_bulk = include_default
3985 if not run_bulk:
3986 return (regular_names, [], temp_names)
3988 type_filter = []
3989 if ObjectKind.TABLE in kind:
3990 type_filter.append("U")
3991 if ObjectKind.VIEW in kind:
3992 type_filter.append("V")
3993 # SQL Server does not support materialized views, so ignore them
3994 if not type_filter:
3995 return (regular_names, [], temp_names)
3997 has_regular_names = bool(regular_names)
3998 query = self._partition_filter_names_query(has_regular_names)
3999 params = {"schema": owner, "type_filter": type_filter}
4000 if has_regular_names:
4001 params["filter_names"] = regular_names
4003 multi_object_names = connection.scalars(query, params).all()
4005 return (regular_names, multi_object_names, temp_names)
4007 @staticmethod
4008 def _multi_name_map(filter_names):
4009 """Build a case-insensitive ``server_name -> user_name`` mapper.
4011 MSSQL object names are case-insensitive under the default
4012 collation, so a user may pass ``"sOmEtAbLe"`` for a table
4013 physically stored as ``"SomeTable"``. The bulk SQL returns the
4014 server-side casing, but result keys must use the user-supplied
4015 casing so that downstream code (e.g.
4016 :meth:`.Inspector.reflect_table`) can find the entry.
4018 Returns a callable ``(server_name) -> user_name``. When
4019 ``filter_names`` is empty/None, returns identity.
4020 """
4021 if not filter_names:
4022 return lambda n: n
4023 lookup = {n.lower(): n for n in filter_names}
4024 return lambda n: lookup.get(n.lower(), n)
4026 @_db_plus_owner_multi
4027 def get_multi_columns(
4028 self,
4029 connection,
4030 dbname,
4031 owner,
4032 schema,
4033 filter_names,
4034 scope,
4035 kind,
4036 **kw,
4037 ):
4038 regular_names, multi_object_names, temp_names = (
4039 self._partition_filter_names(
4040 connection, owner, filter_names, scope, kind
4041 )
4042 )
4044 result = {}
4046 if multi_object_names:
4047 name_map = self._multi_name_map(regular_names)
4048 self._fetch_multi_columns(
4049 connection,
4050 owner=owner,
4051 names=multi_object_names,
4052 schema=schema,
4053 name_map=name_map,
4054 result=result,
4055 exec_opts={"schema_translate_map": {}},
4056 )
4058 if temp_names:
4059 self._fetch_multi_columns_temp(
4060 connection,
4061 temp_names=temp_names,
4062 schema=schema,
4063 result=result,
4064 )
4066 return result.items()
4068 def _fetch_multi_columns(
4069 self, connection, owner, names, schema, name_map, result, exec_opts
4070 ):
4071 """Execute the unified columns select for a single catalog pass.
4073 Used by :meth:`.get_multi_columns` for both the main-database pass
4074 (no ``schema_translate_map``) and the tempdb pass (with
4075 ``schema_translate_map={"sys": "tempdb.sys"}`` applied).
4076 """
4077 s = self._columns_select()
4079 rp = connection.execute(
4080 s,
4081 {"owner": owner, "filter_names": names},
4082 execution_options=exec_opts,
4083 )
4085 for row in rp.mappings():
4086 table_name = name_map(row["table_name"])
4087 cdict = self._parse_column_info(
4088 name=row["column_name"],
4089 type_=row["type_name"],
4090 base_type=row["base_type"],
4091 nullable=row["is_nullable"] == 1,
4092 maxlen=row["max_length"],
4093 numericprec=row["precision"],
4094 numericscale=row["scale"],
4095 default=row["default_value"],
4096 collation=row["collation_name"],
4097 definition=row["computed_definition"],
4098 is_persisted=row["is_persisted"],
4099 is_identity=row["is_identity"],
4100 identity_start=row["seed_value"],
4101 identity_increment=row["increment_value"],
4102 comment=row["comment"],
4103 )
4104 result.setdefault((schema, table_name), []).append(cdict)
4106 for n in names:
4107 key = (schema, name_map(n))
4108 if key not in result:
4109 result[key] = ReflectionDefaults.columns()
4111 def _fetch_multi_columns_temp(
4112 self, connection, temp_names, schema, result
4113 ):
4114 """Run the unified columns select against tempdb for temp names.
4116 Resolves each ``#name`` to its mangled tempdb name (one round
4117 trip per temp via :meth:`._get_internal_temp_table_name`), then
4118 runs a single bulk select against ``tempdb.sys.*`` via
4119 ``schema_translate_map``. The result rows (keyed by mangled
4120 name) are mapped back to the original ``#name`` for the caller.
4121 """
4122 resolved_by_owner, temp_name_map = self._resolve_temp_names(
4123 connection, temp_names
4124 )
4125 for temp_owner, mangled_names in resolved_by_owner.items():
4126 self._fetch_multi_columns(
4127 connection,
4128 owner=temp_owner,
4129 names=mangled_names,
4130 schema=schema,
4131 name_map=temp_name_map,
4132 result=result,
4133 exec_opts={"schema_translate_map": {"sys": "tempdb.sys"}},
4134 )
4136 def _resolve_temp_names(self, connection, temp_names):
4137 """Resolve user-facing temp names (``#foo``) to their mangled
4138 tempdb names, grouped by owner.
4140 Returns ``(resolved_by_owner, temp_name_map)``:
4142 * ``resolved_by_owner`` -- ``{owner: [mangled_name, ...]}``,
4143 one bulk query per owner (typically just ``'dbo'``).
4144 * ``temp_name_map`` -- a function that maps mangled names back
4145 to the user-supplied name.
4147 Names that don't resolve (no matching temp table in tempdb) are
4148 silently dropped. The single-table reflection wrappers handle
4149 the resulting absence by raising :exc:`.NoSuchTableError` via
4150 :meth:`._value_or_raise`.
4151 """
4152 resolved_by_owner = {}
4153 original_by_mangled = {}
4154 for name in temp_names:
4155 try:
4156 temp_owner, mangled = self._get_internal_temp_table_name(
4157 connection, name
4158 )
4159 except exc.NoSuchTableError:
4160 continue
4161 resolved_by_owner.setdefault(temp_owner, []).append(mangled)
4162 original_by_mangled[mangled] = name
4164 def temp_name_map(server_name):
4165 return original_by_mangled.get(server_name, server_name)
4167 return resolved_by_owner, temp_name_map
4169 @_db_plus_owner_multi
4170 def get_multi_pk_constraint(
4171 self,
4172 connection,
4173 dbname,
4174 owner,
4175 schema,
4176 filter_names,
4177 scope,
4178 kind,
4179 **kw,
4180 ):
4181 regular_names, multi_object_names, temp_names = (
4182 self._partition_filter_names(
4183 connection, owner, filter_names, scope, kind
4184 )
4185 )
4187 result = {}
4189 if multi_object_names:
4190 name_map = self._multi_name_map(regular_names)
4191 self._fetch_multi_pk_constraint(
4192 connection,
4193 owner=owner,
4194 names=multi_object_names,
4195 schema=schema,
4196 name_map=name_map,
4197 result=result,
4198 exec_opts={"schema_translate_map": {}},
4199 )
4201 if temp_names:
4202 self._fetch_multi_pk_constraint_temp(
4203 connection,
4204 temp_names=temp_names,
4205 schema=schema,
4206 result=result,
4207 )
4209 return result.items()
4211 @lru_cache()
4212 def _pk_constraint_select(self):
4213 """Build the unified Core sys.* select for PK constraint reflection.
4215 Returns a ``select()`` that includes ``table_name`` and ``owner``
4216 in the result so callers can group rows by table. The clustered
4217 flag comes directly from ``sys.indexes.type`` (1 = clustered).
4218 """
4219 sys_key_constraints = ischema.sys_key_constraints
4220 sys_indexes = ischema.sys_indexes
4221 sys_index_columns = ischema.sys_index_columns
4222 sys_columns = ischema.sys_columns
4223 sys_objects = ischema.sys_objects
4224 sys_schemas = ischema.sys_schemas
4226 s = (
4227 sql.select(
4228 sys_objects.c.name.label("table_name"),
4229 sys_schemas.c.name.label("owner"),
4230 sys_key_constraints.c.name.label("constraint_name"),
4231 sys_columns.c.name.label("column_name"),
4232 sys_index_columns.c.key_ordinal,
4233 sys_indexes.c.type.label("index_type"),
4234 )
4235 .select_from(sys_key_constraints)
4236 .join(
4237 sys_objects,
4238 onclause=sys_key_constraints.c.parent_object_id
4239 == sys_objects.c.object_id,
4240 )
4241 .join(
4242 sys_schemas,
4243 onclause=sys_objects.c.schema_id == sys_schemas.c.schema_id,
4244 )
4245 .join(
4246 sys_index_columns,
4247 onclause=sql.and_(
4248 sys_index_columns.c.object_id
4249 == sys_key_constraints.c.parent_object_id,
4250 sys_index_columns.c.index_id
4251 == sys_key_constraints.c.unique_index_id,
4252 ),
4253 )
4254 .join(
4255 sys_columns,
4256 onclause=sql.and_(
4257 sys_columns.c.object_id == sys_index_columns.c.object_id,
4258 sys_columns.c.column_id == sys_index_columns.c.column_id,
4259 ),
4260 )
4261 .join(
4262 sys_indexes,
4263 onclause=sql.and_(
4264 sys_indexes.c.object_id
4265 == sys_key_constraints.c.parent_object_id,
4266 sys_indexes.c.index_id
4267 == sys_key_constraints.c.unique_index_id,
4268 ),
4269 )
4270 .where(
4271 sys_key_constraints.c.type == "PK",
4272 ischema.sys_schemas.c.name == sql.bindparam("owner"),
4273 ischema.sys_objects.c.name.in_(
4274 sql.bindparam("filter_names", expanding=True)
4275 ),
4276 )
4277 .order_by(sys_objects.c.name, sys_index_columns.c.key_ordinal)
4278 )
4279 return s
4281 def _fetch_multi_pk_constraint(
4282 self, connection, owner, names, schema, name_map, result, exec_opts
4283 ):
4284 """Execute the unified pk_constraint select for one catalog pass.
4286 Used by :meth:`.get_multi_pk_constraint` for both the main DB
4287 pass and the tempdb pass (with ``schema_translate_map`` set).
4288 """
4289 s = self._pk_constraint_select()
4291 rp = connection.execute(
4292 s,
4293 {"owner": owner, "filter_names": names},
4294 execution_options=exec_opts,
4295 )
4297 for row in rp.mappings():
4298 table_name = name_map(row["table_name"])
4299 key = (schema, table_name)
4300 if key not in result:
4301 result[key] = {
4302 "constrained_columns": [],
4303 "name": row["constraint_name"],
4304 "dialect_options": {
4305 "mssql_clustered": row["index_type"] == 1
4306 },
4307 }
4308 result[key]["constrained_columns"].append(row["column_name"])
4310 for n in names:
4311 key = (schema, name_map(n))
4312 if key not in result:
4313 result[key] = ReflectionDefaults.pk_constraint()
4315 def _fetch_multi_pk_constraint_temp(
4316 self, connection, temp_names, schema, result
4317 ):
4318 """Run the unified pk_constraint select against tempdb."""
4319 resolved_by_owner, temp_name_map = self._resolve_temp_names(
4320 connection, temp_names
4321 )
4323 for temp_owner, mangled_names in resolved_by_owner.items():
4324 self._fetch_multi_pk_constraint(
4325 connection,
4326 owner=temp_owner,
4327 names=mangled_names,
4328 schema=schema,
4329 name_map=temp_name_map,
4330 result=result,
4331 exec_opts={"schema_translate_map": {"sys": "tempdb.sys"}},
4332 )
4334 @_db_plus_owner_multi
4335 def get_multi_foreign_keys(
4336 self,
4337 connection,
4338 dbname,
4339 owner,
4340 schema,
4341 filter_names,
4342 scope,
4343 kind,
4344 **kw,
4345 ):
4346 regular_names, multi_object_names, temp_names = (
4347 self._partition_filter_names(
4348 connection, owner, filter_names, scope, kind
4349 )
4350 )
4352 final = {}
4354 if multi_object_names:
4355 name_map = self._multi_name_map(regular_names)
4356 self._fetch_multi_foreign_keys(
4357 connection,
4358 dbname=dbname,
4359 owner=owner,
4360 names=multi_object_names,
4361 schema=schema,
4362 name_map=name_map,
4363 final=final,
4364 )
4366 # FK reflection for temp tables: the INFORMATION_SCHEMA query in
4367 # _fk_query_sql does not see tempdb objects (it queries the main
4368 # DB's catalog). The pre-existing single-table behavior was to
4369 # return empty foreign_keys for temp tables that exist. We
4370 # preserve that here by checking existence and returning the
4371 # default (empty) reflection for each temp that is reachable.
4372 has_temp_tables = self._internal_has_multi_table(
4373 connection, temp_names, owner="dbo", schema=schema, **kw
4374 )
4375 for name in temp_names:
4376 key = (schema, name)
4377 if has_temp_tables.get(key):
4378 final[key] = ReflectionDefaults.foreign_keys()
4380 return final.items()
4382 def _fetch_multi_foreign_keys(
4383 self, connection, dbname, owner, names, schema, name_map, final
4384 ):
4385 """Execute the unified FK query for the main-DB pass.
4387 Uses the _fk_query_sql sql text, restricted to the provided owner and
4388 IN-list of names. Groups result rows by table and constraint
4389 name into the final dict.
4390 """
4391 rp = connection.execute(
4392 sql.text(
4393 self._fk_query_sql(
4394 fk_info_where=(
4395 "ischema_key_col.table_schema = :owner"
4396 "\n AND ischema_key_col.table_name"
4397 " IN :filter_names"
4398 ),
4399 extra_cols="fk_info.table_name,",
4400 )
4401 )
4402 .bindparams(
4403 sql.bindparam("owner", owner, ischema.CoerceUnicode()),
4404 sql.bindparam("filter_names", names, expanding=True),
4405 )
4406 .columns(
4407 constraint_schema=sqltypes.Unicode(),
4408 constraint_name=sqltypes.Unicode(),
4409 table_name=sqltypes.Unicode(),
4410 constrained_column=sqltypes.Unicode(),
4411 referred_table_schema=sqltypes.Unicode(),
4412 referred_table_name=sqltypes.Unicode(),
4413 referred_column=sqltypes.Unicode(),
4414 )
4415 )
4417 grouped = {}
4418 for r in rp.all():
4419 (
4420 table_name,
4421 _, # constraint_schema
4422 rfknm,
4423 _, # ordinal
4424 scol,
4425 rschema,
4426 rtbl,
4427 rcol,
4428 _, # match
4429 fkuprule,
4430 fkdelrule,
4431 ) = r
4433 key = (schema, name_map(table_name))
4434 if key not in grouped:
4435 grouped[key] = util.defaultdict(
4436 lambda: {
4437 "name": None,
4438 "constrained_columns": [],
4439 "referred_schema": None,
4440 "referred_table": None,
4441 "referred_columns": [],
4442 "options": {},
4443 }
4444 )
4446 rec = grouped[key][rfknm]
4447 rec["name"] = rfknm
4449 if fkuprule != "NO ACTION":
4450 rec["options"]["onupdate"] = fkuprule
4451 if fkdelrule != "NO ACTION":
4452 rec["options"]["ondelete"] = fkdelrule
4454 if not rec["referred_table"]:
4455 rec["referred_table"] = rtbl
4456 if schema is not None or owner != rschema:
4457 if dbname:
4458 rschema = dbname + "." + rschema
4459 rec["referred_schema"] = rschema
4461 rec["constrained_columns"].append(scol)
4462 rec["referred_columns"].append(rcol)
4464 for key, fk_dict in grouped.items():
4465 final[key] = list(fk_dict.values())
4467 for n in names:
4468 key = (schema, name_map(n))
4469 if key not in final:
4470 final[key] = ReflectionDefaults.foreign_keys()
4472 @_db_plus_owner_multi
4473 def get_multi_indexes(
4474 self,
4475 connection,
4476 dbname,
4477 owner,
4478 schema,
4479 filter_names,
4480 scope,
4481 kind,
4482 **kw,
4483 ):
4484 regular_names, multi_object_names, temp_names = (
4485 self._partition_filter_names(
4486 connection, owner, filter_names, scope, kind
4487 )
4488 )
4490 result = {}
4492 if multi_object_names:
4493 name_map = self._multi_name_map(regular_names)
4494 self._fetch_multi_indexes(
4495 connection,
4496 owner=owner,
4497 names=multi_object_names,
4498 schema=schema,
4499 name_map=name_map,
4500 result=result,
4501 exec_opts={"schema_translate_map": {}},
4502 )
4504 if temp_names:
4505 self._fetch_multi_indexes_temp(
4506 connection,
4507 temp_names=temp_names,
4508 schema=schema,
4509 result=result,
4510 )
4512 return result.items()
4514 @lru_cache()
4515 def _indexes_metadata_select(self):
4516 """Build the Core sys.* select for index metadata (one row per
4517 index per table).
4519 Used by :meth:`._fetch_multi_indexes` for both the main-DB pass
4520 and the tempdb pass (via ``schema_translate_map``). Replaces a
4521 previous ``sql.text()`` body that did not honor
4522 ``schema_translate_map`` (which only rewrites Core
4523 schema-bearing objects, not literal SQL text).
4524 """
4525 sys_indexes = ischema.sys_indexes
4526 sys_objects = ischema.sys_objects
4527 sys_schemas = ischema.sys_schemas
4529 if self.server_version_info >= MS_2008_VERSION:
4530 filter_definition = sys_indexes.c.filter_definition
4531 else:
4532 filter_definition = sql.null()
4534 s = (
4535 sql.select(
4536 sys_objects.c.name.label("table_name"),
4537 sys_indexes.c.index_id,
4538 sys_indexes.c.is_unique,
4539 sys_indexes.c.name,
4540 sys_indexes.c.type,
4541 filter_definition.label("filter_definition"),
4542 )
4543 .select_from(sys_indexes)
4544 .join(
4545 sys_objects,
4546 onclause=sys_indexes.c.object_id == sys_objects.c.object_id,
4547 )
4548 .join(
4549 sys_schemas,
4550 onclause=sys_schemas.c.schema_id == sys_objects.c.schema_id,
4551 )
4552 .where(
4553 sys_indexes.c.is_primary_key == 0,
4554 sys_indexes.c.type != 0,
4555 ischema.sys_schemas.c.name == sql.bindparam("owner"),
4556 ischema.sys_objects.c.name.in_(
4557 sql.bindparam("filter_names", expanding=True)
4558 ),
4559 )
4560 .order_by(sys_objects.c.name, sys_indexes.c.name)
4561 )
4562 return s
4564 @lru_cache()
4565 def _indexes_columns_select(self):
4566 """Build the Core sys.* select for index columns (one row per
4567 index column per index per table)."""
4568 sys_index_columns = ischema.sys_index_columns
4569 sys_columns = ischema.sys_columns
4570 sys_objects = ischema.sys_objects
4571 sys_schemas = ischema.sys_schemas
4573 s = (
4574 sql.select(
4575 sys_objects.c.name.label("table_name"),
4576 sys_index_columns.c.index_id,
4577 sys_columns.c.name,
4578 sys_index_columns.c.is_included_column,
4579 )
4580 .select_from(sys_index_columns)
4581 .join(
4582 sys_columns,
4583 onclause=sql.and_(
4584 sys_columns.c.object_id == sys_index_columns.c.object_id,
4585 sys_columns.c.column_id == sys_index_columns.c.column_id,
4586 ),
4587 )
4588 .join(
4589 sys_objects,
4590 onclause=sys_objects.c.object_id
4591 == sys_index_columns.c.object_id,
4592 )
4593 .join(
4594 sys_schemas,
4595 onclause=sys_schemas.c.schema_id == sys_objects.c.schema_id,
4596 )
4597 .where(
4598 ischema.sys_schemas.c.name == sql.bindparam("owner"),
4599 ischema.sys_objects.c.name.in_(
4600 sql.bindparam("filter_names", expanding=True)
4601 ),
4602 )
4603 .order_by(
4604 sys_objects.c.name,
4605 sys_index_columns.c.index_id,
4606 sys_index_columns.c.key_ordinal,
4607 )
4608 )
4609 return s
4611 def _fetch_multi_indexes(
4612 self, connection, owner, names, schema, name_map, result, exec_opts
4613 ):
4614 """Execute the unified indexes queries for one catalog pass.
4616 Runs two Core sys.* selects (index metadata + index columns),
4617 grouped by table. Used for both the main-DB pass and the
4618 tempdb pass (with ``schema_translate_map={"sys": "tempdb.sys"}``
4619 applied).
4620 """
4621 meta_q = self._indexes_metadata_select()
4623 rp = connection.execute(
4624 meta_q,
4625 {"owner": owner, "filter_names": names},
4626 execution_options=exec_opts,
4627 )
4629 # {table_name: {index_id: index_dict}}
4630 indexes_by_table = {}
4631 for row in rp.mappings():
4632 tname = name_map(row["table_name"])
4633 if tname not in indexes_by_table:
4634 indexes_by_table[tname] = {}
4636 current = {
4637 "name": row["name"],
4638 "unique": row["is_unique"] == 1,
4639 "column_names": [],
4640 "dialect_options": {"mssql_include": []},
4641 }
4642 do = current["dialect_options"]
4643 index_type = row["type"]
4644 if index_type in {1, 2}:
4645 do["mssql_clustered"] = index_type == 1
4646 if index_type in {5, 6}:
4647 do["mssql_clustered"] = index_type == 5
4648 do["mssql_columnstore"] = True
4649 if row["filter_definition"] is not None:
4650 do["mssql_where"] = row["filter_definition"]
4652 indexes_by_table[tname][row["index_id"]] = current
4654 cols_q = self._indexes_columns_select()
4656 rp2 = connection.execute(
4657 cols_q,
4658 {"owner": owner, "filter_names": names},
4659 execution_options=exec_opts,
4660 )
4662 for row in rp2.mappings():
4663 tname = name_map(row["table_name"])
4664 idx_id = row["index_id"]
4665 if tname not in indexes_by_table:
4666 continue
4667 if idx_id not in indexes_by_table[tname]:
4668 continue
4669 index_def = indexes_by_table[tname][idx_id]
4670 do = index_def["dialect_options"]
4671 is_colstore = do.get("mssql_columnstore")
4672 is_clustered = do.get("mssql_clustered")
4673 if not (is_colstore and is_clustered):
4674 # a clustered columnstore index includes all columns but does
4675 # not want them in the index definition
4676 if row["is_included_column"] and not is_colstore:
4677 # a noncludsted columnstore index reports that includes
4678 # columns but requires that are listed as normal columns
4679 do["mssql_include"].append(row["name"])
4680 else:
4681 index_def["column_names"].append(row["name"])
4683 for tname, idx_dict in indexes_by_table.items():
4684 result[(schema, tname)] = list(idx_dict.values())
4686 for n in names:
4687 key = (schema, name_map(n))
4688 if key not in result:
4689 result[key] = ReflectionDefaults.indexes()
4691 def _fetch_multi_indexes_temp(
4692 self, connection, temp_names, schema, result
4693 ):
4694 """Run the unified indexes queries against tempdb."""
4695 resolved_by_owner, temp_name_map = self._resolve_temp_names(
4696 connection, temp_names
4697 )
4699 for temp_owner, mangled_names in resolved_by_owner.items():
4700 self._fetch_multi_indexes(
4701 connection,
4702 owner=temp_owner,
4703 names=mangled_names,
4704 schema=schema,
4705 name_map=temp_name_map,
4706 result=result,
4707 exec_opts={"schema_translate_map": {"sys": "tempdb.sys"}},
4708 )
4710 @_db_plus_owner_multi
4711 def get_multi_table_comment(
4712 self,
4713 connection,
4714 dbname,
4715 owner,
4716 schema,
4717 filter_names,
4718 scope,
4719 kind,
4720 **kw,
4721 ):
4722 if not self.supports_comments:
4723 raise NotImplementedError(
4724 "Can't get table comments on current SQL Server "
4725 "version in use"
4726 )
4728 regular_names, multi_object_names, temp_names = (
4729 self._partition_filter_names(
4730 connection, owner, filter_names, scope, kind
4731 )
4732 )
4734 result = {}
4736 if multi_object_names:
4737 name_map = self._multi_name_map(regular_names)
4738 self._fetch_multi_table_comment(
4739 connection,
4740 owner=owner,
4741 names=multi_object_names,
4742 schema=schema,
4743 name_map=name_map,
4744 result=result,
4745 exec_opts={"schema_translate_map": {}},
4746 )
4748 if temp_names:
4749 self._fetch_multi_table_comment_temp(
4750 connection,
4751 temp_names=temp_names,
4752 schema=schema,
4753 result=result,
4754 )
4756 return result.items()
4758 @lru_cache()
4759 def _table_comment_select(self):
4760 """Build the Core sys.* select for table comment reflection."""
4761 sys_objects = ischema.sys_objects
4762 sys_schemas = ischema.sys_schemas
4763 extended_properties = ischema.extended_properties
4765 s = (
4766 sql.select(
4767 sys_objects.c.name.label("table_name"),
4768 extended_properties.c.value.label("comment"),
4769 )
4770 .select_from(sys_objects)
4771 .join(
4772 sys_schemas,
4773 onclause=sys_objects.c.schema_id == sys_schemas.c.schema_id,
4774 )
4775 .outerjoin(
4776 extended_properties,
4777 onclause=sql.and_(
4778 extended_properties.c["class"] == 1,
4779 extended_properties.c.name == "MS_Description",
4780 extended_properties.c.major_id == sys_objects.c.object_id,
4781 extended_properties.c.minor_id == 0,
4782 ),
4783 )
4784 .where(
4785 sys_objects.c.type.in_(("U", "V")),
4786 ischema.sys_schemas.c.name == sql.bindparam("owner"),
4787 ischema.sys_objects.c.name.in_(
4788 sql.bindparam("filter_names", expanding=True)
4789 ),
4790 )
4791 )
4792 return s
4794 def _fetch_multi_table_comment(
4795 self, connection, owner, names, schema, name_map, result, exec_opts
4796 ):
4797 """Execute the unified table comment query for one catalog pass.
4799 Core sys.* select so ``schema_translate_map`` actually rewrites
4800 the catalog reference for the tempdb pass.
4801 """
4802 q = self._table_comment_select()
4804 rp = connection.execute(
4805 q,
4806 {"owner": owner, "filter_names": names},
4807 execution_options=exec_opts,
4808 )
4810 for row in rp.mappings():
4811 table_name = name_map(row["table_name"])
4812 comment = row["comment"]
4813 result[(schema, table_name)] = (
4814 {"text": comment} if comment else {"text": None}
4815 )
4817 for n in names:
4818 key = (schema, name_map(n))
4819 if key not in result:
4820 result[key] = ReflectionDefaults.table_comment()
4822 def _fetch_multi_table_comment_temp(
4823 self, connection, temp_names, schema, result
4824 ):
4825 """Run the unified table comment query against tempdb."""
4826 resolved_by_owner, temp_name_map = self._resolve_temp_names(
4827 connection, temp_names
4828 )
4829 for temp_owner, mangled_names in resolved_by_owner.items():
4830 self._fetch_multi_table_comment(
4831 connection,
4832 owner=temp_owner,
4833 names=mangled_names,
4834 schema=schema,
4835 name_map=temp_name_map,
4836 result=result,
4837 exec_opts={"schema_translate_map": {"sys": "tempdb.sys"}},
4838 )
4840 # override mixin methods for which this dialect has no native
4841 # get_multi_* implementation, preventing infinite recursion
4842 # through DefaultDialect._default_multi_reflect
4844 def get_unique_constraints(
4845 self, connection, table_name, schema=None, **kw
4846 ):
4847 raise NotImplementedError()
4849 def get_check_constraints(self, connection, table_name, schema=None, **kw):
4850 raise NotImplementedError()
4852 def get_table_options(self, connection, table_name, schema=None, **kw):
4853 raise NotImplementedError()