Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/connectors/aioodbc.py: 55%

1# connectors/aioodbc.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# mypy: ignore-errors 

8 

9from __future__ import annotations 

10 

11from typing import TYPE_CHECKING 

12 

13from .asyncio import AsyncAdapt_dbapi_connection 

14from .asyncio import AsyncAdapt_dbapi_cursor 

15from .asyncio import AsyncAdapt_dbapi_ss_cursor 

16from .pyodbc import PyODBCConnector 

17from ..connectors.asyncio import AsyncAdapt_dbapi_module 

18from ..util.concurrency import await_ 

19 

20if TYPE_CHECKING: 

21 from ..engine.interfaces import ConnectArgsType 

22 from ..engine.url import URL 

23 

24 

25class AsyncAdapt_aioodbc_cursor(AsyncAdapt_dbapi_cursor): 

26 __slots__ = () 

27 

28 def setinputsizes(self, *inputsizes): 

29 # see https://github.com/aio-libs/aioodbc/issues/451 

30 return self._cursor._impl.setinputsizes(*inputsizes) 

31 

32 # how it's supposed to work 

33 # return await_(self._cursor.setinputsizes(*inputsizes)) 

34 

35 @property 

36 def fast_executemany(self): 

37 return self._cursor._impl.fast_executemany 

38 

39 @fast_executemany.setter 

40 def fast_executemany(self, value): 

41 self._cursor._impl.fast_executemany = value 

42 

43 

44class AsyncAdapt_aioodbc_ss_cursor( 

45 AsyncAdapt_aioodbc_cursor, AsyncAdapt_dbapi_ss_cursor 

46): 

47 __slots__ = () 

48 

49 

50class AsyncAdapt_aioodbc_connection(AsyncAdapt_dbapi_connection): 

51 _cursor_cls = AsyncAdapt_aioodbc_cursor 

52 _ss_cursor_cls = AsyncAdapt_aioodbc_ss_cursor 

53 __slots__ = () 

54 

55 @property 

56 def autocommit(self): 

57 return self._connection.autocommit 

58 

59 @autocommit.setter 

60 def autocommit(self, value): 

61 # https://github.com/aio-libs/aioodbc/issues/448 

62 # self._connection.autocommit = value 

63 

64 self._connection._conn.autocommit = value 

65 

66 def ping(self, reconnect): 

67 return await_(self._connection.ping(reconnect)) 

68 

69 def add_output_converter(self, *arg, **kw): 

70 self._connection.add_output_converter(*arg, **kw) 

71 

72 def character_set_name(self): 

73 return self._connection.character_set_name() 

74 

75 def cursor(self, server_side=False): 

76 # aioodbc sets connection=None when closed and just fails with 

77 # AttributeError here. Here we use the same ProgrammingError + 

78 # message that pyodbc uses, so it triggers is_disconnect() as well. 

79 if self._connection.closed: 

80 raise self.dbapi.ProgrammingError( 

81 "Attempt to use a closed connection." 

82 ) 

83 return super().cursor(server_side=server_side) 

84 

85 def rollback(self): 

86 # aioodbc sets connection=None when closed and just fails with 

87 # AttributeError here. should be a no-op 

88 if not self._connection.closed: 

89 super().rollback() 

90 

91 def commit(self): 

92 # aioodbc sets connection=None when closed and just fails with 

93 # AttributeError here. should be a no-op 

94 if not self._connection.closed: 

95 super().commit() 

96 

97 def close(self): 

98 # aioodbc sets connection=None when closed and just fails with 

99 # AttributeError here. should be a no-op 

100 if not self._connection.closed: 

101 super().close() 

102 

103 

104class AsyncAdapt_aioodbc_dbapi(AsyncAdapt_dbapi_module): 

105 def __init__(self, aioodbc, pyodbc): 

106 super().__init__(aioodbc, dbapi_module=pyodbc) 

107 self.aioodbc = aioodbc 

108 self.pyodbc = pyodbc 

109 self.paramstyle = pyodbc.paramstyle 

110 self._init_dbapi_attributes() 

111 self.Cursor = AsyncAdapt_dbapi_cursor 

112 self.version = pyodbc.version 

113 

114 def _init_dbapi_attributes(self): 

115 for name in ( 

116 "Warning", 

117 "Error", 

118 "InterfaceError", 

119 "DataError", 

120 "DatabaseError", 

121 "OperationalError", 

122 "InterfaceError", 

123 "IntegrityError", 

124 "ProgrammingError", 

125 "InternalError", 

126 "NotSupportedError", 

127 "SQL_DRIVER_NAME", 

128 "NUMBER", 

129 "STRING", 

130 "DATETIME", 

131 "BINARY", 

132 "Binary", 

133 "BinaryNull", 

134 "SQL_VARCHAR", 

135 "SQL_WVARCHAR", 

136 "SQL_DECIMAL", 

137 ): 

138 setattr(self, name, getattr(self.pyodbc, name)) 

139 

140 def connect(self, *arg, **kw): 

141 creator_fn = kw.pop("async_creator_fn", self.aioodbc.connect) 

142 

143 return await_( 

144 AsyncAdapt_aioodbc_connection.create( 

145 self, 

146 creator_fn(*arg, **kw), 

147 ) 

148 ) 

149 

150 

151class aiodbcConnector(PyODBCConnector): 

152 is_async = True 

153 supports_statement_cache = True 

154 

155 supports_server_side_cursors = True 

156 

157 @classmethod 

158 def import_dbapi(cls): 

159 return AsyncAdapt_aioodbc_dbapi( 

160 __import__("aioodbc"), __import__("pyodbc") 

161 ) 

162 

163 def create_connect_args(self, url: URL) -> ConnectArgsType: 

164 arg, kw = super().create_connect_args(url) 

165 if arg and arg[0]: 

166 kw["dsn"] = arg[0] 

167 

168 return (), kw 

169 

170 def get_driver_connection(self, connection): 

171 return connection._connection