codekingpro/portable-devtools

1# inspection.py
2# Copyright (C) 2005-2024 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
8"""The inspection module provides the :func:`_sa.inspect` function,
9which delivers runtime information about a wide variety
10of SQLAlchemy objects, both within the Core as well as the
11ORM.
12
13The :func:`_sa.inspect` function is the entry point to SQLAlchemy's
14public API for viewing the configuration and construction
15of in-memory objects.   Depending on the type of object
16passed to :func:`_sa.inspect`, the return value will either be
17a related object which provides a known interface, or in many
18cases it will return the object itself.
19
20The rationale for :func:`_sa.inspect` is twofold.  One is that
21it replaces the need to be aware of a large variety of "information
22getting" functions in SQLAlchemy, such as
23:meth:`_reflection.Inspector.from_engine` (deprecated in 1.4),
24:func:`.orm.attributes.instance_state`, :func:`_orm.class_mapper`,
25and others.    The other is that the return value of :func:`_sa.inspect`
26is guaranteed to obey a documented API, thus allowing third party
27tools which build on top of SQLAlchemy configurations to be constructed
28in a forwards-compatible way.
29
30"""
31from __future__ import annotations
32
33from typing import Any
34from typing import Callable
35from typing import Dict
36from typing import Generic
37from typing import Optional
38from typing import overload
39from typing import Type
40from typing import TypeVar
41from typing import Union
42
43from . import exc
44from .util.typing import Literal
45from .util.typing import Protocol
46
47_T = TypeVar("_T", bound=Any)
48_TCov = TypeVar("_TCov", bound=Any, covariant=True)
49_F = TypeVar("_F", bound=Callable[..., Any])
50
51_IN = TypeVar("_IN", bound=Any)
52
53_registrars: Dict[type, Union[Literal[True], Callable[[Any], Any]]] = {}
54
55
56class Inspectable(Generic[_T]):
57    """define a class as inspectable.
58
59    This allows typing to set up a linkage between an object that
60    can be inspected and the type of inspection it returns.
61
62    Unfortunately we cannot at the moment get all classes that are
63    returned by inspection to suit this interface as we get into
64    MRO issues.
65
66    """
67
68    __slots__ = ()
69
70
71class _InspectableTypeProtocol(Protocol[_TCov]):
72    """a protocol defining a method that's used when a type (ie the class
73    itself) is passed to inspect().
74
75    """
76
77    def _sa_inspect_type(self) -> _TCov: ...
78
79
80class _InspectableProtocol(Protocol[_TCov]):
81    """a protocol defining a method that's used when an instance is
82    passed to inspect().
83
84    """
85
86    def _sa_inspect_instance(self) -> _TCov: ...
87
88
89@overload
90def inspect(
91    subject: Type[_InspectableTypeProtocol[_IN]], raiseerr: bool = True
92) -> _IN: ...
93
94
95@overload
96def inspect(
97    subject: _InspectableProtocol[_IN], raiseerr: bool = True
98) -> _IN: ...
99
100
101@overload
102def inspect(subject: Inspectable[_IN], raiseerr: bool = True) -> _IN: ...
103
104
105@overload
106def inspect(subject: Any, raiseerr: Literal[False] = ...) -> Optional[Any]: ...
107
108
109@overload
110def inspect(subject: Any, raiseerr: bool = True) -> Any: ...
111
112
113def inspect(subject: Any, raiseerr: bool = True) -> Any:
114    """Produce an inspection object for the given target.
115
116    The returned value in some cases may be the
117    same object as the one given, such as if a
118    :class:`_orm.Mapper` object is passed.   In other
119    cases, it will be an instance of the registered
120    inspection type for the given object, such as
121    if an :class:`_engine.Engine` is passed, an
122    :class:`_reflection.Inspector` object is returned.
123
124    :param subject: the subject to be inspected.
125    :param raiseerr: When ``True``, if the given subject
126     does not
127     correspond to a known SQLAlchemy inspected type,
128     :class:`sqlalchemy.exc.NoInspectionAvailable`
129     is raised.  If ``False``, ``None`` is returned.
130
131    """
132    type_ = type(subject)
133    for cls in type_.__mro__:
134        if cls in _registrars:
135            reg = _registrars.get(cls, None)
136            if reg is None:
137                continue
138            elif reg is True:
139                return subject
140            ret = reg(subject)
141            if ret is not None:
142                return ret
143    else:
144        reg = ret = None
145
146    if raiseerr and (reg is None or ret is None):
147        raise exc.NoInspectionAvailable(
148            "No inspection system is "
149            "available for object of type %s" % type_
150        )
151    return ret
152
153
154def _inspects(
155    *types: Type[Any],
156) -> Callable[[_F], _F]:
157    def decorate(fn_or_cls: _F) -> _F:
158        for type_ in types:
159            if type_ in _registrars:
160                raise AssertionError("Type %s is already registered" % type_)
161            _registrars[type_] = fn_or_cls
162        return fn_or_cls
163
164    return decorate
165
166
167_TT = TypeVar("_TT", bound="Type[Any]")
168
169
170def _self_inspects(cls: _TT) -> _TT:
171    if cls in _registrars:
172        raise AssertionError("Type %s is already registered" % cls)
173    _registrars[cls] = True
174    return cls
175