1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
# engine/characteristics.py
# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
from __future__ import annotations
import abc
import typing
from typing import Any
from typing import ClassVar
if typing.TYPE_CHECKING:
from .base import Connection
from .interfaces import DBAPIConnection
from .interfaces import Dialect
class ConnectionCharacteristic(abc.ABC):
"""An abstract base for an object that can set, get and reset a
per-connection characteristic, typically one that gets reset when the
connection is returned to the connection pool.
transaction isolation is the canonical example, and the
``IsolationLevelCharacteristic`` implementation provides this for the
``DefaultDialect``.
The ``ConnectionCharacteristic`` class should call upon the ``Dialect`` for
the implementation of each method. The object exists strictly to serve as
a dialect visitor that can be placed into the
``DefaultDialect.connection_characteristics`` dictionary where it will take
effect for calls to :meth:`_engine.Connection.execution_options` and
related APIs.
.. versionadded:: 1.4
"""
__slots__ = ()
transactional: ClassVar[bool] = False
@abc.abstractmethod
def reset_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection
) -> None:
"""Reset the characteristic on the DBAPI connection to its default
value."""
@abc.abstractmethod
def set_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection, value: Any
) -> None:
"""set characteristic on the DBAPI connection to a given value."""
def set_connection_characteristic(
self,
dialect: Dialect,
conn: Connection,
dbapi_conn: DBAPIConnection,
value: Any,
) -> None:
"""set characteristic on the :class:`_engine.Connection` to a given
value.
.. versionadded:: 2.0.30 - added to support elements that are local
to the :class:`_engine.Connection` itself.
"""
self.set_characteristic(dialect, dbapi_conn, value)
@abc.abstractmethod
def get_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection
) -> Any:
"""Given a DBAPI connection, get the current value of the
characteristic.
"""
def get_connection_characteristic(
self, dialect: Dialect, conn: Connection, dbapi_conn: DBAPIConnection
) -> Any:
"""Given a :class:`_engine.Connection`, get the current value of the
characteristic.
.. versionadded:: 2.0.30 - added to support elements that are local
to the :class:`_engine.Connection` itself.
"""
return self.get_characteristic(dialect, dbapi_conn)
class IsolationLevelCharacteristic(ConnectionCharacteristic):
"""Manage the isolation level on a DBAPI connection"""
transactional: ClassVar[bool] = True
def reset_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection
) -> None:
dialect.reset_isolation_level(dbapi_conn)
def set_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection, value: Any
) -> None:
dialect._assert_and_set_isolation_level(dbapi_conn, value)
def get_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection
) -> Any:
return dialect.get_isolation_level(dbapi_conn)
class LoggingTokenCharacteristic(ConnectionCharacteristic):
"""Manage the 'logging_token' option of a :class:`_engine.Connection`.
.. versionadded:: 2.0.30
"""
transactional: ClassVar[bool] = False
def reset_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection
) -> None:
pass
def set_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection, value: Any
) -> None:
raise NotImplementedError()
def set_connection_characteristic(
self,
dialect: Dialect,
conn: Connection,
dbapi_conn: DBAPIConnection,
value: Any,
) -> None:
if value:
conn._message_formatter = lambda msg: "[%s] %s" % (value, msg)
else:
del conn._message_formatter
def get_characteristic(
self, dialect: Dialect, dbapi_conn: DBAPIConnection
) -> Any:
raise NotImplementedError()
def get_connection_characteristic(
self, dialect: Dialect, conn: Connection, dbapi_conn: DBAPIConnection
) -> Any:
return conn._execution_options.get("logging_token", None)
|
