codekingpro/portable-devtools

1# connectors/pyodbc.py
2# Copyright (C) 2005-2024 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
8from __future__ import annotations
9
10import re
11from types import ModuleType
12import typing
13from typing import Any
14from typing import Dict
15from typing import List
16from typing import Optional
17from typing import Tuple
18from typing import Union
19from urllib.parse import unquote_plus
20
21from . import Connector
22from .. import ExecutionContext
23from .. import pool
24from .. import util
25from ..engine import ConnectArgsType
26from ..engine import Connection
27from ..engine import interfaces
28from ..engine import URL
29from ..sql.type_api import TypeEngine
30
31if typing.TYPE_CHECKING:
32    from ..engine.interfaces import IsolationLevel
33
34
35class PyODBCConnector(Connector):
36    driver = "pyodbc"
37
38    # this is no longer False for pyodbc in general
39    supports_sane_rowcount_returning = True
40    supports_sane_multi_rowcount = False
41
42    supports_native_decimal = True
43    default_paramstyle = "named"
44
45    fast_executemany = False
46
47    # for non-DSN connections, this *may* be used to
48    # hold the desired driver name
49    pyodbc_driver_name: Optional[str] = None
50
51    dbapi: ModuleType
52
53    def __init__(self, use_setinputsizes: bool = False, **kw: Any):
54        super().__init__(**kw)
55        if use_setinputsizes:
56            self.bind_typing = interfaces.BindTyping.SETINPUTSIZES
57
58    @classmethod
59    def import_dbapi(cls) -> ModuleType:
60        return __import__("pyodbc")
61
62    def create_connect_args(self, url: URL) -> ConnectArgsType:
63        opts = url.translate_connect_args(username="user")
64        opts.update(url.query)
65
66        keys = opts
67
68        query = url.query
69
70        connect_args: Dict[str, Any] = {}
71        connectors: List[str]
72
73        for param in ("ansi", "unicode_results", "autocommit"):
74            if param in keys:
75                connect_args[param] = util.asbool(keys.pop(param))
76
77        if "odbc_connect" in keys:
78            connectors = [unquote_plus(keys.pop("odbc_connect"))]
79        else:
80
81            def check_quote(token: str) -> str:
82                if ";" in str(token) or str(token).startswith("{"):
83                    token = "{%s}" % token.replace("}", "}}")
84                return token
85
86            keys = {k: check_quote(v) for k, v in keys.items()}
87
88            dsn_connection = "dsn" in keys or (
89                "host" in keys and "database" not in keys
90            )
91            if dsn_connection:
92                connectors = [
93                    "dsn=%s" % (keys.pop("host", "") or keys.pop("dsn", ""))
94                ]
95            else:
96                port = ""
97                if "port" in keys and "port" not in query:
98                    port = ",%d" % int(keys.pop("port"))
99
100                connectors = []
101                driver = keys.pop("driver", self.pyodbc_driver_name)
102                if driver is None and keys:
103                    # note if keys is empty, this is a totally blank URL
104                    util.warn(
105                        "No driver name specified; "
106                        "this is expected by PyODBC when using "
107                        "DSN-less connections"
108                    )
109                else:
110                    connectors.append("DRIVER={%s}" % driver)
111
112                connectors.extend(
113                    [
114                        "Server=%s%s" % (keys.pop("host", ""), port),
115                        "Database=%s" % keys.pop("database", ""),
116                    ]
117                )
118
119            user = keys.pop("user", None)
120            if user:
121                connectors.append("UID=%s" % user)
122                pwd = keys.pop("password", "")
123                if pwd:
124                    connectors.append("PWD=%s" % pwd)
125            else:
126                authentication = keys.pop("authentication", None)
127                if authentication:
128                    connectors.append("Authentication=%s" % authentication)
129                else:
130                    connectors.append("Trusted_Connection=Yes")
131
132            # if set to 'Yes', the ODBC layer will try to automagically
133            # convert textual data from your database encoding to your
134            # client encoding.  This should obviously be set to 'No' if
135            # you query a cp1253 encoded database from a latin1 client...
136            if "odbc_autotranslate" in keys:
137                connectors.append(
138                    "AutoTranslate=%s" % keys.pop("odbc_autotranslate")
139                )
140
141            connectors.extend(["%s=%s" % (k, v) for k, v in keys.items()])
142
143        return ((";".join(connectors),), connect_args)
144
145    def is_disconnect(
146        self,
147        e: Exception,
148        connection: Optional[
149            Union[pool.PoolProxiedConnection, interfaces.DBAPIConnection]
150        ],
151        cursor: Optional[interfaces.DBAPICursor],
152    ) -> bool:
153        if isinstance(e, self.dbapi.ProgrammingError):
154            return "The cursor's connection has been closed." in str(
155                e
156            ) or "Attempt to use a closed connection." in str(e)
157        else:
158            return False
159
160    def _dbapi_version(self) -> interfaces.VersionInfoType:
161        if not self.dbapi:
162            return ()
163        return self._parse_dbapi_version(self.dbapi.version)
164
165    def _parse_dbapi_version(self, vers: str) -> interfaces.VersionInfoType:
166        m = re.match(r"(?:py.*-)?([\d\.]+)(?:-(\w+))?", vers)
167        if not m:
168            return ()
169        vers_tuple: interfaces.VersionInfoType = tuple(
170            [int(x) for x in m.group(1).split(".")]
171        )
172        if m.group(2):
173            vers_tuple += (m.group(2),)
174        return vers_tuple
175
176    def _get_server_version_info(
177        self, connection: Connection
178    ) -> interfaces.VersionInfoType:
179        # NOTE: this function is not reliable, particularly when
180        # freetds is in use.   Implement database-specific server version
181        # queries.
182        dbapi_con = connection.connection.dbapi_connection
183        version: Tuple[Union[int, str], ...] = ()
184        r = re.compile(r"[.\-]")
185        for n in r.split(dbapi_con.getinfo(self.dbapi.SQL_DBMS_VER)):  # type: ignore[union-attr]  # noqa: E501
186            try:
187                version += (int(n),)
188            except ValueError:
189                pass
190        return tuple(version)
191
192    def do_set_input_sizes(
193        self,
194        cursor: interfaces.DBAPICursor,
195        list_of_tuples: List[Tuple[str, Any, TypeEngine[Any]]],
196        context: ExecutionContext,
197    ) -> None:
198        # the rules for these types seems a little strange, as you can pass
199        # non-tuples as well as tuples, however it seems to assume "0"
200        # for the subsequent values if you don't pass a tuple which fails
201        # for types such as pyodbc.SQL_WLONGVARCHAR, which is the datatype
202        # that ticket #5649 is targeting.
203
204        # NOTE: as of #6058, this won't be called if the use_setinputsizes
205        # parameter were not passed to the dialect, or if no types were
206        # specified in list_of_tuples
207
208        # as of #8177 for 2.0 we assume use_setinputsizes=True and only
209        # omit the setinputsizes calls for .executemany() with
210        # fast_executemany=True
211
212        if (
213            context.execute_style is interfaces.ExecuteStyle.EXECUTEMANY
214            and self.fast_executemany
215        ):
216            return
217
218        cursor.setinputsizes(
219            [
220                (
221                    (dbtype, None, None)
222                    if not isinstance(dbtype, tuple)
223                    else dbtype
224                )
225                for key, dbtype, sqltype in list_of_tuples
226            ]
227        )
228
229    def get_isolation_level_values(
230        self, dbapi_connection: interfaces.DBAPIConnection
231    ) -> List[IsolationLevel]:
232        return super().get_isolation_level_values(dbapi_connection) + [
233            "AUTOCOMMIT"
234        ]
235
236    def set_isolation_level(
237        self,
238        dbapi_connection: interfaces.DBAPIConnection,
239        level: IsolationLevel,
240    ) -> None:
241        # adjust for ConnectionFairy being present
242        # allows attribute set e.g. "connection.autocommit = True"
243        # to work properly
244
245        if level == "AUTOCOMMIT":
246            dbapi_connection.autocommit = True
247        else:
248            dbapi_connection.autocommit = False
249            super().set_isolation_level(dbapi_connection, level)
250