Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/sqlalchemy/connectors/mxodbc.py : 29%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# connectors/mxodbc.py
2# Copyright (C) 2005-2020 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: http://www.opensource.org/licenses/mit-license.php
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.
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.
18For more info on mxODBC, see http://www.egenix.com/
20"""
22import re
23import sys
24import warnings
26from . import Connector
29class MxODBCConnector(Connector):
30 driver = "mxodbc"
32 supports_sane_multi_rowcount = False
33 supports_unicode_statements = True
34 supports_unicode_binds = True
36 supports_native_decimal = True
38 @classmethod
39 def dbapi(cls):
40 # this classmethod will normally be replaced by an instance
41 # attribute of the same name, so this is normally only called once.
42 cls._load_mx_exceptions()
43 platform = sys.platform
44 if platform == "win32":
45 from mx.ODBC import Windows as Module
46 # this can be the string "linux2", and possibly others
47 elif "linux" in platform:
48 from mx.ODBC import unixODBC as Module
49 elif platform == "darwin":
50 from mx.ODBC import iODBC as Module
51 else:
52 raise ImportError("Unrecognized platform for mxODBC import")
53 return Module
55 @classmethod
56 def _load_mx_exceptions(cls):
57 """ Import mxODBC exception classes into the module namespace,
58 as if they had been imported normally. This is done here
59 to avoid requiring all SQLAlchemy users to install mxODBC.
60 """
61 global InterfaceError, ProgrammingError
62 from mx.ODBC import InterfaceError
63 from mx.ODBC import ProgrammingError
65 def on_connect(self):
66 def connect(conn):
67 conn.stringformat = self.dbapi.MIXED_STRINGFORMAT
68 conn.datetimeformat = self.dbapi.PYDATETIME_DATETIMEFORMAT
69 conn.decimalformat = self.dbapi.DECIMAL_DECIMALFORMAT
70 conn.errorhandler = self._error_handler()
72 return connect
74 def _error_handler(self):
75 """ Return a handler that adjusts mxODBC's raised Warnings to
76 emit Python standard warnings.
77 """
78 from mx.ODBC.Error import Warning as MxOdbcWarning
80 def error_handler(connection, cursor, errorclass, errorvalue):
81 if issubclass(errorclass, MxOdbcWarning):
82 errorclass.__bases__ = (Warning,)
83 warnings.warn(
84 message=str(errorvalue), category=errorclass, stacklevel=2
85 )
86 else:
87 raise errorclass(errorvalue)
89 return error_handler
91 def create_connect_args(self, url):
92 r"""Return a tuple of \*args, \**kwargs for creating a connection.
94 The mxODBC 3.x connection constructor looks like this:
96 connect(dsn, user='', password='',
97 clear_auto_commit=1, errorhandler=None)
99 This method translates the values in the provided uri
100 into args and kwargs needed to instantiate an mxODBC Connection.
102 The arg 'errorhandler' is not used by SQLAlchemy and will
103 not be populated.
105 """
106 opts = url.translate_connect_args(username="user")
107 opts.update(url.query)
108 args = opts.pop("host")
109 opts.pop("port", None)
110 opts.pop("database", None)
111 return (args,), opts
113 def is_disconnect(self, e, connection, cursor):
114 # TODO: eGenix recommends checking connection.closed here
115 # Does that detect dropped connections ?
116 if isinstance(e, self.dbapi.ProgrammingError):
117 return "connection already closed" in str(e)
118 elif isinstance(e, self.dbapi.Error):
119 return "[08S01]" in str(e)
120 else:
121 return False
123 def _get_server_version_info(self, connection):
124 # eGenix suggests using conn.dbms_version instead
125 # of what we're doing here
126 dbapi_con = connection.connection
127 version = []
128 r = re.compile(r"[.\-]")
129 # 18 == pyodbc.SQL_DBMS_VER
130 for n in r.split(dbapi_con.getinfo(18)[1]):
131 try:
132 version.append(int(n))
133 except ValueError:
134 version.append(n)
135 return tuple(version)
137 def _get_direct(self, context):
138 if context:
139 native_odbc_execute = context.execution_options.get(
140 "native_odbc_execute", "auto"
141 )
142 # default to direct=True in all cases, is more generally
143 # compatible especially with SQL Server
144 return False if native_odbc_execute is True else True
145 else:
146 return True
148 def do_executemany(self, cursor, statement, parameters, context=None):
149 cursor.executemany(
150 statement, parameters, direct=self._get_direct(context)
151 )
153 def do_execute(self, cursor, statement, parameters, context=None):
154 cursor.execute(statement, parameters, direct=self._get_direct(context))