:ticket:`11250`
+.. _change_12869:
+
+Support for mssql-python driver
+--------------------------------
+
+SQLAlchemy 2.1 adds support for the ``mssql-python`` driver, Microsoft's
+official Python driver for SQL Server. This driver represents a modern
+alternative to the widely-used ``pyodbc`` driver, with native support for
+several SQL Server-specific features.
+
+The ``mssql-python`` driver can be used by specifying ``mssql+mssqlpython``
+in the connection URL::
+
+ from sqlalchemy import create_engine
+
+ # Basic connection
+ engine = create_engine("mssql+mssqlpython://user:password@hostname/database")
+
+ # With Windows Authentication
+ engine = create_engine(
+ "mssql+mssqlpython://hostname/database?authentication=ActiveDirectoryIntegrated"
+ )
+
+The ``mssql-python`` driver is available from PyPI:
+
+.. sourcecode:: text
+
+ pip install mssql-python
+
+.. seealso::
+
+ :ref:`mssql_python` - Documentation for the mssql-python dialect
+
+:ticket:`12869`
+
Oracle Database
===============
--- /dev/null
+.. change::
+ :tags: feature, mssql
+ :tickets: 12869
+
+ Added support for the ``mssql-python`` driver, Microsoft's official Python
+ driver for SQL Server.
+
+ .. seealso::
+
+ :ref:`change_12869`
+
------
.. automodule:: sqlalchemy.dialects.mssql.pyodbc
+.. _mssql_python:
+
+mssql-python
+------------
+.. automodule:: sqlalchemy.dialects.mssql.mssqlpython
+
pymssql
-------
.. automodule:: sqlalchemy.dialects.mssql.pymssql
from . import aioodbc # noqa
from . import base # noqa
+from . import mssqlpython # noqa
from . import pymssql # noqa
from . import pyodbc # noqa
from .base import BIGINT
from __future__ import annotations
+from .base import MSExecutionContext
from .pyodbc import MSDialect_pyodbc
-from .pyodbc import MSExecutionContext_pyodbc
from ...connectors.aioodbc import aiodbcConnector
-class MSExecutionContext_aioodbc(MSExecutionContext_pyodbc):
+class MSExecutionContext_aioodbc(MSExecutionContext):
def create_server_side_cursor(self):
return self._dbapi_connection.cursor(server_side=True)
class MSExecutionContext(default.DefaultExecutionContext):
+ _embedded_scope_identity = False
_enable_identity_insert = False
_select_lastrowid = False
_lastrowid = None
self,
)
+ # don't embed the scope_identity select into an
+ # "INSERT .. DEFAULT VALUES"
+ if (
+ self._select_lastrowid
+ and self.dialect.scope_identity_must_be_embedded
+ and self.dialect.use_scope_identity
+ and len(self.parameters[0])
+ ):
+ self._embedded_scope_identity = True
+
+ self.statement += "; select scope_identity()"
+
def post_exec(self):
- """Disable IDENTITY_INSERT if enabled."""
conn = self.root_connection
if self.isinsert or self.isupdate or self.isdelete:
self._rowcount = self.cursor.rowcount
- if self._select_lastrowid:
+ # handle INSERT with embedded SELECT SCOPE_IDENTITY() call
+ if self._embedded_scope_identity:
+ # Fetch the last inserted id from the manipulated statement
+ # We may have to skip over a number of result sets with
+ # no data (due to triggers, etc.) so run up to three times
+
+ row = None
+ for _ in range(3):
+ if self.cursor.description:
+ rows = self.cursor.fetchall()
+ if rows:
+ row = rows[0]
+ break
+ else:
+ self.cursor.nextset()
+
+ self._lastrowid = int(row[0]) if row else None
+
+ self.cursor_fetch_strategy = _cursor._NO_CURSOR_DML
+
+ elif self._select_lastrowid:
if self.dialect.use_scope_identity:
conn._cursor_execute(
self.cursor,
)
if self._enable_identity_insert:
+ # Disable IDENTITY_INSERT if enabled.
if TYPE_CHECKING:
assert is_sql_compiler(self.compiled)
assert isinstance(self.compiled.compile_state, DMLState)
supports_default_values = True
supports_empty_insert = False
favor_returning_over_lastrowid = True
+ scope_identity_must_be_embedded = False
returns_native_bytes = True
'attempting to query the "{}" view.'.format(err, view_name)
) from err
else:
+
row = cursor.fetchone()
return row[0].upper()
finally:
--- /dev/null
+# dialects/mssql/mssqlpython.py
+# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: https://www.opensource.org/licenses/mit-license.php
+# mypy: ignore-errors
+
+"""
+.. dialect:: mssql+mssqlpython
+ :name: mssqlpython
+ :dbapi: mssql-python
+ :connectstring: mssql+mssqlpython://<username>:<password>@<host>:<port>/<dbname>
+ :url: https://github.com/microsoft/mssql-python
+
+mssql-python is a driver for Microsoft SQL Server produced by Microsoft.
+
+.. versionadded:: 2.1.0b2
+
+
+The driver is generally similar to pyodbc in most aspects as it is based
+on the same ODBC framework.
+
+Connection Strings
+------------------
+
+Examples of connecting with the mssql-python driver::
+
+ from sqlalchemy import create_engine
+
+ # Basic connection
+ engine = create_engine(
+ "mssql+mssqlpython://user:password@hostname/database"
+ )
+
+ # With Windows Authentication
+ engine = create_engine(
+ "mssql+mssqlpython://hostname/database?authentication=ActiveDirectoryIntegrated"
+ )
+
+""" # noqa
+
+from __future__ import annotations
+
+import re
+from typing import Any
+from typing import Dict
+from typing import List
+from typing import Optional
+from typing import TYPE_CHECKING
+from typing import Union
+
+from .base import MSDialect
+from .pyodbc import _ms_numeric_pyodbc
+from ... import util
+from ...sql import sqltypes
+
+if TYPE_CHECKING:
+ from ... import pool
+ from ...engine import interfaces
+ from ...engine.interfaces import ConnectArgsType
+ from ...engine.interfaces import DBAPIModule
+ from ...engine.interfaces import IsolationLevel
+ from ...engine.interfaces import URL
+
+
+class _MSNumeric_mssqlpython(_ms_numeric_pyodbc, sqltypes.Numeric):
+ pass
+
+
+class _MSFloat_mssqlpython(_ms_numeric_pyodbc, sqltypes.Float):
+ pass
+
+
+class MSDialect_mssqlpython(MSDialect):
+ driver = "mssqlpython"
+ supports_statement_cache = True
+
+ supports_sane_rowcount_returning = True
+ supports_sane_multi_rowcount = True
+ supports_native_uuid = True
+ scope_identity_must_be_embedded = True
+
+ supports_native_decimal = True
+
+ # used by pyodbc _ms_numeric_pyodbc class
+ _need_decimal_fix = True
+
+ colspecs = util.update_copy(
+ MSDialect.colspecs,
+ {
+ sqltypes.Numeric: _MSNumeric_mssqlpython,
+ sqltypes.Float: _MSFloat_mssqlpython,
+ },
+ )
+
+ def __init__(self, enable_pooling=False, **kw):
+ super().__init__(**kw)
+ if not enable_pooling and self.dbapi is not None:
+ self.loaded_dbapi.pooling(enabled=False)
+
+ @classmethod
+ def import_dbapi(cls) -> DBAPIModule:
+ return __import__("mssql_python")
+
+ def create_connect_args(self, url: URL) -> ConnectArgsType:
+ opts = url.translate_connect_args(username="user")
+ opts.update(url.query)
+
+ keys = opts
+
+ query = url.query
+
+ connect_args: Dict[str, Any] = {}
+ connectors: List[str]
+
+ def check_quote(token: str) -> str:
+ if ";" in str(token) or str(token).startswith("{"):
+ token = "{%s}" % token.replace("}", "}}")
+ return token
+
+ keys = {k: check_quote(v) for k, v in keys.items()}
+
+ port = ""
+ if "port" in keys and "port" not in query:
+ port = ",%d" % int(keys.pop("port"))
+
+ connectors = []
+
+ connectors.extend(
+ [
+ "Server=%s%s" % (keys.pop("host", ""), port),
+ "Database=%s" % keys.pop("database", ""),
+ ]
+ )
+
+ user = keys.pop("user", None)
+ if user:
+ connectors.append("UID=%s" % user)
+ pwd = keys.pop("password", "")
+ if pwd:
+ connectors.append("PWD=%s" % pwd)
+ else:
+ authentication = keys.pop("authentication", None)
+ if authentication:
+ connectors.append("Authentication=%s" % authentication)
+
+ connectors.extend(["%s=%s" % (k, v) for k, v in keys.items()])
+
+ return ((";".join(connectors),), connect_args)
+
+ def is_disconnect(
+ self,
+ e: Exception,
+ connection: Optional[
+ Union[pool.PoolProxiedConnection, interfaces.DBAPIConnection]
+ ],
+ cursor: Optional[interfaces.DBAPICursor],
+ ) -> bool:
+ if isinstance(e, self.loaded_dbapi.ProgrammingError):
+ return (
+ "The cursor's connection has been closed." in str(e)
+ or "Attempt to use a closed connection." in str(e)
+ or "Driver Error: Operation cannot be performed" in str(e)
+ )
+ elif isinstance(e, self.loaded_dbapi.InterfaceError):
+ return bool(re.search(r"Cannot .* on closed connection", str(e)))
+ else:
+ return False
+
+ def _dbapi_version(self) -> interfaces.VersionInfoType:
+ if not self.dbapi:
+ return ()
+ return self._parse_dbapi_version(self.dbapi.version)
+
+ def _parse_dbapi_version(self, vers: str) -> interfaces.VersionInfoType:
+ m = re.match(r"(?:py.*-)?([\d\.]+)(?:-(\w+))?", vers)
+ if not m:
+ return ()
+ vers_tuple: interfaces.VersionInfoType = tuple(
+ [int(x) for x in m.group(1).split(".")]
+ )
+ if m.group(2):
+ vers_tuple += (m.group(2),)
+ return vers_tuple
+
+ def _get_server_version_info(self, connection):
+ vers = connection.exec_driver_sql("select @@version").scalar()
+ m = re.match(r"Microsoft .*? - (\d+)\.(\d+)\.(\d+)\.(\d+)", vers)
+ if m:
+ return tuple(int(x) for x in m.group(1, 2, 3, 4))
+ else:
+ return None
+
+ def get_isolation_level_values(
+ self, dbapi_connection: interfaces.DBAPIConnection
+ ) -> List[IsolationLevel]:
+ return [
+ *super().get_isolation_level_values(dbapi_connection),
+ "AUTOCOMMIT",
+ ]
+
+ def set_isolation_level(
+ self,
+ dbapi_connection: interfaces.DBAPIConnection,
+ level: IsolationLevel,
+ ) -> None:
+ if level == "AUTOCOMMIT":
+ dbapi_connection.autocommit = True
+ else:
+ dbapi_connection.autocommit = False
+ super().set_isolation_level(dbapi_connection, level)
+
+ def detect_autocommit_setting(
+ self, dbapi_conn: interfaces.DBAPIConnection
+ ) -> bool:
+ return bool(dbapi_conn.autocommit)
+
+
+dialect = MSDialect_mssqlpython
from ...schema import MetaData
from ...schema import Table
from ...testing.provision import create_db
+from ...testing.provision import dbapi_error
from ...testing.provision import drop_all_schema_objects_pre_tables
from ...testing.provision import drop_db
from ...testing.provision import generate_driver_url
new_url = url.set(drivername="%s+%s" % (backend, driver))
- if driver not in ("pyodbc", "aioodbc"):
+ if driver == "pymssql" and url.get_driver_name() != "pymssql":
new_url = new_url.set(query="")
+ elif driver == "mssqlpython" and url.get_driver_name() != "mssqlpython":
+ new_url = new_url.set(query={"Encrypt": "No"})
if driver == "aioodbc":
new_url = new_url.update_query_dict({"MARS_Connection": "Yes"})
if sequence.start is None:
sequence.start = 1
return sequence
+
+
+@dbapi_error.for_db("mssql")
+def dbapi_error(cfg, cls, message):
+ if cfg.db.driver == "mssqlpython":
+ return cls(message, "placeholder for mssqlpython")
+ else:
+ return cls(message)
from .base import BINARY
from .base import DATETIMEOFFSET
from .base import MSDialect
-from .base import MSExecutionContext
from .base import VARBINARY
from .json import JSON as _MSJson
from .json import JSONIndexType as _MSJsonIndexType
from ... import types as sqltypes
from ... import util
from ...connectors.pyodbc import PyODBCConnector
-from ...engine import cursor as _cursor
class _ms_numeric_pyodbc:
return dbapi.SQL_WVARCHAR
-class MSExecutionContext_pyodbc(MSExecutionContext):
- _embedded_scope_identity = False
-
- def pre_exec(self):
- """where appropriate, issue "select scope_identity()" in the same
- statement.
-
- Background on why "scope_identity()" is preferable to "@@identity":
- https://msdn.microsoft.com/en-us/library/ms190315.aspx
-
- Background on why we attempt to embed "scope_identity()" into the same
- statement as the INSERT:
- https://code.google.com/p/pyodbc/wiki/FAQs#How_do_I_retrieve_autogenerated/identity_values?
-
- """
-
- super().pre_exec()
-
- # don't embed the scope_identity select into an
- # "INSERT .. DEFAULT VALUES"
- if (
- self._select_lastrowid
- and self.dialect.use_scope_identity
- and len(self.parameters[0])
- ):
- self._embedded_scope_identity = True
-
- self.statement += "; select scope_identity()"
-
- def post_exec(self):
- if self._embedded_scope_identity:
- # Fetch the last inserted id from the manipulated statement
- # We may have to skip over a number of result sets with
- # no data (due to triggers, etc.)
- while True:
- try:
- # fetchall() ensures the cursor is consumed
- # without closing it (FreeTDS particularly)
- rows = self.cursor.fetchall()
- except self.dialect.dbapi.Error:
- # no way around this - nextset() consumes the previous set
- # so we need to just keep flipping
- self.cursor.nextset()
- else:
- if not rows:
- # async adapter drivers just return None here
- self.cursor.nextset()
- continue
- row = rows[0]
- break
-
- self._lastrowid = int(row[0])
-
- self.cursor_fetch_strategy = _cursor._NO_CURSOR_DML
- else:
- super().post_exec()
-
-
class MSDialect_pyodbc(PyODBCConnector, MSDialect):
supports_statement_cache = True
+ scope_identity_must_be_embedded = True
+
# note this parameter is no longer used by the ORM or default dialect
# see #9414
supports_sane_rowcount_returning = False
- execution_ctx_cls = MSExecutionContext_pyodbc
-
colspecs = util.update_copy(
MSDialect.colspecs,
{
sub_params,
context,
)
-
except BaseException as e:
self._handle_dbapi_exception(
e,
Dict[Tuple[Any, ...], Any],
Dict[Any, Any],
]
+
if composite_sentinel:
rows_by_sentinel = {
tuple(
connection.execute(table.delete())
else:
connection.execute(table.delete())
+
+
+@register.init
+def dbapi_error(cfg, cls, message):
+ """create a DBAPI error
+
+ :param cls: the DBAPI class, like ``dialect.dbapi.OperationalError``
+ :param message: message for the error
+
+ """
+ return cls(message)
"nogreenlet": {"cx_oracle", "oracledb"},
"greenlet": {"oracledb_async"},
},
- "mssql": {"nogreenlet": {"pyodbc", "pymssql"}, "greenlet": {"aioodbc"}},
+ "mssql": {
+ "nogreenlet": {"pyodbc", "pymssql", "mssqlpython"},
+ "greenlet": {"aioodbc"},
+ },
}
mssql = ["pyodbc"]
mssql-pymssql = ["pymssql"]
mssql-pyodbc = ["pyodbc"]
+mssql-python = ["mssql-python>=0.13.1"]
mysql = ["mysqlclient>=1.4.0"]
mysql-connector = ["mysql-connector-python"]
mariadb-connector = ["mariadb>=1.0.1,!=1.1.2,!=1.1.5,!=1.1.10"]
tests-mssql = [
"sqlalchemy[mssql]",
"sqlalchemy[mssql-pymssql]",
+ "sqlalchemy[mssql-python]",
]
tests-mssql-asyncio = [
mssql = mssql+pyodbc://scott:tiger^5HHH@mssql2022:1433/test?driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes&Encrypt=Optional
mssql_async = mssql+aioodbc://scott:tiger^5HHH@mssql2022:1433/test?driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes&Encrypt=Optional
pymssql = mssql+pymssql://scott:tiger^5HHH@mssql2022:1433/test
+mssql_python = mssql+mssqlpython://scott:tiger^5HHH@mssql2022:1433/test?Encrypt=No
docker_mssql = mssql+pyodbc://scott:tiger^5HHH@127.0.0.1:1433/test?driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes&Encrypt=Optional
oracle = oracle+oracledb://scott:tiger@oracle23c/freepdb1
cxoracle = oracle+cx_oracle://scott:tiger@oracle23c/freepdb1
older SCOPE_IDENTITY() call, which still works for this scenario.
To enable the workaround, the Table must be instantiated
with the init parameter 'implicit_returning = False'.
+
+ Note: this **wont work** with SELECT @@identity. it has to be
+ SCOPE_IDENTITY(). for pyodbc, this means the function has to be
+ embedded on the same line as the INSERT.
+
"""
# TODO: this same test needs to be tried in a multithreaded context
),
)
- @testing.only_on("mssql+pyodbc")
+ @testing.only_on(["mssql+pyodbc", "mssql+mssqlpython"])
@testing.provide_metadata
def test_embedded_scope_identity(self):
engine = engines.testing_engine(options={"use_scope_identity": True})
with engine.begin() as conn:
conn.execute(t1.insert(), {"data": "somedata"})
- # pyodbc-specific system
+ if testing.against("mssql+pyodbc"):
+ paramstyle = "?"
+ params = ("somedata",)
+ elif testing.against("mssql+mssqlpython"):
+ paramstyle = "%(data)s"
+ params = {"data": "somedata"}
+ else:
+ assert False
+
asserter.assert_(
CursorSQL(
- "INSERT INTO t1 (data) VALUES (?); select scope_identity()",
- ("somedata",),
+ f"INSERT INTO t1 (data) VALUES ({paramstyle}); "
+ "select scope_identity()",
+ params,
consume_statement=False,
)
)
from sqlalchemy import types
from sqlalchemy import Unicode
from sqlalchemy import UnicodeText
+from sqlalchemy import VARCHAR
from sqlalchemy.dialects.mssql import base as mssql
from sqlalchemy.dialects.mssql import NTEXT
from sqlalchemy.dialects.mssql import ROWVERSION
),
None,
True,
+ testing.fails_if(
+ "+mssqlpython",
+ "mssql-python does not seem to report this as an error",
+ ),
),
(
"dto_param_datetime_naive",
lambda: datetime.datetime(2007, 10, 30, 11, 2, 32, 123456),
0,
False,
+ testing.fails_if("+pymssql", "seems to fail on pymssql"),
),
(
"dto_param_string_one",
0,
False,
),
- ("dto_param_string_invalid", lambda: "this is not a date", 0, True),
+ (
+ "dto_param_string_invalid",
+ lambda: "this is not a date",
+ 0,
+ True,
+ ),
id_="iaaa",
argnames="dto_param_value, expected_offset_hours, should_fail",
)
- @testing.skip_if("+pymssql", "offsets dont seem to work")
def test_datetime_offset(
self,
datetimeoffset_fixture,
class StringRoundTripTest(fixtures.TestBase):
- """tests for #8661
-
-
- at the moment most of these are using the default setinputsizes enabled
- behavior, with the exception of the plain executemany() calls for inserts.
-
-
- """
+ """tests for #8661 as well as other string issues"""
__only_on__ = "mssql"
__backend__ = True
)
eq_(result.all(), ["some other data %d" % i for i in range(3)])
+ @testing.combinations(
+ (3,),
+ (9,),
+ (10,),
+ (15,),
+ argnames="size",
+ )
+ @testing.variation("fetchtype", ["all", "one"])
+ def test_cp1252_varchar(
+ self, size, fetchtype: testing.Variation, metadata, connection
+ ):
+ """test for mssql-python's issue with small varchar + cp1252 strings"""
+
+ t = Table("t", metadata, Column("data", VARCHAR(size)))
+
+ t.create(connection)
+
+ data = "réveillé réveillé"[0:size]
+ connection.execute(t.insert(), {"data": data})
+
+ if fetchtype.all:
+ result = connection.execute(select(t.c.data))
+ rows = result.all()
+ received = rows[0][0]
+ elif fetchtype.one:
+ received = connection.scalar(select(t.c.data))
+ else:
+ fetchtype.fail()
+ eq_(received, data)
+
class UniqueIdentifierTest(test_types.UuidTest):
__only_on__ = "mssql"
from sqlalchemy.testing import is_not
from sqlalchemy.testing import is_true
from sqlalchemy.testing import ne_
+from sqlalchemy.testing import provision
from sqlalchemy.testing.assertions import expect_deprecated
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.provision import normalize_sequence
"+mysqlconnector",
"Exception doesn't come back exactly the same from pickle",
)
+ @testing.fails_on(
+ "+mssqlpython",
+ "Exception pickling not working due to additional parameter",
+ )
@testing.fails_on(
"oracle+cx_oracle",
"cx_oracle exception seems to be having some issue with pickling",
conn = engine.connect()
def boom(connection):
- raise engine.dialect.dbapi.OperationalError("rollback failed")
+ raise provision.dbapi_error(
+ config,
+ engine.dialect.dbapi.OperationalError,
+ "rollback failed",
+ )
with patch.object(conn.dialect, "do_rollback", boom):
assert_raises_message(
conn = engine.connect()
def boom(connection):
- raise engine.dialect.dbapi.OperationalError("rollback failed")
+ raise provision.dbapi_error(
+ config,
+ engine.dialect.dbapi.OperationalError,
+ "rollback failed",
+ )
@event.listens_for(conn, "begin")
def _do_begin(conn):
with patch.object(
conn.dialect,
"get_isolation_level",
- Mock(side_effect=ProgrammingError("random error")),
+ Mock(
+ side_effect=provision.dbapi_error(
+ config, ProgrammingError, "random error"
+ )
+ ),
):
assert_raises(MySpecialException, conn.get_isolation_level)
dialect = testing.db.dialect
dbapi = dialect.dbapi
assert not dialect.is_disconnect(
- dbapi.OperationalError("test"), None, None
+ provision.dbapi_error(config, dbapi.OperationalError, "test"),
+ None,
+ None,
)
def test_dont_create_transaction_on_initialize(self):
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import assert_raises_message_context_ok
from sqlalchemy.testing import assert_warns_message
+from sqlalchemy.testing import config
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises
from sqlalchemy.testing import is_true
from sqlalchemy.testing import mock
from sqlalchemy.testing import ne_
+from sqlalchemy.testing import provision
from sqlalchemy.testing.engines import DBAPIProxyConnection
from sqlalchemy.testing.engines import DBAPIProxyCursor
from sqlalchemy.testing.engines import testing_engine
class ExplodeCursor(DBAPIProxyCursor):
def execute(self, stmt, parameters=None, **kw):
if fail and next(fail_count) < 1:
- raise DBAPIError("unhandled disconnect situation")
+ raise provision.dbapi_error(
+ config, DBAPIError, "unhandled disconnect situation"
+ )
else:
return super().execute(stmt, parameters=parameters, **kw)
def non_broken_binary(self):
"""target DBAPI must work fully with binary values"""
+ # for pymssql
# see https://github.com/pymssql/pymssql/issues/504
- return skip_if(["mssql+pymssql"])
+ #
+ # for mssqlpython:
+ # Streaming parameters is not yet supported. Parameter size must be
+ # less than 8192 bytes
+ return skip_if(["mssql+pymssql", "mssql+mssqlpython"])
@property
def binary_comparisons(self):
"no FOR UPDATE NOWAIT support",
)
+ @property
+ def unusual_column_name_characters(self):
+ """target database allows column names that have unusual characters
+ in them, such as dots, spaces, slashes, or percent signs.
+
+ The column names are as always in such a case quoted, however the
+ DB still needs to support those characters in the name somehow.
+
+ """
+ return exclusions.skip_if("+mssqlpython", "waiting on GH issue 464")
+
@property
def subqueries(self):
"""Target database must support subqueries."""
"""
- return exclusions.open()
+ return fails_on("+mssqlpython")
@property
def fetch_null_from_numeric(self):
oracle: oracle_oracledb
mssql: mssql
mssql-py{38,39,310,311,312,313}: mssql_pymssql
+ mssql: mssql-python
install_command=
# TODO: I can find no way to get pip / tox / anyone to have this
# for mssql, aioodbc frequently segfaults under free-threaded builds
mssql: MSSQL={env:TOX_MSSQL:--db mssql}
- mssql: EXTRA_MSSQL_DRIVERS={env:EXTRA_MSSQL_DRIVERS:--dbdriver pyodbc --dbdriver aioodbc --dbdriver pymssql}
- mssql-py314: EXTRA_MSSQL_DRIVERS={env:EXTRA_MSSQL_DRIVERS:--dbdriver pyodbc --dbdriver aioodbc}
+ mssql: EXTRA_MSSQL_DRIVERS={env:EXTRA_MSSQL_DRIVERS:--dbdriver pyodbc --dbdriver aioodbc --dbdriver pymssql --dbdriver mssqlpython}
+ mssql-py314: EXTRA_MSSQL_DRIVERS={env:EXTRA_MSSQL_DRIVERS:--dbdriver pyodbc --dbdriver aioodbc --dbdriver mssqlpython}
mssql-{py313t,py314t,nogreenlet: EXTRA_MSSQL_DRIVERS={env:EXTRA_MSSQL_DRIVERS:--dbdriver pyodbc --dbdriver pymssql}
oracle,mssql,sqlite_file: IDENTS=--write-idents db_idents.txt
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
