← Back
Editing: lambdas.py
# sql/lambdas.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 # mypy: allow-untyped-defs, allow-untyped-calls from __future__ import annotations import collections.abc as collections_abc import inspect import itertools import operator import threading import types from types import CodeType from typing import Any from typing import Callable from typing import cast from typing import List from typing import MutableMapping from typing import Optional from typing import Tuple from typing import Type from typing import TYPE_CHECKING from typing import TypeVar from typing import Union import weakref from . import cache_key as _cache_key from . import coercions from . import elements from . import roles from . import schema from . import visitors from .base import _clone from .base import Executable from .base import Options from .cache_key import CacheConst from .operators import ColumnOperators from .. import exc from .. import inspection from .. import util from ..util.typing import Literal if TYPE_CHECKING: from .elements import BindParameter from .elements import ClauseElement from .roles import SQLRole from .visitors import _CloneCallableType _LambdaCacheType = MutableMapping[ Tuple[Any, ...], Union["NonAnalyzedFunction", "AnalyzedFunction"] ] _BoundParameterGetter = Callable[..., Any] _closure_per_cache_key: _LambdaCacheType = util.LRUCache(1000) _LambdaType = Callable[[], Any] _AnyLambdaType = Callable[..., Any] _StmtLambdaType = Callable[[], Any] _E = TypeVar("_E", bound=Executable) _StmtLambdaElementType = Callable[[_E], Any] class LambdaOptions(Options): enable_tracking = True track_closure_variables = True track_on: Optional[object] = None global_track_bound_values = True track_bound_values = True lambda_cache: Optional[_LambdaCacheType] = None def lambda_stmt( lmb: _StmtLambdaType, enable_tracking: bool = True, track_closure_variables: bool = True, track_on: Optional[object] = None, global_track_bound_values: bool = True, track_bound_values: bool = True, lambda_cache: Optional[_LambdaCacheType] = None, ) -> StatementLambdaElement: """Produce a SQL statement that is cached as a lambda. The Python code object within the lambda is scanned for both Python literals that will become bound parameters as well as closure variables that refer to Core or ORM constructs that may vary. The lambda itself will be invoked only once per particular set of constructs detected. E.g.:: from sqlalchemy import lambda_stmt stmt = lambda_stmt(lambda: table.select()) stmt += lambda s: s.where(table.c.id == 5) result = connection.execute(stmt) The object returned is an instance of :class:`_sql.StatementLambdaElement`. .. versionadded:: 1.4 :param lmb: a Python function, typically a lambda, which takes no arguments and returns a SQL expression construct :param enable_tracking: when False, all scanning of the given lambda for changes in closure variables or bound parameters is disabled. Use for a lambda that produces the identical results in all cases with no parameterization. :param track_closure_variables: when False, changes in closure variables within the lambda will not be scanned. Use for a lambda where the state of its closure variables will never change the SQL structure returned by the lambda. :param track_bound_values: when False, bound parameter tracking will be disabled for the given lambda. Use for a lambda that either does not produce any bound values, or where the initial bound values never change. :param global_track_bound_values: when False, bound parameter tracking will be disabled for the entire statement including additional links added via the :meth:`_sql.StatementLambdaElement.add_criteria` method. :param lambda_cache: a dictionary or other mapping-like object where information about the lambda's Python code as well as the tracked closure variables in the lambda itself will be stored. Defaults to a global LRU cache. This cache is independent of the "compiled_cache" used by the :class:`_engine.Connection` object. .. seealso:: :ref:`engine_lambda_caching` """ return StatementLambdaElement( lmb, roles.StatementRole, LambdaOptions( enable_tracking=enable_tracking, track_on=track_on, track_closure_variables=track_closure_variables, global_track_bound_values=global_track_bound_values, track_bound_values=track_bound_values, lambda_cache=lambda_cache, ), ) class LambdaElement(elements.ClauseElement): """A SQL construct where the state is stored as an un-invoked lambda. The :class:`_sql.LambdaElement` is produced transparently whenever passing lambda expressions into SQL constructs, such as:: stmt = select(table).where(lambda: table.c.col == parameter) The :class:`_sql.LambdaElement` is the base of the :class:`_sql.StatementLambdaElement` which represents a full statement within a lambda. .. versionadded:: 1.4 .. seealso:: :ref:`engine_lambda_caching` """ __visit_name__ = "lambda_element" _is_lambda_element = True _traverse_internals = [ ("_resolved", visitors.InternalTraversal.dp_clauseelement) ] _transforms: Tuple[_CloneCallableType, ...] = () _resolved_bindparams: List[BindParameter[Any]] parent_lambda: Optional[StatementLambdaElement] = None closure_cache_key: Union[Tuple[Any, ...], Literal[CacheConst.NO_CACHE]] role: Type[SQLRole] _rec: Union[AnalyzedFunction, NonAnalyzedFunction] fn: _AnyLambdaType tracker_key: Tuple[CodeType, ...] def __repr__(self): return "%s(%r)" % ( self.__class__.__name__, self.fn.__code__, ) def __init__( self, fn: _LambdaType, role: Type[SQLRole], opts: Union[Type[LambdaOptions], LambdaOptions] = LambdaOptions, apply_propagate_attrs: Optional[ClauseElement] = None, ): self.fn = fn self.role = role self.tracker_key = (fn.__code__,) self.opts = opts if apply_propagate_attrs is None and (role is roles.StatementRole): apply_propagate_attrs = self rec = self._retrieve_tracker_rec(fn, apply_propagate_attrs, opts) if apply_propagate_attrs is not None: propagate_attrs = rec.propagate_attrs if propagate_attrs: apply_propagate_attrs._propagate_attrs = propagate_attrs def _retrieve_tracker_rec(self, fn, apply_propagate_attrs, opts): lambda_cache = opts.lambda_cache if lambda_cache is None: lambda_cache = _closure_per_cache_key tracker_key = self.tracker_key fn = self.fn closure = fn.__closure__ tracker = AnalyzedCode.get( fn, self, opts, ) bindparams: List[BindParameter[Any]] self._resolved_bindparams = bindparams = [] if self.parent_lambda is not None: parent_closure_cache_key = self.parent_lambda.closure_cache_key else: parent_closure_cache_key = () cache_key: Union[Tuple[Any, ...], Literal[CacheConst.NO_CACHE]] if parent_closure_cache_key is not _cache_key.NO_CACHE: anon_map = visitors.anon_map() cache_key = tuple( [ getter(closure, opts, anon_map, bindparams) for getter in tracker.closure_trackers ] ) if _cache_key.NO_CACHE not in anon_map: cache_key = parent_closure_cache_key + cache_key self.closure_cache_key = cache_key try: rec = lambda_cache[tracker_key + cache_key] except KeyError: rec = None else: cache_key = _cache_key.NO_CACHE rec = None else: cache_key = _cache_key.NO_CACHE rec = None self.closure_cache_key = cache_key if rec is None: if cache_key is not _cache_key.NO_CACHE: with AnalyzedCode._generation_mutex: key = tracker_key + cache_key if key not in lambda_cache: rec = AnalyzedFunction( tracker, self, apply_propagate_attrs, fn ) rec.closure_bindparams = list(bindparams) lambda_cache[key] = rec else: rec = lambda_cache[key] else: rec = NonAnalyzedFunction(self._invoke_user_fn(fn)) else: bindparams[:] = [ orig_bind._with_value(new_bind.value, maintain_key=True) for orig_bind, new_bind in zip( rec.closure_bindparams, bindparams ) ] self._rec = rec if cache_key is not _cache_key.NO_CACHE: if self.parent_lambda is not None: bindparams[:0] = self.parent_lambda._resolved_bindparams lambda_element: Optional[LambdaElement] = self while lambda_element is not None: rec = lambda_element._rec if rec.bindparam_trackers: tracker_instrumented_fn = rec.tracker_instrumented_fn for tracker in rec.bindparam_trackers: tracker( lambda_element.fn, tracker_instrumented_fn, bindparams, ) lambda_element = lambda_element.parent_lambda return rec def __getattr__(self, key): return getattr(self._rec.expected_expr, key) @property def _is_sequence(self): return self._rec.is_sequence @property def _select_iterable(self): if self._is_sequence: return itertools.chain.from_iterable( [element._select_iterable for element in self._resolved] ) else: return self._resolved._select_iterable @property def _from_objects(self): if self._is_sequence: return itertools.chain.from_iterable( [element._from_objects for element in self._resolved] ) else: return self._resolved._from_objects def _param_dict(self): return {b.key: b.value for b in self._resolved_bindparams} def _setup_binds_for_tracked_expr(self, expr): bindparam_lookup = {b.key: b for b in self._resolved_bindparams} def replace( element: Optional[visitors.ExternallyTraversible], **kw: Any ) -> Optional[visitors.ExternallyTraversible]: if isinstance(element, elements.BindParameter): if element.key in bindparam_lookup: bind = bindparam_lookup[element.key] if element.expanding: bind.expanding = True bind.expand_op = element.expand_op bind.type = element.type return bind return None if self._rec.is_sequence: expr = [ visitors.replacement_traverse(sub_expr, {}, replace) for sub_expr in expr ] elif getattr(expr, "is_clause_element", False): expr = visitors.replacement_traverse(expr, {}, replace) return expr def _copy_internals( self, clone: _CloneCallableType = _clone, deferred_copy_internals: Optional[_CloneCallableType] = None, **kw: Any, ) -> None: # TODO: this needs A LOT of tests self._resolved = clone( self._resolved, deferred_copy_internals=deferred_copy_internals, **kw, ) @util.memoized_property def _resolved(self): expr = self._rec.expected_expr if self._resolved_bindparams: expr = self._setup_binds_for_tracked_expr(expr) return expr def _gen_cache_key(self, anon_map, bindparams): if self.closure_cache_key is _cache_key.NO_CACHE: anon_map[_cache_key.NO_CACHE] = True return None cache_key = ( self.fn.__code__, self.__class__, ) + self.closure_cache_key parent = self.parent_lambda while parent is not None: assert parent.closure_cache_key is not CacheConst.NO_CACHE parent_closure_cache_key: Tuple[Any, ...] = ( parent.closure_cache_key ) cache_key = ( (parent.fn.__code__,) + parent_closure_cache_key + cache_key ) parent = parent.parent_lambda if self._resolved_bindparams: bindparams.extend(self._resolved_bindparams) return cache_key def _invoke_user_fn(self, fn: _AnyLambdaType, *arg: Any) -> ClauseElement: return fn() # type: ignore[no-any-return] class DeferredLambdaElement(LambdaElement): """A LambdaElement where the lambda accepts arguments and is invoked within the compile phase with special context. This lambda doesn't normally produce its real SQL expression outside of the compile phase. It is passed a fixed set of initial arguments so that it can generate a sample expression. """ def __init__( self, fn: _AnyLambdaType, role: Type[roles.SQLRole], opts: Union[Type[LambdaOptions], LambdaOptions] = LambdaOptions, lambda_args: Tuple[Any, ...] = (), ): self.lambda_args = lambda_args super().__init__(fn, role, opts) def _invoke_user_fn(self, fn, *arg): return fn(*self.lambda_args) def _resolve_with_args(self, *lambda_args: Any) -> ClauseElement: assert isinstance(self._rec, AnalyzedFunction) tracker_fn = self._rec.tracker_instrumented_fn expr = tracker_fn(*lambda_args) expr = coercions.expect(self.role, expr) expr = self._setup_binds_for_tracked_expr(expr) # this validation is getting very close, but not quite, to achieving # #5767. The problem is if the base lambda uses an unnamed column # as is very common with mixins, the parameter name is different # and it produces a false positive; that is, for the documented case # that is exactly what people will be doing, it doesn't work, so # I'm not really sure how to handle this right now. # expected_binds = [ # b._orig_key # for b in self._rec.expr._generate_cache_key()[1] # if b.required # ] # got_binds = [ # b._orig_key for b in expr._generate_cache_key()[1] if b.required # ] # if expected_binds != got_binds: # raise exc.InvalidRequestError( # "Lambda callable at %s produced a different set of bound " # "parameters than its original run: %s" # % (self.fn.__code__, ", ".join(got_binds)) # ) # TODO: TEST TEST TEST, this is very out there for deferred_copy_internals in self._transforms: expr = deferred_copy_internals(expr) return expr # type: ignore def _copy_internals( self, clone=_clone, deferred_copy_internals=None, **kw ): super()._copy_internals( clone=clone, deferred_copy_internals=deferred_copy_internals, # **kw opts=kw, ) # TODO: A LOT A LOT of tests. for _resolve_with_args, we don't know # our expression yet. so hold onto the replacement if deferred_copy_internals: self._transforms += (deferred_copy_internals,) class StatementLambdaElement( roles.AllowsLambdaRole, LambdaElement, Executable ): """Represent a composable SQL statement as a :class:`_sql.LambdaElement`. The :class:`_sql.StatementLambdaElement` is constructed using the :func:`_sql.lambda_stmt` function:: from sqlalchemy import lambda_stmt stmt = lambda_stmt(lambda: select(table)) Once constructed, additional criteria can be built onto the statement by adding subsequent lambdas, which accept the existing statement object as a single parameter:: stmt += lambda s: s.where(table.c.col == parameter) .. versionadded:: 1.4 .. seealso:: :ref:`engine_lambda_caching` """ if TYPE_CHECKING: def __init__( self, fn: _StmtLambdaType, role: Type[SQLRole], opts: Union[Type[LambdaOptions], LambdaOptions] = LambdaOptions, apply_propagate_attrs: Optional[ClauseElement] = None, ): ... def __add__( self, other: _StmtLambdaElementType[Any] ) -> StatementLambdaElement: return self.add_criteria(other) def add_criteria( self, other: _StmtLambdaElementType[Any], enable_tracking: bool = True, track_on: Optional[Any] = None, track_closure_variables: bool = True, track_bound_values: bool = True, ) -> StatementLambdaElement: """Add new criteria to this :class:`_sql.StatementLambdaElement`. E.g.:: >>> def my_stmt(parameter): ... stmt = lambda_stmt( ... lambda: select(table.c.x, table.c.y), ... ) ... stmt = stmt.add_criteria(lambda: table.c.x > parameter) ... return stmt The :meth:`_sql.StatementLambdaElement.add_criteria` method is equivalent to using the Python addition operator to add a new lambda, except that additional arguments may be added including ``track_closure_values`` and ``track_on``:: >>> def my_stmt(self, foo): ... stmt = lambda_stmt( ... lambda: select(func.max(foo.x, foo.y)), ... track_closure_variables=False, ... ) ... stmt = stmt.add_criteria(lambda: self.where_criteria, track_on=[self]) ... return stmt See :func:`_sql.lambda_stmt` for a description of the parameters accepted. """ # noqa: E501 opts = self.opts + dict( enable_tracking=enable_tracking, track_closure_variables=track_closure_variables, global_track_bound_values=self.opts.global_track_bound_values, track_on=track_on, track_bound_values=track_bound_values, ) return LinkedLambdaElement(other, parent_lambda=self, opts=opts) def _execute_on_connection( self, connection, distilled_params, execution_options ): if TYPE_CHECKING: assert isinstance(self._rec.expected_expr, ClauseElement) if self._rec.expected_expr.supports_execution: return connection._execute_clauseelement( self, distilled_params, execution_options ) else: raise exc.ObjectNotExecutableError(self) @property def _proxied(self) -> Any: return self._rec_expected_expr @property def _with_options(self): return self._proxied._with_options @property def _effective_plugin_target(self): return self._proxied._effective_plugin_target @property def _execution_options(self): return self._proxied._execution_options @property def _all_selected_columns(self): return self._proxied._all_selected_columns @property def is_select(self): return self._proxied.is_select @property def is_update(self): return self._proxied.is_update @property def is_insert(self): return self._proxied.is_insert @property def is_text(self): return self._proxied.is_text @property def is_delete(self): return self._proxied.is_delete @property def is_dml(self): return self._proxied.is_dml def spoil(self) -> NullLambdaStatement: """Return a new :class:`.StatementLambdaElement` that will run all lambdas unconditionally each time. """ return NullLambdaStatement(self.fn()) class NullLambdaStatement(roles.AllowsLambdaRole, elements.ClauseElement): """Provides the :class:`.StatementLambdaElement` API but does not cache or analyze lambdas. the lambdas are instead invoked immediately. The intended use is to isolate issues that may arise when using lambda statements. """ __visit_name__ = "lambda_element" _is_lambda_element = True _traverse_internals = [ ("_resolved", visitors.InternalTraversal.dp_clauseelement) ] def __init__(self, statement): self._resolved = statement self._propagate_attrs = statement._propagate_attrs def __getattr__(self, key): return getattr(self._resolved, key) def __add__(self, other): statement = other(self._resolved) return NullLambdaStatement(statement) def add_criteria(self, other, **kw): statement = other(self._resolved) return NullLambdaStatement(statement) def _execute_on_connection( self, connection, distilled_params, execution_options ): if self._resolved.supports_execution: return connection._execute_clauseelement( self, distilled_params, execution_options ) else: raise exc.ObjectNotExecutableError(self) class LinkedLambdaElement(StatementLambdaElement): """Represent subsequent links of a :class:`.StatementLambdaElement`.""" parent_lambda: StatementLambdaElement def __init__( self, fn: _StmtLambdaElementType[Any], parent_lambda: StatementLambdaElement, opts: Union[Type[LambdaOptions], LambdaOptions], ): self.opts = opts self.fn = fn self.parent_lambda = parent_lambda self.tracker_key = parent_lambda.tracker_key + (fn.__code__,) self._retrieve_tracker_rec(fn, self, opts) self._propagate_attrs = parent_lambda._propagate_attrs def _invoke_user_fn(self, fn, *arg): return fn(self.parent_lambda._resolved) class AnalyzedCode: __slots__ = ( "track_closure_variables", "track_bound_values", "bindparam_trackers", "closure_trackers", "build_py_wrappers", ) _fns: weakref.WeakKeyDictionary[CodeType, AnalyzedCode] = ( weakref.WeakKeyDictionary() ) _generation_mutex = threading.RLock() @classmethod def get(cls, fn, lambda_element, lambda_kw, **kw): try: # TODO: validate kw haven't changed? return cls._fns[fn.__code__] except KeyError: pass with cls._generation_mutex: # check for other thread already created object if fn.__code__ in cls._fns: return cls._fns[fn.__code__] analyzed: AnalyzedCode cls._fns[fn.__code__] = analyzed = AnalyzedCode( fn, lambda_element, lambda_kw, **kw ) return analyzed def __init__(self, fn, lambda_element, opts): if inspect.ismethod(fn): raise exc.ArgumentError( "Method %s may not be passed as a SQL expression" % fn ) closure = fn.__closure__ self.track_bound_values = ( opts.track_bound_values and opts.global_track_bound_values ) enable_tracking = opts.enable_tracking track_on = opts.track_on track_closure_variables = opts.track_closure_variables self.track_closure_variables = track_closure_variables and not track_on # a list of callables generated from _bound_parameter_getter_* # functions. Each of these uses a PyWrapper object to retrieve # a parameter value self.bindparam_trackers = [] # a list of callables generated from _cache_key_getter_* functions # these callables work to generate a cache key for the lambda # based on what's inside its closure variables. self.closure_trackers = [] self.build_py_wrappers = [] if enable_tracking: if track_on: self._init_track_on(track_on) self._init_globals(fn) if closure: self._init_closure(fn) self._setup_additional_closure_trackers(fn, lambda_element, opts) def _init_track_on(self, track_on): self.closure_trackers.extend( self._cache_key_getter_track_on(idx, elem) for idx, elem in enumerate(track_on) ) def _init_globals(self, fn): build_py_wrappers = self.build_py_wrappers bindparam_trackers = self.bindparam_trackers track_bound_values = self.track_bound_values for name in fn.__code__.co_names: if name not in fn.__globals__: continue _bound_value = self._roll_down_to_literal(fn.__globals__[name]) if coercions._deep_is_literal(_bound_value): build_py_wrappers.append((name, None)) if track_bound_values: bindparam_trackers.append( self._bound_parameter_getter_func_globals(name) ) def _init_closure(self, fn): build_py_wrappers = self.build_py_wrappers closure = fn.__closure__ track_bound_values = self.track_bound_values track_closure_variables = self.track_closure_variables bindparam_trackers = self.bindparam_trackers closure_trackers = self.closure_trackers for closure_index, (fv, cell) in enumerate( zip(fn.__code__.co_freevars, closure) ): _bound_value = self._roll_down_to_literal(cell.cell_contents) if coercions._deep_is_literal(_bound_value): build_py_wrappers.append((fv, closure_index)) if track_bound_values: bindparam_trackers.append( self._bound_parameter_getter_func_closure( fv, closure_index ) ) else: # for normal cell contents, add them to a list that # we can compare later when we get new lambdas. if # any identities have changed, then we will # recalculate the whole lambda and run it again. if track_closure_variables: closure_trackers.append( self._cache_key_getter_closure_variable( fn, fv, closure_index, cell.cell_contents ) ) def _setup_additional_closure_trackers(self, fn, lambda_element, opts): # an additional step is to actually run the function, then # go through the PyWrapper objects that were set up to catch a bound # parameter. then if they *didn't* make a param, oh they're another # object in the closure we have to track for our cache key. so # create trackers to catch those. analyzed_function = AnalyzedFunction( self, lambda_element, None, fn, ) closure_trackers = self.closure_trackers for pywrapper in analyzed_function.closure_pywrappers: if not pywrapper._sa__has_param: closure_trackers.append( self._cache_key_getter_tracked_literal(fn, pywrapper) ) @classmethod def _roll_down_to_literal(cls, element): is_clause_element = hasattr(element, "__clause_element__") if is_clause_element: while not isinstance( element, (elements.ClauseElement, schema.SchemaItem, type) ): try: element = element.__clause_element__() except AttributeError: break if not is_clause_element: insp = inspection.inspect(element, raiseerr=False) if insp is not None: try: return insp.__clause_element__() except AttributeError: return insp # TODO: should we coerce consts None/True/False here? return element else: return element def _bound_parameter_getter_func_globals(self, name): """Return a getter that will extend a list of bound parameters with new entries from the ``__globals__`` collection of a particular lambda. """ def extract_parameter_value( current_fn, tracker_instrumented_fn, result ): wrapper = tracker_instrumented_fn.__globals__[name] object.__getattribute__(wrapper, "_extract_bound_parameters")( current_fn.__globals__[name], result ) return extract_parameter_value def _bound_parameter_getter_func_closure(self, name, closure_index): """Return a getter that will extend a list of bound parameters with new entries from the ``__closure__`` collection of a particular lambda. """ def extract_parameter_value( current_fn, tracker_instrumented_fn, result ): wrapper = tracker_instrumented_fn.__closure__[ closure_index ].cell_contents object.__getattribute__(wrapper, "_extract_bound_parameters")( current_fn.__closure__[closure_index].cell_contents, result ) return extract_parameter_value def _cache_key_getter_track_on(self, idx, elem): """Return a getter that will extend a cache key with new entries from the "track_on" parameter passed to a :class:`.LambdaElement`. """ if isinstance(elem, tuple): # tuple must contain hascachekey elements def get(closure, opts, anon_map, bindparams): return tuple( tup_elem._gen_cache_key(anon_map, bindparams) for tup_elem in opts.track_on[idx] ) elif isinstance(elem, _cache_key.HasCacheKey): def get(closure, opts, anon_map, bindparams): return opts.track_on[idx]._gen_cache_key(anon_map, bindparams) else: def get(closure, opts, anon_map, bindparams): return opts.track_on[idx] return get def _cache_key_getter_closure_variable( self, fn, variable_name, idx, cell_contents, use_clause_element=False, use_inspect=False, ): """Return a getter that will extend a cache key with new entries from the ``__closure__`` collection of a particular lambda. """ if isinstance(cell_contents, _cache_key.HasCacheKey): def get(closure, opts, anon_map, bindparams): obj = closure[idx].cell_contents if use_inspect: obj = inspection.inspect(obj) elif use_clause_element: while hasattr(obj, "__clause_element__"): if not getattr(obj, "is_clause_element", False): obj = obj.__clause_element__() return obj._gen_cache_key(anon_map, bindparams) elif isinstance(cell_contents, types.FunctionType): def get(closure, opts, anon_map, bindparams): return closure[idx].cell_contents.__code__ elif isinstance(cell_contents, collections_abc.Sequence): def get(closure, opts, anon_map, bindparams): contents = closure[idx].cell_contents try: return tuple( elem._gen_cache_key(anon_map, bindparams) for elem in contents ) except AttributeError as ae: self._raise_for_uncacheable_closure_variable( variable_name, fn, from_=ae ) else: # if the object is a mapped class or aliased class, or some # other object in the ORM realm of things like that, imitate # the logic used in coercions.expect() to roll it down to the # SQL element element = cell_contents is_clause_element = False while hasattr(element, "__clause_element__"): is_clause_element = True if not getattr(element, "is_clause_element", False): element = element.__clause_element__() else: break if not is_clause_element: insp = inspection.inspect(element, raiseerr=False) if insp is not None: return self._cache_key_getter_closure_variable( fn, variable_name, idx, insp, use_inspect=True ) else: return self._cache_key_getter_closure_variable( fn, variable_name, idx, element, use_clause_element=True ) self._raise_for_uncacheable_closure_variable(variable_name, fn) return get def _raise_for_uncacheable_closure_variable( self, variable_name, fn, from_=None ): raise exc.InvalidRequestError( "Closure variable named '%s' inside of lambda callable %s " "does not refer to a cacheable SQL element, and also does not " "appear to be serving as a SQL literal bound value based on " "the default " "SQL expression returned by the function. This variable " "needs to remain outside the scope of a SQL-generating lambda " "so that a proper cache key may be generated from the " "lambda's state. Evaluate this variable outside of the " "lambda, set track_on=[<elements>] to explicitly select " "closure elements to track, or set " "track_closure_variables=False to exclude " "closure variables from being part of the cache key." % (variable_name, fn.__code__), ) from from_ def _cache_key_getter_tracked_literal(self, fn, pytracker): """Return a getter that will extend a cache key with new entries from the ``__closure__`` collection of a particular lambda. this getter differs from _cache_key_getter_closure_variable in that these are detected after the function is run, and PyWrapper objects have recorded that a particular literal value is in fact not being interpreted as a bound parameter. """ elem = pytracker._sa__to_evaluate closure_index = pytracker._sa__closure_index variable_name = pytracker._sa__name return self._cache_key_getter_closure_variable( fn, variable_name, closure_index, elem ) class NonAnalyzedFunction: __slots__ = ("expr",) closure_bindparams: Optional[List[BindParameter[Any]]] = None bindparam_trackers: Optional[List[_BoundParameterGetter]] = None is_sequence = False expr: ClauseElement def __init__(self, expr: ClauseElement): self.expr = expr @property def expected_expr(self) -> ClauseElement: return self.expr class AnalyzedFunction: __slots__ = ( "analyzed_code", "fn", "closure_pywrappers", "tracker_instrumented_fn", "expr", "bindparam_trackers", "expected_expr", "is_sequence", "propagate_attrs", "closure_bindparams", ) closure_bindparams: Optional[List[BindParameter[Any]]] expected_expr: Union[ClauseElement, List[ClauseElement]] bindparam_trackers: Optional[List[_BoundParameterGetter]] def __init__( self, analyzed_code, lambda_element, apply_propagate_attrs, fn, ): self.analyzed_code = analyzed_code self.fn = fn self.bindparam_trackers = analyzed_code.bindparam_trackers self._instrument_and_run_function(lambda_element) self._coerce_expression(lambda_element, apply_propagate_attrs) def _instrument_and_run_function(self, lambda_element): analyzed_code = self.analyzed_code fn = self.fn self.closure_pywrappers = closure_pywrappers = [] build_py_wrappers = analyzed_code.build_py_wrappers if not build_py_wrappers: self.tracker_instrumented_fn = tracker_instrumented_fn = fn self.expr = lambda_element._invoke_user_fn(tracker_instrumented_fn) else: track_closure_variables = analyzed_code.track_closure_variables closure = fn.__closure__ # will form the __closure__ of the function when we rebuild it if closure: new_closure = { fv: cell.cell_contents for fv, cell in zip(fn.__code__.co_freevars, closure) } else: new_closure = {} # will form the __globals__ of the function when we rebuild it new_globals = fn.__globals__.copy() for name, closure_index in build_py_wrappers: if closure_index is not None: value = closure[closure_index].cell_contents new_closure[name] = bind = PyWrapper( fn, name, value, closure_index=closure_index, track_bound_values=( self.analyzed_code.track_bound_values ), ) if track_closure_variables: closure_pywrappers.append(bind) else: value = fn.__globals__[name] new_globals[name] = bind = PyWrapper(fn, name, value) # rewrite the original fn. things that look like they will # become bound parameters are wrapped in a PyWrapper. self.tracker_instrumented_fn = tracker_instrumented_fn = ( self._rewrite_code_obj( fn, [new_closure[name] for name in fn.__code__.co_freevars], new_globals, ) ) # now invoke the function. This will give us a new SQL # expression, but all the places that there would be a bound # parameter, the PyWrapper in its place will give us a bind # with a predictable name we can match up later. # additionally, each PyWrapper will log that it did in fact # create a parameter, otherwise, it's some kind of Python # object in the closure and we want to track that, to make # sure it doesn't change to something else, or if it does, # that we create a different tracked function with that # variable. self.expr = lambda_element._invoke_user_fn(tracker_instrumented_fn) def _coerce_expression(self, lambda_element, apply_propagate_attrs): """Run the tracker-generated expression through coercion rules. After the user-defined lambda has been invoked to produce a statement for re-use, run it through coercion rules to both check that it's the correct type of object and also to coerce it to its useful form. """ parent_lambda = lambda_element.parent_lambda expr = self.expr if parent_lambda is None: if isinstance(expr, collections_abc.Sequence): self.expected_expr = [ cast( "ClauseElement", coercions.expect( lambda_element.role, sub_expr, apply_propagate_attrs=apply_propagate_attrs, ), ) for sub_expr in expr ] self.is_sequence = True else: self.expected_expr = cast( "ClauseElement", coercions.expect( lambda_element.role, expr, apply_propagate_attrs=apply_propagate_attrs, ), ) self.is_sequence = False else: self.expected_expr = expr self.is_sequence = False if apply_propagate_attrs is not None: self.propagate_attrs = apply_propagate_attrs._propagate_attrs else: self.propagate_attrs = util.EMPTY_DICT def _rewrite_code_obj(self, f, cell_values, globals_): """Return a copy of f, with a new closure and new globals yes it works in pypy :P """ argrange = range(len(cell_values)) code = "def make_cells():\n" if cell_values: code += " (%s) = (%s)\n" % ( ", ".join("i%d" % i for i in argrange), ", ".join("o%d" % i for i in argrange), ) code += " def closure():\n" code += " return %s\n" % ", ".join("i%d" % i for i in argrange) code += " return closure.__closure__" vars_ = {"o%d" % i: cell_values[i] for i in argrange} exec(code, vars_, vars_) closure = vars_["make_cells"]() func = type(f)( f.__code__, globals_, f.__name__, f.__defaults__, closure ) func.__annotations__ = f.__annotations__ func.__kwdefaults__ = f.__kwdefaults__ func.__doc__ = f.__doc__ func.__module__ = f.__module__ return func class PyWrapper(ColumnOperators): """A wrapper object that is injected into the ``__globals__`` and ``__closure__`` of a Python function. When the function is instrumented with :class:`.PyWrapper` objects, it is then invoked just once in order to set up the wrappers. We look through all the :class:`.PyWrapper` objects we made to find the ones that generated a :class:`.BindParameter` object, e.g. the expression system interpreted something as a literal. Those positions in the globals/closure are then ones that we will look at, each time a new lambda comes in that refers to the same ``__code__`` object. In this way, we keep a single version of the SQL expression that this lambda produced, without calling upon the Python function that created it more than once, unless its other closure variables have changed. The expression is then transformed to have the new bound values embedded into it. """ def __init__( self, fn, name, to_evaluate, closure_index=None, getter=None, track_bound_values=True, ): self.fn = fn self._name = name self._to_evaluate = to_evaluate self._param = None self._has_param = False self._bind_paths = {} self._getter = getter self._closure_index = closure_index self.track_bound_values = track_bound_values def __call__(self, *arg, **kw): elem = object.__getattribute__(self, "_to_evaluate") value = elem(*arg, **kw) if ( self._sa_track_bound_values and coercions._deep_is_literal(value) and not isinstance( # TODO: coverage where an ORM option or similar is here value, _cache_key.HasCacheKey, ) ): name = object.__getattribute__(self, "_name") raise exc.InvalidRequestError( "Can't invoke Python callable %s() inside of lambda " "expression argument at %s; lambda SQL constructs should " "not invoke functions from closure variables to produce " "literal values since the " "lambda SQL system normally extracts bound values without " "actually " "invoking the lambda or any functions within it. Call the " "function outside of the " "lambda and assign to a local variable that is used in the " "lambda as a closure variable, or set " "track_bound_values=False if the return value of this " "function is used in some other way other than a SQL bound " "value." % (name, self._sa_fn.__code__) ) else: return value def operate(self, op, *other, **kwargs): elem = object.__getattribute__(self, "_py_wrapper_literal")() return op(elem, *other, **kwargs) def reverse_operate(self, op, other, **kwargs): elem = object.__getattribute__(self, "_py_wrapper_literal")() return op(other, elem, **kwargs) def _extract_bound_parameters(self, starting_point, result_list): param = object.__getattribute__(self, "_param") if param is not None: param = param._with_value(starting_point, maintain_key=True) result_list.append(param) for pywrapper in object.__getattribute__(self, "_bind_paths").values(): getter = object.__getattribute__(pywrapper, "_getter") element = getter(starting_point) pywrapper._sa__extract_bound_parameters(element, result_list) def _py_wrapper_literal(self, expr=None, operator=None, **kw): param = object.__getattribute__(self, "_param") to_evaluate = object.__getattribute__(self, "_to_evaluate") if param is None: name = object.__getattribute__(self, "_name") self._param = param = elements.BindParameter( name, required=False, unique=True, _compared_to_operator=operator, _compared_to_type=expr.type if expr is not None else None, ) self._has_param = True return param._with_value(to_evaluate, maintain_key=True) def __bool__(self): to_evaluate = object.__getattribute__(self, "_to_evaluate") return bool(to_evaluate) def __getattribute__(self, key): if key.startswith("_sa_"): return object.__getattribute__(self, key[4:]) elif key in ( "__clause_element__", "operate", "reverse_operate", "_py_wrapper_literal", "__class__", "__dict__", ): return object.__getattribute__(self, key) if key.startswith("__"): elem = object.__getattribute__(self, "_to_evaluate") return getattr(elem, key) else: return self._sa__add_getter(key, operator.attrgetter) def __iter__(self): elem = object.__getattribute__(self, "_to_evaluate") return iter(elem) def __getitem__(self, key): elem = object.__getattribute__(self, "_to_evaluate") if not hasattr(elem, "__getitem__"): raise AttributeError("__getitem__") if isinstance(key, PyWrapper): # TODO: coverage raise exc.InvalidRequestError( "Dictionary keys / list indexes inside of a cached " "lambda must be Python literals only" ) return self._sa__add_getter(key, operator.itemgetter) def _add_getter(self, key, getter_fn): bind_paths = object.__getattribute__(self, "_bind_paths") bind_path_key = (key, getter_fn) if bind_path_key in bind_paths: return bind_paths[bind_path_key] getter = getter_fn(key) elem = object.__getattribute__(self, "_to_evaluate") value = getter(elem) rolled_down_value = AnalyzedCode._roll_down_to_literal(value) if coercions._deep_is_literal(rolled_down_value): wrapper = PyWrapper(self._sa_fn, key, value, getter=getter) bind_paths[bind_path_key] = wrapper return wrapper else: return value @inspection._inspects(LambdaElement) def insp(lmb): return inspection.inspect(lmb._resolved)
Save File
Cancel