1# connectors/mxodbc.py
2# Copyright (C) 2005-2021 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
8"""
9Provide a SQLALchemy connector for the eGenix mxODBC commercial
10Python adapter for ODBC. This is not a free product, but eGenix
11provides SQLAlchemy with a license for use in continuous integration
12testing.
13
14This has been tested for use with mxODBC 3.1.2 on SQL Server 2005
15and 2008, using the SQL Server Native driver. However, it is
16possible for this to be used on other database platforms.
17
18For more info on mxODBC, see https://www.egenix.com/
19
20.. deprecated:: 1.4 The mxODBC DBAPI is deprecated and will be removed
21   in a future version. Please use one of the supported DBAPIs to
22   connect to mssql.
23
24"""
25
26import re
27import sys
28import warnings
29
30from . import Connector
31from ..util import warn_deprecated
32
33
34class MxODBCConnector(Connector):
35    driver = "mxodbc"
36
37    supports_sane_multi_rowcount = False
38    supports_unicode_statements = True
39    supports_unicode_binds = True
40
41    supports_native_decimal = True
42
43    @classmethod
44    def dbapi(cls):
45        # this classmethod will normally be replaced by an instance
46        # attribute of the same name, so this is normally only called once.
47        cls._load_mx_exceptions()
48        platform = sys.platform
49        if platform == "win32":
50            from mx.ODBC import Windows as Module
51        # this can be the string "linux2", and possibly others
52        elif "linux" in platform:
53            from mx.ODBC import unixODBC as Module
54        elif platform == "darwin":
55            from mx.ODBC import iODBC as Module
56        else:
57            raise ImportError("Unrecognized platform for mxODBC import")
58
59        warn_deprecated(
60            "The mxODBC DBAPI is deprecated and will be removed"
61            "in a future version. Please use one of the supported DBAPIs to"
62            "connect to mssql.",
63            version="1.4",
64        )
65        return Module
66
67    @classmethod
68    def _load_mx_exceptions(cls):
69        """Import mxODBC exception classes into the module namespace,
70        as if they had been imported normally. This is done here
71        to avoid requiring all SQLAlchemy users to install mxODBC.
72        """
73        global InterfaceError, ProgrammingError
74        from mx.ODBC import InterfaceError
75        from mx.ODBC import ProgrammingError
76
77    def on_connect(self):
78        def connect(conn):
79            conn.stringformat = self.dbapi.MIXED_STRINGFORMAT
80            conn.datetimeformat = self.dbapi.PYDATETIME_DATETIMEFORMAT
81            conn.decimalformat = self.dbapi.DECIMAL_DECIMALFORMAT
82            conn.errorhandler = self._error_handler()
83
84        return connect
85
86    def _error_handler(self):
87        """Return a handler that adjusts mxODBC's raised Warnings to
88        emit Python standard warnings.
89        """
90        from mx.ODBC.Error import Warning as MxOdbcWarning
91
92        def error_handler(connection, cursor, errorclass, errorvalue):
93            if issubclass(errorclass, MxOdbcWarning):
94                errorclass.__bases__ = (Warning,)
95                warnings.warn(
96                    message=str(errorvalue), category=errorclass, stacklevel=2
97                )
98            else:
99                raise errorclass(errorvalue)
100
101        return error_handler
102
103    def create_connect_args(self, url):
104        r"""Return a tuple of \*args, \**kwargs for creating a connection.
105
106        The mxODBC 3.x connection constructor looks like this:
107
108            connect(dsn, user='', password='',
109                    clear_auto_commit=1, errorhandler=None)
110
111        This method translates the values in the provided URI
112        into args and kwargs needed to instantiate an mxODBC Connection.
113
114        The arg 'errorhandler' is not used by SQLAlchemy and will
115        not be populated.
116
117        """
118        opts = url.translate_connect_args(username="user")
119        opts.update(url.query)
120        args = opts.pop("host")
121        opts.pop("port", None)
122        opts.pop("database", None)
123        return (args,), opts
124
125    def is_disconnect(self, e, connection, cursor):
126        # TODO: eGenix recommends checking connection.closed here
127        # Does that detect dropped connections ?
128        if isinstance(e, self.dbapi.ProgrammingError):
129            return "connection already closed" in str(e)
130        elif isinstance(e, self.dbapi.Error):
131            return "[08S01]" in str(e)
132        else:
133            return False
134
135    def _get_server_version_info(self, connection):
136        # eGenix suggests using conn.dbms_version instead
137        # of what we're doing here
138        dbapi_con = connection.connection
139        version = []
140        r = re.compile(r"[.\-]")
141        # 18 == pyodbc.SQL_DBMS_VER
142        for n in r.split(dbapi_con.getinfo(18)[1]):
143            try:
144                version.append(int(n))
145            except ValueError:
146                version.append(n)
147        return tuple(version)
148
149    def _get_direct(self, context):
150        if context:
151            native_odbc_execute = context.execution_options.get(
152                "native_odbc_execute", "auto"
153            )
154            # default to direct=True in all cases, is more generally
155            # compatible especially with SQL Server
156            return False if native_odbc_execute is True else True
157        else:
158            return True
159
160    def do_executemany(self, cursor, statement, parameters, context=None):
161        cursor.executemany(
162            statement, parameters, direct=self._get_direct(context)
163        )
164
165    def do_execute(self, cursor, statement, parameters, context=None):
166        cursor.execute(statement, parameters, direct=self._get_direct(context))
167