codekingpro/portable-devtools

1# dialects/mssql/pymssql.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
8
9
10"""
11.. dialect:: mssql+pymssql
12    :name: pymssql
13    :dbapi: pymssql
14    :connectstring: mssql+pymssql://<username>:<password>@<freetds_name>/?charset=utf8
15
16pymssql is a Python module that provides a Python DBAPI interface around
17`FreeTDS <https://www.freetds.org/>`_.
18
19.. versionchanged:: 2.0.5
20
21    pymssql was restored to SQLAlchemy's continuous integration testing
22
23
24"""  # noqa
25import re
26
27from .base import MSDialect
28from .base import MSIdentifierPreparer
29from ... import types as sqltypes
30from ... import util
31from ...engine import processors
32
33
34class _MSNumeric_pymssql(sqltypes.Numeric):
35    def result_processor(self, dialect, type_):
36        if not self.asdecimal:
37            return processors.to_float
38        else:
39            return sqltypes.Numeric.result_processor(self, dialect, type_)
40
41
42class MSIdentifierPreparer_pymssql(MSIdentifierPreparer):
43    def __init__(self, dialect):
44        super().__init__(dialect)
45        # pymssql has the very unusual behavior that it uses pyformat
46        # yet does not require that percent signs be doubled
47        self._double_percents = False
48
49
50class MSDialect_pymssql(MSDialect):
51    supports_statement_cache = True
52    supports_native_decimal = True
53    supports_native_uuid = True
54    driver = "pymssql"
55
56    preparer = MSIdentifierPreparer_pymssql
57
58    colspecs = util.update_copy(
59        MSDialect.colspecs,
60        {sqltypes.Numeric: _MSNumeric_pymssql, sqltypes.Float: sqltypes.Float},
61    )
62
63    @classmethod
64    def import_dbapi(cls):
65        module = __import__("pymssql")
66        # pymmsql < 2.1.1 doesn't have a Binary method.  we use string
67        client_ver = tuple(int(x) for x in module.__version__.split("."))
68        if client_ver < (2, 1, 1):
69            # TODO: monkeypatching here is less than ideal
70            module.Binary = lambda x: x if hasattr(x, "decode") else str(x)
71
72        if client_ver < (1,):
73            util.warn(
74                "The pymssql dialect expects at least "
75                "the 1.0 series of the pymssql DBAPI."
76            )
77        return module
78
79    def _get_server_version_info(self, connection):
80        vers = connection.exec_driver_sql("select @@version").scalar()
81        m = re.match(r"Microsoft .*? - (\d+)\.(\d+)\.(\d+)\.(\d+)", vers)
82        if m:
83            return tuple(int(x) for x in m.group(1, 2, 3, 4))
84        else:
85            return None
86
87    def create_connect_args(self, url):
88        opts = url.translate_connect_args(username="user")
89        opts.update(url.query)
90        port = opts.pop("port", None)
91        if port and "host" in opts:
92            opts["host"] = "%s:%s" % (opts["host"], port)
93        return ([], opts)
94
95    def is_disconnect(self, e, connection, cursor):
96        for msg in (
97            "Adaptive Server connection timed out",
98            "Net-Lib error during Connection reset by peer",
99            "message 20003",  # connection timeout
100            "Error 10054",
101            "Not connected to any MS SQL server",
102            "Connection is closed",
103            "message 20006",  # Write to the server failed
104            "message 20017",  # Unexpected EOF from the server
105            "message 20047",  # DBPROCESS is dead or not enabled
106            "The server failed to resume the transaction",
107        ):
108            if msg in str(e):
109                return True
110        else:
111            return False
112
113    def get_isolation_level_values(self, dbapi_connection):
114        return super().get_isolation_level_values(dbapi_connection) + [
115            "AUTOCOMMIT"
116        ]
117
118    def set_isolation_level(self, dbapi_connection, level):
119        if level == "AUTOCOMMIT":
120            dbapi_connection.autocommit(True)
121        else:
122            dbapi_connection.autocommit(False)
123            super().set_isolation_level(dbapi_connection, level)
124
125
126dialect = MSDialect_pymssql
127