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

1# mysql/cymysql.py 

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

7r""" 

8 

9.. dialect:: mysql+cymysql 

10 :name: CyMySQL 

11 :dbapi: cymysql 

12 :connectstring: mysql+cymysql://<username>:<password>@<host>/<dbname>[?<options>] 

13 :url: https://github.com/nakagami/CyMySQL 

14 

15.. note:: 

16 

17 The CyMySQL dialect is **not tested as part of SQLAlchemy's continuous 

18 integration** and may have unresolved issues. The recommended MySQL 

19 dialects are mysqlclient and PyMySQL. 

20 

21""" # noqa 

22 

23from .base import BIT 

24from .base import MySQLDialect 

25from .mysqldb import MySQLDialect_mysqldb 

26from ... import util 

27 

28 

29class _cymysqlBIT(BIT): 

30 def result_processor(self, dialect, coltype): 

31 """Convert MySQL's 64 bit, variable length binary string to a long.""" 

32 

33 def process(value): 

34 if value is not None: 

35 v = 0 

36 for i in util.iterbytes(value): 

37 v = v << 8 | i 

38 return v 

39 return value 

40 

41 return process 

42 

43 

44class MySQLDialect_cymysql(MySQLDialect_mysqldb): 

45 driver = "cymysql" 

46 supports_statement_cache = True 

47 

48 description_encoding = None 

49 supports_sane_rowcount = True 

50 supports_sane_multi_rowcount = False 

51 supports_unicode_statements = True 

52 

53 colspecs = util.update_copy(MySQLDialect.colspecs, {BIT: _cymysqlBIT}) 

54 

55 @classmethod 

56 def dbapi(cls): 

57 return __import__("cymysql") 

58 

59 def _detect_charset(self, connection): 

60 return connection.connection.charset 

61 

62 def _extract_error_code(self, exception): 

63 return exception.errno 

64 

65 def is_disconnect(self, e, connection, cursor): 

66 if isinstance(e, self.dbapi.OperationalError): 

67 return self._extract_error_code(e) in ( 

68 2006, 

69 2013, 

70 2014, 

71 2045, 

72 2055, 

73 ) 

74 elif isinstance(e, self.dbapi.InterfaceError): 

75 # if underlying connection is closed, 

76 # this is the error you get 

77 return True 

78 else: 

79 return False 

80 

81 

82dialect = MySQLDialect_cymysql