codekingpro/portable-devtools

1# dialects/mysql/cymysql.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
8r"""
9
10.. dialect:: mysql+cymysql
11    :name: CyMySQL
12    :dbapi: cymysql
13    :connectstring: mysql+cymysql://<username>:<password>@<host>/<dbname>[?<options>]
14    :url: https://github.com/nakagami/CyMySQL
15
16.. note::
17
18    The CyMySQL dialect is **not tested as part of SQLAlchemy's continuous
19    integration** and may have unresolved issues.  The recommended MySQL
20    dialects are mysqlclient and PyMySQL.
21
22"""  # noqa
23from __future__ import annotations
24
25from typing import Any
26from typing import Iterable
27from typing import Optional
28from typing import TYPE_CHECKING
29from typing import Union
30
31from .base import MySQLDialect
32from .mysqldb import MySQLDialect_mysqldb
33from .types import BIT
34from ... import util
35
36if TYPE_CHECKING:
37    from ...engine.base import Connection
38    from ...engine.interfaces import DBAPIConnection
39    from ...engine.interfaces import DBAPICursor
40    from ...engine.interfaces import DBAPIModule
41    from ...engine.interfaces import Dialect
42    from ...engine.interfaces import PoolProxiedConnection
43    from ...sql.type_api import _ResultProcessorType
44
45
46class _cymysqlBIT(BIT):
47    def result_processor(
48        self, dialect: Dialect, coltype: object
49    ) -> Optional[_ResultProcessorType[Any]]:
50        """Convert MySQL's 64 bit, variable length binary string to a long."""
51
52        def process(value: Optional[Iterable[int]]) -> Optional[int]:
53            if value is not None:
54                v = 0
55                for i in iter(value):
56                    v = v << 8 | i
57                return v
58            return value
59
60        return process
61
62
63class MySQLDialect_cymysql(MySQLDialect_mysqldb):
64    driver = "cymysql"
65    supports_statement_cache = True
66
67    description_encoding = None
68    supports_sane_rowcount = True
69    supports_sane_multi_rowcount = False
70    supports_unicode_statements = True
71
72    colspecs = util.update_copy(MySQLDialect.colspecs, {BIT: _cymysqlBIT})
73
74    @classmethod
75    def import_dbapi(cls) -> DBAPIModule:
76        return __import__("cymysql")
77
78    def _detect_charset(self, connection: Connection) -> str:
79        return connection.connection.charset  # type: ignore[no-any-return]
80
81    def _extract_error_code(self, exception: DBAPIModule.Error) -> int:
82        return exception.errno  # type: ignore[no-any-return]
83
84    def is_disconnect(
85        self,
86        e: DBAPIModule.Error,
87        connection: Optional[Union[PoolProxiedConnection, DBAPIConnection]],
88        cursor: Optional[DBAPICursor],
89    ) -> bool:
90        if isinstance(e, self.loaded_dbapi.OperationalError):
91            return self._extract_error_code(e) in (
92                2006,
93                2013,
94                2014,
95                2045,
96                2055,
97            )
98        elif isinstance(e, self.loaded_dbapi.InterfaceError):
99            # if underlying connection is closed,
100            # this is the error you get
101            return True
102        else:
103            return False
104
105
106dialect = MySQLDialect_cymysql
107