codekingpro/portable-devtools

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