Edit: cache_key.py
# sql/cache_key.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 enum from itertools import zip_longest import typing from typing import Any from typing import Callable from typing import Dict from typing import Iterable from typing import Iterator from typing import List from typing import MutableMapping from typing import NamedTuple from typing import Optional from typing import Sequence from typing import Tuple from typing import Union from .visitors import anon_map from .visitors import HasTraversalDispatch from .visitors import HasTraverseInternals from .visitors import InternalTraversal from .visitors import prefix_anon_map from .. import util from ..inspection import inspect from ..util import HasMemoized from ..util.typing import Literal from ..util.typing import Protocol if typing.TYPE_CHECKING: from .elements import BindParameter from .elements import ClauseElement from .elements import ColumnElement from .visitors import _TraverseInternalsType from ..engine.interfaces import _CoreSingleExecuteParams class _CacheKeyTraversalDispatchType(Protocol): def __call__( s, self: HasCacheKey, visitor: _CacheKeyTraversal ) -> _CacheKeyTraversalDispatchTypeReturn: ... class CacheConst(enum.Enum): NO_CACHE = 0 NO_CACHE = CacheConst.NO_CACHE _CacheKeyTraversalType = Union[ "_TraverseInternalsType", Literal[CacheConst.NO_CACHE], Literal[None] ] class CacheTraverseTarget(enum.Enum): CACHE_IN_PLACE = 0 CALL_GEN_CACHE_KEY = 1 STATIC_CACHE_KEY = 2 PROPAGATE_ATTRS = 3 ANON_NAME = 4 ( CACHE_IN_PLACE, CALL_GEN_CACHE_KEY, STATIC_CACHE_KEY, PROPAGATE_ATTRS, ANON_NAME, ) = tuple(CacheTraverseTarget) _CacheKeyTraversalDispatchTypeReturn = Sequence[ Tuple[ str, Any, Union[ Callable[..., Tuple[Any, ...]], CacheTraverseTarget, InternalTraversal, ], ] ] class HasCacheKey: """Mixin for objects which can produce a cache key. This class is usually in a hierarchy that starts with the :class:`.HasTraverseInternals` base, but this is optional. Currently, the class should be able to work on its own without including :class:`.HasTraverseInternals`. .. seealso:: :class:`.CacheKey` :ref:`sql_caching` """ __slots__ = () _cache_key_traversal: _CacheKeyTraversalType = NO_CACHE _is_has_cache_key = True _hierarchy_supports_caching = True """private attribute which may be set to False to prevent the inherit_cache warning from being emitted for a hierarchy of subclasses. Currently applies to the :class:`.ExecutableDDLElement` hierarchy which does not implement caching. """ inherit_cache: Optional[bool] = None """Indicate if this :class:`.HasCacheKey` instance should make use of the cache key generation scheme used by its immediate superclass. The attribute defaults to ``None``, which indicates that a construct has not yet taken into account whether or not its appropriate for it to participate in caching; this is functionally equivalent to setting the value to ``False``, except that a warning is also emitted. This flag can be set to ``True`` on a particular class, if the SQL that corresponds to the object does not change based on attributes which are local to this class, and not its superclass. .. seealso:: :ref:`compilerext_caching` - General guideslines for setting the :attr:`.HasCacheKey.inherit_cache` attribute for third-party or user defined SQL constructs. """ __slots__ = () _generated_cache_key_traversal: Any @classmethod def _generate_cache_attrs( cls, ) -> Union[_CacheKeyTraversalDispatchType, Literal[CacheConst.NO_CACHE]]: """generate cache key dispatcher for a new class. This sets the _generated_cache_key_traversal attribute once called so should only be called once per class. """ inherit_cache = cls.__dict__.get("inherit_cache", None) inherit = bool(inherit_cache) if inherit: _cache_key_traversal = getattr(cls, "_cache_key_traversal", None) if _cache_key_traversal is None: try: assert issubclass(cls, HasTraverseInternals) _cache_key_traversal = cls._traverse_internals except AttributeError: cls._generated_cache_key_traversal = NO_CACHE return NO_CACHE assert _cache_key_traversal is not NO_CACHE, ( f"class {cls} has _cache_key_traversal=NO_CACHE, " "which conflicts with inherit_cache=True" ) # TODO: wouldn't we instead get this from our superclass? # also, our superclass may not have this yet, but in any case, # we'd generate for the superclass that has it. this is a little # more complicated, so for the moment this is a little less # efficient on startup but simpler. return _cache_key_traversal_visitor.generate_dispatch( cls, _cache_key_traversal, "_generated_cache_key_traversal", ) else: _cache_key_traversal = cls.__dict__.get( "_cache_key_traversal", None ) if _cache_key_traversal is None: _cache_key_traversal = cls.__dict__.get( "_traverse_internals", None ) if _cache_key_traversal is None: cls._generated_cache_key_traversal = NO_CACHE if ( inherit_cache is None and cls._hierarchy_supports_caching ): util.warn( "Class %s will not make use of SQL compilation " "caching as it does not set the 'inherit_cache' " "attribute to ``True``. This can have " "significant performance implications including " "some performance degradations in comparison to " "prior SQLAlchemy versions. Set this attribute " "to True if this object can make use of the cache " "key generated by the superclass. Alternatively, " "this attribute may be set to False which will " "disable this warning." % (cls.__name__), code="cprf", ) return NO_CACHE return _cache_key_traversal_visitor.generate_dispatch( cls, _cache_key_traversal, "_generated_cache_key_traversal", ) @util.preload_module("sqlalchemy.sql.elements") def _gen_cache_key( self, anon_map: anon_map, bindparams: List[BindParameter[Any]] ) -> Optional[Tuple[Any, ...]]: """return an optional cache key. The cache key is a tuple which can contain any series of objects that are hashable and also identifies this object uniquely within the presence of a larger SQL expression or statement, for the purposes of caching the resulting query. The cache key should be based on the SQL compiled structure that would ultimately be produced. That is, two structures that are composed in exactly the same way should produce the same cache key; any difference in the structures that would affect the SQL string or the type handlers should result in a different cache key. If a structure cannot produce a useful cache key, the NO_CACHE symbol should be added to the anon_map and the method should return None. """ cls = self.__class__ id_, found = anon_map.get_anon(self) if found: return (id_, cls) dispatcher: Union[ Literal[CacheConst.NO_CACHE], _CacheKeyTraversalDispatchType, ] try: dispatcher = cls.__dict__["_generated_cache_key_traversal"] except KeyError: # traversals.py -> _preconfigure_traversals() # may be used to run these ahead of time, but # is not enabled right now. # this block will generate any remaining dispatchers. dispatcher = cls._generate_cache_attrs() if dispatcher is NO_CACHE: anon_map[NO_CACHE] = True return None result: Tuple[Any, ...] = (id_, cls) # inline of _cache_key_traversal_visitor.run_generated_dispatch() for attrname, obj, meth in dispatcher( self, _cache_key_traversal_visitor ): if obj is not None: # TODO: see if C code can help here as Python lacks an # efficient switch construct if meth is STATIC_CACHE_KEY: sck = obj._static_cache_key if sck is NO_CACHE: anon_map[NO_CACHE] = True return None result += (attrname, sck) elif meth is ANON_NAME: elements = util.preloaded.sql_elements if isinstance(obj, elements._anonymous_label): obj = obj.apply_map(anon_map) # type: ignore result += (attrname, obj) elif meth is CALL_GEN_CACHE_KEY: result += ( attrname, obj._gen_cache_key(anon_map, bindparams), ) # remaining cache functions are against # Python tuples, dicts, lists, etc. so we can skip # if they are empty elif obj: if meth is CACHE_IN_PLACE: result += (attrname, obj) elif meth is PROPAGATE_ATTRS: result += ( attrname, obj["compile_state_plugin"], ( obj["plugin_subject"]._gen_cache_key( anon_map, bindparams ) if obj["plugin_subject"] else None ), ) elif meth is InternalTraversal.dp_annotations_key: # obj is here is the _annotations dict. Table uses # a memoized version of it. however in other cases, # we generate it given anon_map as we may be from a # Join, Aliased, etc. # see #8790 if self._gen_static_annotations_cache_key: # type: ignore # noqa: E501 result += self._annotations_cache_key # type: ignore # noqa: E501 else: result += self._gen_annotations_cache_key(anon_map) # type: ignore # noqa: E501 elif ( meth is InternalTraversal.dp_clauseelement_list or meth is InternalTraversal.dp_clauseelement_tuple or meth is InternalTraversal.dp_memoized_select_entities ): result += ( attrname, tuple( [ elem._gen_cache_key(anon_map, bindparams) for elem in obj ] ), ) else: result += meth( # type: ignore attrname, obj, self, anon_map, bindparams ) return result def _generate_cache_key(self) -> Optional[CacheKey]: """return a cache key. The cache key is a tuple which can contain any series of objects that are hashable and also identifies this object uniquely within the presence of a larger SQL expression or statement, for the purposes of caching the resulting query. The cache key should be based on the SQL compiled structure that would ultimately be produced. That is, two structures that are composed in exactly the same way should produce the same cache key; any difference in the structures that would affect the SQL string or the type handlers should result in a different cache key. The cache key returned by this method is an instance of :class:`.CacheKey`, which consists of a tuple representing the cache key, as well as a list of :class:`.BindParameter` objects which are extracted from the expression. While two expressions that produce identical cache key tuples will themselves generate identical SQL strings, the list of :class:`.BindParameter` objects indicates the bound values which may have different values in each one; these bound parameters must be consulted in order to execute the statement with the correct parameters. a :class:`_expression.ClauseElement` structure that does not implement a :meth:`._gen_cache_key` method and does not implement a :attr:`.traverse_internals` attribute will not be cacheable; when such an element is embedded into a larger structure, this method will return None, indicating no cache key is available. """ bindparams: List[BindParameter[Any]] = [] _anon_map = anon_map() key = self._gen_cache_key(_anon_map, bindparams) if NO_CACHE in _anon_map: return None else: assert key is not None return CacheKey(key, bindparams) @classmethod def _generate_cache_key_for_object( cls, obj: HasCacheKey ) -> Optional[CacheKey]: bindparams: List[BindParameter[Any]] = [] _anon_map = anon_map() key = obj._gen_cache_key(_anon_map, bindparams) if NO_CACHE in _anon_map: return None else: assert key is not None return CacheKey(key, bindparams) class HasCacheKeyTraverse(HasTraverseInternals, HasCacheKey): pass class MemoizedHasCacheKey(HasCacheKey, HasMemoized): __slots__ = () @HasMemoized.memoized_instancemethod def _generate_cache_key(self) -> Optional[CacheKey]: return HasCacheKey._generate_cache_key(self) class SlotsMemoizedHasCacheKey(HasCacheKey, util.MemoizedSlots): __slots__ = () def _memoized_method__generate_cache_key(self) -> Optional[CacheKey]: return HasCacheKey._generate_cache_key(self) class CacheKey(NamedTuple): """The key used to identify a SQL statement construct in the SQL compilation cache. .. seealso:: :ref:`sql_caching` """ key: Tuple[Any, ...] bindparams: Sequence[BindParameter[Any]] # can't set __hash__ attribute because it interferes # with namedtuple # can't use "if not TYPE_CHECKING" because mypy rejects it # inside of a NamedTuple def __hash__(self) -> Optional[int]: # type: ignore """CacheKey itself is not hashable - hash the .key portion""" return None def to_offline_string( self, statement_cache: MutableMapping[Any, str], statement: ClauseElement, parameters: _CoreSingleExecuteParams, ) -> str: """Generate an "offline string" form of this :class:`.CacheKey` The "offline string" is basically the string SQL for the statement plus a repr of the bound parameter values in series. Whereas the :class:`.CacheKey` object is dependent on in-memory identities in order to work as a cache key, the "offline" version is suitable for a cache that will work for other processes as well. The given ``statement_cache`` is a dictionary-like object where the string form of the statement itself will be cached. This dictionary should be in a longer lived scope in order to reduce the time spent stringifying statements. """ if self.key not in statement_cache: statement_cache[self.key] = sql_str = str(statement) else: sql_str = statement_cache[self.key] if not self.bindparams: param_tuple = tuple(parameters[key] for key in sorted(parameters)) else: param_tuple = tuple( parameters.get(bindparam.key, bindparam.value) for bindparam in self.bindparams ) return repr((sql_str, param_tuple)) def __eq__(self, other: Any) -> bool: return bool(self.key == other.key) def __ne__(self, other: Any) -> bool: return not (self.key == other.key) @classmethod def _diff_tuples(cls, left: CacheKey, right: CacheKey) -> str: ck1 = CacheKey(left, []) ck2 = CacheKey(right, []) return ck1._diff(ck2) def _whats_different(self, other: CacheKey) -> Iterator[str]: k1 = self.key k2 = other.key stack: List[int] = [] pickup_index = 0 while True: s1, s2 = k1, k2 for idx in stack: s1 = s1[idx] s2 = s2[idx] for idx, (e1, e2) in enumerate(zip_longest(s1, s2)): if idx < pickup_index: continue if e1 != e2: if isinstance(e1, tuple) and isinstance(e2, tuple): stack.append(idx) break else: yield "key%s[%d]: %s != %s" % ( "".join("[%d]" % id_ for id_ in stack), idx, e1, e2, ) else: pickup_index = stack.pop(-1) break def _diff(self, other: CacheKey) -> str: return ", ".join(self._whats_different(other)) def __str__(self) -> str: stack: List[Union[Tuple[Any, ...], HasCacheKey]] = [self.key] output = [] sentinel = object() indent = -1 while stack: elem = stack.pop(0) if elem is sentinel: output.append((" " * (indent * 2)) + "),") indent -= 1 elif isinstance(elem, tuple): if not elem: output.append((" " * ((indent + 1) * 2)) + "()") else: indent += 1 stack = list(elem) + [sentinel] + stack output.append((" " * (indent * 2)) + "(") else: if isinstance(elem, HasCacheKey): repr_ = "<%s object at %s>" % ( type(elem).__name__, hex(id(elem)), ) else: repr_ = repr(elem) output.append((" " * (indent * 2)) + " " + repr_ + ", ") return "CacheKey(key=%s)" % ("\n".join(output),) def _generate_param_dict(self) -> Dict[str, Any]: """used for testing""" _anon_map = prefix_anon_map() return {b.key % _anon_map: b.effective_value for b in self.bindparams} @util.preload_module("sqlalchemy.sql.elements") def _apply_params_to_element( self, original_cache_key: CacheKey, target_element: ColumnElement[Any] ) -> ColumnElement[Any]: if target_element._is_immutable or original_cache_key is self: return target_element elements = util.preloaded.sql_elements return elements._OverrideBinds( target_element, self.bindparams, original_cache_key.bindparams ) def _ad_hoc_cache_key_from_args( tokens: Tuple[Any, ...], traverse_args: Iterable[Tuple[str, InternalTraversal]], args: Iterable[Any], ) -> Tuple[Any, ...]: """a quick cache key generator used by reflection.flexi_cache.""" bindparams: List[BindParameter[Any]] = [] _anon_map = anon_map() tup = tokens for (attrname, sym), arg in zip(traverse_args, args): key = sym.name visit_key = key.replace("dp_", "visit_") if arg is None: tup += (attrname, None) continue meth = getattr(_cache_key_traversal_visitor, visit_key) if meth is CACHE_IN_PLACE: tup += (attrname, arg) elif meth in ( CALL_GEN_CACHE_KEY, STATIC_CACHE_KEY, ANON_NAME, PROPAGATE_ATTRS, ): raise NotImplementedError( f"Haven't implemented symbol {meth} for ad-hoc key from args" ) else: tup += meth(attrname, arg, None, _anon_map, bindparams) return tup class _CacheKeyTraversal(HasTraversalDispatch): # very common elements are inlined into the main _get_cache_key() method # to produce a dramatic savings in Python function call overhead visit_has_cache_key = visit_clauseelement = CALL_GEN_CACHE_KEY visit_clauseelement_list = InternalTraversal.dp_clauseelement_list visit_annotations_key = InternalTraversal.dp_annotations_key visit_clauseelement_tuple = InternalTraversal.dp_clauseelement_tuple visit_memoized_select_entities = ( InternalTraversal.dp_memoized_select_entities ) visit_string = visit_boolean = visit_operator = visit_plain_obj = ( CACHE_IN_PLACE ) visit_statement_hint_list = CACHE_IN_PLACE visit_type = STATIC_CACHE_KEY visit_anon_name = ANON_NAME visit_propagate_attrs = PROPAGATE_ATTRS def visit_with_context_options( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return tuple((fn.__code__, c_key) for fn, c_key in obj) def visit_inspectable( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return (attrname, inspect(obj)._gen_cache_key(anon_map, bindparams)) def visit_string_list( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return tuple(obj) def visit_multi( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return ( attrname, ( obj._gen_cache_key(anon_map, bindparams) if isinstance(obj, HasCacheKey) else obj ), ) def visit_multi_list( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return ( attrname, tuple( ( elem._gen_cache_key(anon_map, bindparams) if isinstance(elem, HasCacheKey) else elem ) for elem in obj ), ) def visit_has_cache_key_tuples( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: if not obj: return () return ( attrname, tuple( tuple( elem._gen_cache_key(anon_map, bindparams) for elem in tup_elem ) for tup_elem in obj ), ) def visit_has_cache_key_list( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: if not obj: return () return ( attrname, tuple(elem._gen_cache_key(anon_map, bindparams) for elem in obj), ) def visit_executable_options( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: if not obj: return () return ( attrname, tuple( elem._gen_cache_key(anon_map, bindparams) for elem in obj if elem._is_has_cache_key ), ) def visit_inspectable_list( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return self.visit_has_cache_key_list( attrname, [inspect(o) for o in obj], parent, anon_map, bindparams ) def visit_clauseelement_tuples( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return self.visit_has_cache_key_tuples( attrname, obj, parent, anon_map, bindparams ) def visit_fromclause_ordered_set( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: if not obj: return () return ( attrname, tuple([elem._gen_cache_key(anon_map, bindparams) for elem in obj]), ) def visit_clauseelement_unordered_set( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: if not obj: return () cache_keys = [ elem._gen_cache_key(anon_map, bindparams) for elem in obj ] return ( attrname, tuple( sorted(cache_keys) ), # cache keys all start with (id_, class) ) def visit_named_ddl_element( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return (attrname, obj.name) def visit_prefix_sequence( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: if not obj: return () return ( attrname, tuple( [ (clause._gen_cache_key(anon_map, bindparams), strval) for clause, strval in obj ] ), ) def visit_setup_join_tuple( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return tuple( ( target._gen_cache_key(anon_map, bindparams), ( onclause._gen_cache_key(anon_map, bindparams) if onclause is not None else None ), ( from_._gen_cache_key(anon_map, bindparams) if from_ is not None else None ), tuple([(key, flags[key]) for key in sorted(flags)]), ) for (target, onclause, from_, flags) in obj ) def visit_table_hint_list( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: if not obj: return () return ( attrname, tuple( [ ( clause._gen_cache_key(anon_map, bindparams), dialect_name, text, ) for (clause, dialect_name), text in obj.items() ] ), ) def visit_plain_dict( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return (attrname, tuple([(key, obj[key]) for key in sorted(obj)])) def visit_dialect_options( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return ( attrname, tuple( ( dialect_name, tuple( [ (key, obj[dialect_name][key]) for key in sorted(obj[dialect_name]) ] ), ) for dialect_name in sorted(obj) ), ) def visit_string_clauseelement_dict( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return ( attrname, tuple( (key, obj[key]._gen_cache_key(anon_map, bindparams)) for key in sorted(obj) ), ) def visit_string_multi_dict( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return ( attrname, tuple( ( key, ( value._gen_cache_key(anon_map, bindparams) if isinstance(value, HasCacheKey) else value ), ) for key, value in [(key, obj[key]) for key in sorted(obj)] ), ) def visit_fromclause_canonical_column_collection( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: # inlining into the internals of ColumnCollection return ( attrname, tuple( col._gen_cache_key(anon_map, bindparams) for k, col, _ in obj._collection ), ) def visit_unknown_structure( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: anon_map[NO_CACHE] = True return () def visit_dml_ordered_values( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: return ( attrname, tuple( ( ( key._gen_cache_key(anon_map, bindparams) if hasattr(key, "__clause_element__") else key ), value._gen_cache_key(anon_map, bindparams), ) for key, value in obj ), ) def visit_dml_values( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: # in py37 we can assume two dictionaries created in the same # insert ordering will retain that sorting return ( attrname, tuple( ( ( k._gen_cache_key(anon_map, bindparams) if hasattr(k, "__clause_element__") else k ), obj[k]._gen_cache_key(anon_map, bindparams), ) for k in obj ), ) def visit_dml_multi_values( self, attrname: str, obj: Any, parent: Any, anon_map: anon_map, bindparams: List[BindParameter[Any]], ) -> Tuple[Any, ...]: # multivalues are simply not cacheable right now anon_map[NO_CACHE] = True return () _cache_key_traversal_visitor = _CacheKeyTraversal()
Cancel