Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/characteristics.py: 64%

1# engine/characteristics.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 

7from __future__ import annotations 

8 

9import abc 

10import typing 

11from typing import Any 

12from typing import ClassVar 

13 

14if typing.TYPE_CHECKING: 

15 from .base import Connection 

16 from .interfaces import DBAPIConnection 

17 from .interfaces import Dialect 

18 

19 

20class ConnectionCharacteristic(abc.ABC): 

21 """An abstract base for an object that can set, get and reset a 

22 per-connection characteristic, typically one that gets reset when the 

23 connection is returned to the connection pool. 

24 

25 transaction isolation is the canonical example, and the 

26 ``IsolationLevelCharacteristic`` implementation provides this for the 

27 ``DefaultDialect``. 

28 

29 The ``ConnectionCharacteristic`` class should call upon the ``Dialect`` for 

30 the implementation of each method. The object exists strictly to serve as 

31 a dialect visitor that can be placed into the 

32 ``DefaultDialect.connection_characteristics`` dictionary where it will take 

33 effect for calls to :meth:`_engine.Connection.execution_options` and 

34 related APIs. 

35 

36 .. versionadded:: 1.4 

37 

38 """ 

39 

40 __slots__ = () 

41 

42 transactional: ClassVar[bool] = False 

43 

44 @abc.abstractmethod 

45 def reset_characteristic( 

46 self, dialect: Dialect, dbapi_conn: DBAPIConnection 

47 ) -> None: 

48 """Reset the characteristic on the DBAPI connection to its default 

49 value.""" 

50 

51 @abc.abstractmethod 

52 def set_characteristic( 

53 self, dialect: Dialect, dbapi_conn: DBAPIConnection, value: Any 

54 ) -> None: 

55 """set characteristic on the DBAPI connection to a given value.""" 

56 

57 def set_connection_characteristic( 

58 self, 

59 dialect: Dialect, 

60 conn: Connection, 

61 dbapi_conn: DBAPIConnection, 

62 value: Any, 

63 ) -> None: 

64 """set characteristic on the :class:`_engine.Connection` to a given 

65 value. 

66 

67 .. versionadded:: 2.0.30 - added to support elements that are local 

68 to the :class:`_engine.Connection` itself. 

69 

70 """ 

71 self.set_characteristic(dialect, dbapi_conn, value) 

72 

73 @abc.abstractmethod 

74 def get_characteristic( 

75 self, dialect: Dialect, dbapi_conn: DBAPIConnection 

76 ) -> Any: 

77 """Given a DBAPI connection, get the current value of the 

78 characteristic. 

79 

80 """ 

81 

82 def get_connection_characteristic( 

83 self, dialect: Dialect, conn: Connection, dbapi_conn: DBAPIConnection 

84 ) -> Any: 

85 """Given a :class:`_engine.Connection`, get the current value of the 

86 characteristic. 

87 

88 .. versionadded:: 2.0.30 - added to support elements that are local 

89 to the :class:`_engine.Connection` itself. 

90 

91 """ 

92 return self.get_characteristic(dialect, dbapi_conn) 

93 

94 

95class IsolationLevelCharacteristic(ConnectionCharacteristic): 

96 """Manage the isolation level on a DBAPI connection""" 

97 

98 transactional: ClassVar[bool] = True 

99 

100 def reset_characteristic( 

101 self, dialect: Dialect, dbapi_conn: DBAPIConnection 

102 ) -> None: 

103 dialect.reset_isolation_level(dbapi_conn) 

104 

105 def set_characteristic( 

106 self, dialect: Dialect, dbapi_conn: DBAPIConnection, value: Any 

107 ) -> None: 

108 dialect._assert_and_set_isolation_level(dbapi_conn, value) 

109 

110 def get_characteristic( 

111 self, dialect: Dialect, dbapi_conn: DBAPIConnection 

112 ) -> Any: 

113 return dialect.get_isolation_level(dbapi_conn) 

114 

115 

116class LoggingTokenCharacteristic(ConnectionCharacteristic): 

117 """Manage the 'logging_token' option of a :class:`_engine.Connection`. 

118 

119 .. versionadded:: 2.0.30 

120 

121 """ 

122 

123 transactional: ClassVar[bool] = False 

124 

125 def reset_characteristic( 

126 self, dialect: Dialect, dbapi_conn: DBAPIConnection 

127 ) -> None: 

128 pass 

129 

130 def set_characteristic( 

131 self, dialect: Dialect, dbapi_conn: DBAPIConnection, value: Any 

132 ) -> None: 

133 raise NotImplementedError() 

134 

135 def set_connection_characteristic( 

136 self, 

137 dialect: Dialect, 

138 conn: Connection, 

139 dbapi_conn: DBAPIConnection, 

140 value: Any, 

141 ) -> None: 

142 if value: 

143 conn._message_formatter = lambda msg: "[%s] %s" % (value, msg) 

144 else: 

145 del conn._message_formatter 

146 

147 def get_characteristic( 

148 self, dialect: Dialect, dbapi_conn: DBAPIConnection 

149 ) -> Any: 

150 raise NotImplementedError() 

151 

152 def get_connection_characteristic( 

153 self, dialect: Dialect, conn: Connection, dbapi_conn: DBAPIConnection 

154 ) -> Any: 

155 return conn._execution_options.get("logging_token", None)