codekingpro/portable-devtools

1# ext/instrumentation.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
9"""Extensible class instrumentation.
10
11The :mod:`sqlalchemy.ext.instrumentation` package provides for alternate
12systems of class instrumentation within the ORM.  Class instrumentation
13refers to how the ORM places attributes on the class which maintain
14data and track changes to that data, as well as event hooks installed
15on the class.
16
17.. note::
18    The extension package is provided for the benefit of integration
19    with other object management packages, which already perform
20    their own instrumentation.  It is not intended for general use.
21
22For examples of how the instrumentation extension is used,
23see the example :ref:`examples_instrumentation`.
24
25"""
26import weakref
27
28from .. import util
29from ..orm import attributes
30from ..orm import base as orm_base
31from ..orm import collections
32from ..orm import exc as orm_exc
33from ..orm import instrumentation as orm_instrumentation
34from ..orm import util as orm_util
35from ..orm.instrumentation import _default_dict_getter
36from ..orm.instrumentation import _default_manager_getter
37from ..orm.instrumentation import _default_opt_manager_getter
38from ..orm.instrumentation import _default_state_getter
39from ..orm.instrumentation import ClassManager
40from ..orm.instrumentation import InstrumentationFactory
41
42
43INSTRUMENTATION_MANAGER = "__sa_instrumentation_manager__"
44"""Attribute, elects custom instrumentation when present on a mapped class.
45
46Allows a class to specify a slightly or wildly different technique for
47tracking changes made to mapped attributes and collections.
48
49Only one instrumentation implementation is allowed in a given object
50inheritance hierarchy.
51
52The value of this attribute must be a callable and will be passed a class
53object.  The callable must return one of:
54
55  - An instance of an :class:`.InstrumentationManager` or subclass
56  - An object implementing all or some of InstrumentationManager (TODO)
57  - A dictionary of callables, implementing all or some of the above (TODO)
58  - An instance of a :class:`.ClassManager` or subclass
59
60This attribute is consulted by SQLAlchemy instrumentation
61resolution, once the :mod:`sqlalchemy.ext.instrumentation` module
62has been imported.  If custom finders are installed in the global
63instrumentation_finders list, they may or may not choose to honor this
64attribute.
65
66"""
67
68
69def find_native_user_instrumentation_hook(cls):
70    """Find user-specified instrumentation management for a class."""
71    return getattr(cls, INSTRUMENTATION_MANAGER, None)
72
73
74instrumentation_finders = [find_native_user_instrumentation_hook]
75"""An extensible sequence of callables which return instrumentation
76implementations
77
78When a class is registered, each callable will be passed a class object.
79If None is returned, the
80next finder in the sequence is consulted.  Otherwise the return must be an
81instrumentation factory that follows the same guidelines as
82sqlalchemy.ext.instrumentation.INSTRUMENTATION_MANAGER.
83
84By default, the only finder is find_native_user_instrumentation_hook, which
85searches for INSTRUMENTATION_MANAGER.  If all finders return None, standard
86ClassManager instrumentation is used.
87
88"""
89
90
91class ExtendedInstrumentationRegistry(InstrumentationFactory):
92    """Extends :class:`.InstrumentationFactory` with additional
93    bookkeeping, to accommodate multiple types of
94    class managers.
95
96    """
97
98    _manager_finders = weakref.WeakKeyDictionary()
99    _state_finders = weakref.WeakKeyDictionary()
100    _dict_finders = weakref.WeakKeyDictionary()
101    _extended = False
102
103    def _locate_extended_factory(self, class_):
104        for finder in instrumentation_finders:
105            factory = finder(class_)
106            if factory is not None:
107                manager = self._extended_class_manager(class_, factory)
108                return manager, factory
109        else:
110            return None, None
111
112    def _check_conflicts(self, class_, factory):
113        existing_factories = self._collect_management_factories_for(
114            class_
115        ).difference([factory])
116        if existing_factories:
117            raise TypeError(
118                "multiple instrumentation implementations specified "
119                "in %s inheritance hierarchy: %r"
120                % (class_.__name__, list(existing_factories))
121            )
122
123    def _extended_class_manager(self, class_, factory):
124        manager = factory(class_)
125        if not isinstance(manager, ClassManager):
126            manager = _ClassInstrumentationAdapter(class_, manager)
127
128        if factory != ClassManager and not self._extended:
129            # somebody invoked a custom ClassManager.
130            # reinstall global "getter" functions with the more
131            # expensive ones.
132            self._extended = True
133            _install_instrumented_lookups()
134
135        self._manager_finders[class_] = manager.manager_getter()
136        self._state_finders[class_] = manager.state_getter()
137        self._dict_finders[class_] = manager.dict_getter()
138        return manager
139
140    def _collect_management_factories_for(self, cls):
141        """Return a collection of factories in play or specified for a
142        hierarchy.
143
144        Traverses the entire inheritance graph of a cls and returns a
145        collection of instrumentation factories for those classes. Factories
146        are extracted from active ClassManagers, if available, otherwise
147        instrumentation_finders is consulted.
148
149        """
150        hierarchy = util.class_hierarchy(cls)
151        factories = set()
152        for member in hierarchy:
153            manager = self.opt_manager_of_class(member)
154            if manager is not None:
155                factories.add(manager.factory)
156            else:
157                for finder in instrumentation_finders:
158                    factory = finder(member)
159                    if factory is not None:
160                        break
161                else:
162                    factory = None
163                factories.add(factory)
164        factories.discard(None)
165        return factories
166
167    def unregister(self, class_):
168        super().unregister(class_)
169        if class_ in self._manager_finders:
170            del self._manager_finders[class_]
171            del self._state_finders[class_]
172            del self._dict_finders[class_]
173
174    def opt_manager_of_class(self, cls):
175        try:
176            finder = self._manager_finders.get(
177                cls, _default_opt_manager_getter
178            )
179        except TypeError:
180            # due to weakref lookup on invalid object
181            return None
182        else:
183            return finder(cls)
184
185    def manager_of_class(self, cls):
186        try:
187            finder = self._manager_finders.get(cls, _default_manager_getter)
188        except TypeError:
189            # due to weakref lookup on invalid object
190            raise orm_exc.UnmappedClassError(
191                cls, f"Can't locate an instrumentation manager for class {cls}"
192            )
193        else:
194            manager = finder(cls)
195            if manager is None:
196                raise orm_exc.UnmappedClassError(
197                    cls,
198                    f"Can't locate an instrumentation manager for class {cls}",
199                )
200            return manager
201
202    def state_of(self, instance):
203        if instance is None:
204            raise AttributeError("None has no persistent state.")
205        return self._state_finders.get(
206            instance.__class__, _default_state_getter
207        )(instance)
208
209    def dict_of(self, instance):
210        if instance is None:
211            raise AttributeError("None has no persistent state.")
212        return self._dict_finders.get(
213            instance.__class__, _default_dict_getter
214        )(instance)
215
216
217orm_instrumentation._instrumentation_factory = _instrumentation_factory = (
218    ExtendedInstrumentationRegistry()
219)
220orm_instrumentation.instrumentation_finders = instrumentation_finders
221
222
223class InstrumentationManager:
224    """User-defined class instrumentation extension.
225
226    :class:`.InstrumentationManager` can be subclassed in order
227    to change
228    how class instrumentation proceeds. This class exists for
229    the purposes of integration with other object management
230    frameworks which would like to entirely modify the
231    instrumentation methodology of the ORM, and is not intended
232    for regular usage.  For interception of class instrumentation
233    events, see :class:`.InstrumentationEvents`.
234
235    The API for this class should be considered as semi-stable,
236    and may change slightly with new releases.
237
238    """
239
240    # r4361 added a mandatory (cls) constructor to this interface.
241    # given that, perhaps class_ should be dropped from all of these
242    # signatures.
243
244    def __init__(self, class_):
245        pass
246
247    def manage(self, class_, manager):
248        setattr(class_, "_default_class_manager", manager)
249
250    def unregister(self, class_, manager):
251        delattr(class_, "_default_class_manager")
252
253    def manager_getter(self, class_):
254        def get(cls):
255            return cls._default_class_manager
256
257        return get
258
259    def instrument_attribute(self, class_, key, inst):
260        pass
261
262    def post_configure_attribute(self, class_, key, inst):
263        pass
264
265    def install_descriptor(self, class_, key, inst):
266        setattr(class_, key, inst)
267
268    def uninstall_descriptor(self, class_, key):
269        delattr(class_, key)
270
271    def install_member(self, class_, key, implementation):
272        setattr(class_, key, implementation)
273
274    def uninstall_member(self, class_, key):
275        delattr(class_, key)
276
277    def instrument_collection_class(self, class_, key, collection_class):
278        return collections.prepare_instrumentation(collection_class)
279
280    def get_instance_dict(self, class_, instance):
281        return instance.__dict__
282
283    def initialize_instance_dict(self, class_, instance):
284        pass
285
286    def install_state(self, class_, instance, state):
287        setattr(instance, "_default_state", state)
288
289    def remove_state(self, class_, instance):
290        delattr(instance, "_default_state")
291
292    def state_getter(self, class_):
293        return lambda instance: getattr(instance, "_default_state")
294
295    def dict_getter(self, class_):
296        return lambda inst: self.get_instance_dict(class_, inst)
297
298
299class _ClassInstrumentationAdapter(ClassManager):
300    """Adapts a user-defined InstrumentationManager to a ClassManager."""
301
302    def __init__(self, class_, override):
303        self._adapted = override
304        self._get_state = self._adapted.state_getter(class_)
305        self._get_dict = self._adapted.dict_getter(class_)
306
307        ClassManager.__init__(self, class_)
308
309    def manage(self):
310        self._adapted.manage(self.class_, self)
311
312    def unregister(self):
313        self._adapted.unregister(self.class_, self)
314
315    def manager_getter(self):
316        return self._adapted.manager_getter(self.class_)
317
318    def instrument_attribute(self, key, inst, propagated=False):
319        ClassManager.instrument_attribute(self, key, inst, propagated)
320        if not propagated:
321            self._adapted.instrument_attribute(self.class_, key, inst)
322
323    def post_configure_attribute(self, key):
324        super().post_configure_attribute(key)
325        self._adapted.post_configure_attribute(self.class_, key, self[key])
326
327    def install_descriptor(self, key, inst):
328        self._adapted.install_descriptor(self.class_, key, inst)
329
330    def uninstall_descriptor(self, key):
331        self._adapted.uninstall_descriptor(self.class_, key)
332
333    def install_member(self, key, implementation):
334        self._adapted.install_member(self.class_, key, implementation)
335
336    def uninstall_member(self, key):
337        self._adapted.uninstall_member(self.class_, key)
338
339    def instrument_collection_class(self, key, collection_class):
340        return self._adapted.instrument_collection_class(
341            self.class_, key, collection_class
342        )
343
344    def initialize_collection(self, key, state, factory):
345        delegate = getattr(self._adapted, "initialize_collection", None)
346        if delegate:
347            return delegate(key, state, factory)
348        else:
349            return ClassManager.initialize_collection(
350                self, key, state, factory
351            )
352
353    def new_instance(self, state=None):
354        instance = self.class_.__new__(self.class_)
355        self.setup_instance(instance, state)
356        return instance
357
358    def _new_state_if_none(self, instance):
359        """Install a default InstanceState if none is present.
360
361        A private convenience method used by the __init__ decorator.
362        """
363        if self.has_state(instance):
364            return False
365        else:
366            return self.setup_instance(instance)
367
368    def setup_instance(self, instance, state=None):
369        self._adapted.initialize_instance_dict(self.class_, instance)
370
371        if state is None:
372            state = self._state_constructor(instance, self)
373
374        # the given instance is assumed to have no state
375        self._adapted.install_state(self.class_, instance, state)
376        return state
377
378    def teardown_instance(self, instance):
379        self._adapted.remove_state(self.class_, instance)
380
381    def has_state(self, instance):
382        try:
383            self._get_state(instance)
384        except orm_exc.NO_STATE:
385            return False
386        else:
387            return True
388
389    def state_getter(self):
390        return self._get_state
391
392    def dict_getter(self):
393        return self._get_dict
394
395
396def _install_instrumented_lookups():
397    """Replace global class/object management functions
398    with ExtendedInstrumentationRegistry implementations, which
399    allow multiple types of class managers to be present,
400    at the cost of performance.
401
402    This function is called only by ExtendedInstrumentationRegistry
403    and unit tests specific to this behavior.
404
405    The _reinstall_default_lookups() function can be called
406    after this one to re-establish the default functions.
407
408    """
409    _install_lookups(
410        dict(
411            instance_state=_instrumentation_factory.state_of,
412            instance_dict=_instrumentation_factory.dict_of,
413            manager_of_class=_instrumentation_factory.manager_of_class,
414            opt_manager_of_class=_instrumentation_factory.opt_manager_of_class,
415        )
416    )
417
418
419def _reinstall_default_lookups():
420    """Restore simplified lookups."""
421    _install_lookups(
422        dict(
423            instance_state=_default_state_getter,
424            instance_dict=_default_dict_getter,
425            manager_of_class=_default_manager_getter,
426            opt_manager_of_class=_default_opt_manager_getter,
427        )
428    )
429    _instrumentation_factory._extended = False
430
431
432def _install_lookups(lookups):
433    global instance_state, instance_dict
434    global manager_of_class, opt_manager_of_class
435    instance_state = lookups["instance_state"]
436    instance_dict = lookups["instance_dict"]
437    manager_of_class = lookups["manager_of_class"]
438    opt_manager_of_class = lookups["opt_manager_of_class"]
439    orm_base.instance_state = attributes.instance_state = (
440        orm_instrumentation.instance_state
441    ) = instance_state
442    orm_base.instance_dict = attributes.instance_dict = (
443        orm_instrumentation.instance_dict
444    ) = instance_dict
445    orm_base.manager_of_class = attributes.manager_of_class = (
446        orm_instrumentation.manager_of_class
447    ) = manager_of_class
448    orm_base.opt_manager_of_class = orm_util.opt_manager_of_class = (
449        attributes.opt_manager_of_class
450    ) = orm_instrumentation.opt_manager_of_class = opt_manager_of_class
451