Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/mysql/cymysql.py: 54%

1# dialects/mysql/cymysql.py 

2# Copyright (C) 2005-2025 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