codekingpro/portable-devtools

1# dialects/mysql/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# mypy: ignore-errors
8
9
10r"""
11
12
13.. dialect:: mysql+pyodbc
14    :name: PyODBC
15    :dbapi: pyodbc
16    :connectstring: mysql+pyodbc://<username>:<password>@<dsnname>
17    :url: https://pypi.org/project/pyodbc/
18
19.. note::
20
21    The PyODBC for MySQL dialect is **not tested as part of
22    SQLAlchemy's continuous integration**.
23    The recommended MySQL dialects are mysqlclient and PyMySQL.
24    However, if you want to use the mysql+pyodbc dialect and require
25    full support for ``utf8mb4`` characters (including supplementary
26    characters like emoji) be sure to use a current release of
27    MySQL Connector/ODBC and specify the "ANSI" (**not** "Unicode")
28    version of the driver in your DSN or connection string.
29
30Pass through exact pyodbc connection string::
31
32    import urllib
33    connection_string = (
34        'DRIVER=MySQL ODBC 8.0 ANSI Driver;'
35        'SERVER=localhost;'
36        'PORT=3307;'
37        'DATABASE=mydb;'
38        'UID=root;'
39        'PWD=(whatever);'
40        'charset=utf8mb4;'
41    )
42    params = urllib.parse.quote_plus(connection_string)
43    connection_uri = "mysql+pyodbc:///?odbc_connect=%s" % params
44
45"""  # noqa
46
47import re
48
49from .base import MySQLDialect
50from .base import MySQLExecutionContext
51from .types import TIME
52from ... import exc
53from ... import util
54from ...connectors.pyodbc import PyODBCConnector
55from ...sql.sqltypes import Time
56
57
58class _pyodbcTIME(TIME):
59    def result_processor(self, dialect, coltype):
60        def process(value):
61            # pyodbc returns a datetime.time object; no need to convert
62            return value
63
64        return process
65
66
67class MySQLExecutionContext_pyodbc(MySQLExecutionContext):
68    def get_lastrowid(self):
69        cursor = self.create_cursor()
70        cursor.execute("SELECT LAST_INSERT_ID()")
71        lastrowid = cursor.fetchone()[0]
72        cursor.close()
73        return lastrowid
74
75
76class MySQLDialect_pyodbc(PyODBCConnector, MySQLDialect):
77    supports_statement_cache = True
78    colspecs = util.update_copy(MySQLDialect.colspecs, {Time: _pyodbcTIME})
79    supports_unicode_statements = True
80    execution_ctx_cls = MySQLExecutionContext_pyodbc
81
82    pyodbc_driver_name = "MySQL"
83
84    def _detect_charset(self, connection):
85        """Sniff out the character set in use for connection results."""
86
87        # Prefer 'character_set_results' for the current connection over the
88        # value in the driver.  SET NAMES or individual variable SETs will
89        # change the charset without updating the driver's view of the world.
90        #
91        # If it's decided that issuing that sort of SQL leaves you SOL, then
92        # this can prefer the driver value.
93
94        # set this to None as _fetch_setting attempts to use it (None is OK)
95        self._connection_charset = None
96        try:
97            value = self._fetch_setting(connection, "character_set_client")
98            if value:
99                return value
100        except exc.DBAPIError:
101            pass
102
103        util.warn(
104            "Could not detect the connection character set.  "
105            "Assuming latin1."
106        )
107        return "latin1"
108
109    def _get_server_version_info(self, connection):
110        return MySQLDialect._get_server_version_info(self, connection)
111
112    def _extract_error_code(self, exception):
113        m = re.compile(r"\((\d+)\)").search(str(exception.args))
114        c = m.group(1)
115        if c:
116            return int(c)
117        else:
118            return None
119
120    def on_connect(self):
121        super_ = super().on_connect()
122
123        def on_connect(conn):
124            if super_ is not None:
125                super_(conn)
126
127            # declare Unicode encoding for pyodbc as per
128            #   https://github.com/mkleehammer/pyodbc/wiki/Unicode
129            pyodbc_SQL_CHAR = 1  # pyodbc.SQL_CHAR
130            pyodbc_SQL_WCHAR = -8  # pyodbc.SQL_WCHAR
131            conn.setdecoding(pyodbc_SQL_CHAR, encoding="utf-8")
132            conn.setdecoding(pyodbc_SQL_WCHAR, encoding="utf-8")
133            conn.setencoding(encoding="utf-8")
134
135        return on_connect
136
137
138dialect = MySQLDialect_pyodbc
139