codekingpro/portable-devtools

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