diff --git a/venv/Lib/site-packages/sqlalchemy/sql/util.py b/venv/Lib/site-packages/sqlalchemy/sql/util.py new file mode 100644 index 0000000..2b4b170 --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/sql/util.py @@ -0,0 +1,1486 @@ +# sql/util.py +# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors +# +# +# 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 + +"""High level utilities which build upon other modules here.""" + +from __future__ import annotations + +from collections import deque +import copy +from itertools import chain +import typing +from typing import AbstractSet +from typing import Any +from typing import Callable +from typing import cast +from typing import Collection +from typing import Dict +from typing import Iterable +from typing import Iterator +from typing import List +from typing import Optional +from typing import overload +from typing import Sequence +from typing import Tuple +from typing import TYPE_CHECKING +from typing import TypeVar +from typing import Union + +from . import coercions +from . import operators +from . import roles +from . import visitors +from ._typing import is_text_clause +from .annotation import _deep_annotate as _deep_annotate # noqa: F401 +from .annotation import _deep_deannotate as _deep_deannotate # noqa: F401 +from .annotation import _shallow_annotate as _shallow_annotate # noqa: F401 +from .base import _expand_cloned +from .base import _from_objects +from .cache_key import HasCacheKey as HasCacheKey # noqa: F401 +from .ddl import sort_tables as sort_tables # noqa: F401 +from .elements import _find_columns as _find_columns +from .elements import _label_reference +from .elements import _textual_label_reference +from .elements import BindParameter +from .elements import ClauseElement +from .elements import ColumnClause +from .elements import ColumnElement +from .elements import Grouping +from .elements import KeyedColumnElement +from .elements import Label +from .elements import NamedColumn +from .elements import Null +from .elements import UnaryExpression +from .schema import Column +from .selectable import Alias +from .selectable import FromClause +from .selectable import FromGrouping +from .selectable import Join +from .selectable import ScalarSelect +from .selectable import SelectBase +from .selectable import TableClause +from .visitors import _ET +from .. import exc +from .. import util +from ..util.typing import Literal +from ..util.typing import Protocol + +if typing.TYPE_CHECKING: + from ._typing import _EquivalentColumnMap + from ._typing import _LimitOffsetType + from ._typing import _TypeEngineArgument + from .elements import BinaryExpression + from .elements import TextClause + from .selectable import _JoinTargetElement + from .selectable import _SelectIterable + from .selectable import Selectable + from .visitors import _TraverseCallableType + from .visitors import ExternallyTraversible + from .visitors import ExternalTraversal + from ..engine.interfaces import _AnyExecuteParams + from ..engine.interfaces import _AnyMultiExecuteParams + from ..engine.interfaces import _AnySingleExecuteParams + from ..engine.interfaces import _CoreSingleExecuteParams + from ..engine.row import Row + +_CE = TypeVar("_CE", bound="ColumnElement[Any]") + + +def join_condition( + a: FromClause, + b: FromClause, + a_subset: Optional[FromClause] = None, + consider_as_foreign_keys: Optional[AbstractSet[ColumnClause[Any]]] = None, +) -> ColumnElement[bool]: + """Create a join condition between two tables or selectables. + + e.g.:: + + join_condition(tablea, tableb) + + would produce an expression along the lines of:: + + tablea.c.id == tableb.c.tablea_id + + The join is determined based on the foreign key relationships + between the two selectables. If there are multiple ways + to join, or no way to join, an error is raised. + + :param a_subset: An optional expression that is a sub-component + of ``a``. An attempt will be made to join to just this sub-component + first before looking at the full ``a`` construct, and if found + will be successful even if there are other ways to join to ``a``. + This allows the "right side" of a join to be passed thereby + providing a "natural join". + + """ + return Join._join_condition( + a, + b, + a_subset=a_subset, + consider_as_foreign_keys=consider_as_foreign_keys, + ) + + +def find_join_source( + clauses: List[FromClause], join_to: FromClause +) -> List[int]: + """Given a list of FROM clauses and a selectable, + return the first index and element from the list of + clauses which can be joined against the selectable. returns + None, None if no match is found. + + e.g.:: + + clause1 = table1.join(table2) + clause2 = table4.join(table5) + + join_to = table2.join(table3) + + find_join_source([clause1, clause2], join_to) == clause1 + + """ + + selectables = list(_from_objects(join_to)) + idx = [] + for i, f in enumerate(clauses): + for s in selectables: + if f.is_derived_from(s): + idx.append(i) + return idx + + +def find_left_clause_that_matches_given( + clauses: Sequence[FromClause], join_from: FromClause +) -> List[int]: + """Given a list of FROM clauses and a selectable, + return the indexes from the list of + clauses which is derived from the selectable. + + """ + + selectables = list(_from_objects(join_from)) + liberal_idx = [] + for i, f in enumerate(clauses): + for s in selectables: + # basic check, if f is derived from s. + # this can be joins containing a table, or an aliased table + # or select statement matching to a table. This check + # will match a table to a selectable that is adapted from + # that table. With Query, this suits the case where a join + # is being made to an adapted entity + if f.is_derived_from(s): + liberal_idx.append(i) + break + + # in an extremely small set of use cases, a join is being made where + # there are multiple FROM clauses where our target table is represented + # in more than one, such as embedded or similar. in this case, do + # another pass where we try to get a more exact match where we aren't + # looking at adaption relationships. + if len(liberal_idx) > 1: + conservative_idx = [] + for idx in liberal_idx: + f = clauses[idx] + for s in selectables: + if set(surface_selectables(f)).intersection( + surface_selectables(s) + ): + conservative_idx.append(idx) + break + if conservative_idx: + return conservative_idx + + return liberal_idx + + +def find_left_clause_to_join_from( + clauses: Sequence[FromClause], + join_to: _JoinTargetElement, + onclause: Optional[ColumnElement[Any]], +) -> List[int]: + """Given a list of FROM clauses, a selectable, + and optional ON clause, return a list of integer indexes from the + clauses list indicating the clauses that can be joined from. + + The presence of an "onclause" indicates that at least one clause can + definitely be joined from; if the list of clauses is of length one + and the onclause is given, returns that index. If the list of clauses + is more than length one, and the onclause is given, attempts to locate + which clauses contain the same columns. + + """ + idx = [] + selectables = set(_from_objects(join_to)) + + # if we are given more than one target clause to join + # from, use the onclause to provide a more specific answer. + # otherwise, don't try to limit, after all, "ON TRUE" is a valid + # on clause + if len(clauses) > 1 and onclause is not None: + resolve_ambiguity = True + cols_in_onclause = _find_columns(onclause) + else: + resolve_ambiguity = False + cols_in_onclause = None + + for i, f in enumerate(clauses): + for s in selectables.difference([f]): + if resolve_ambiguity: + assert cols_in_onclause is not None + if set(f.c).union(s.c).issuperset(cols_in_onclause): + idx.append(i) + break + elif onclause is not None or Join._can_join(f, s): + idx.append(i) + break + + if len(idx) > 1: + # this is the same "hide froms" logic from + # Selectable._get_display_froms + toremove = set( + chain(*[_expand_cloned(f._hide_froms) for f in clauses]) + ) + idx = [i for i in idx if clauses[i] not in toremove] + + # onclause was given and none of them resolved, so assume + # all indexes can match + if not idx and onclause is not None: + return list(range(len(clauses))) + else: + return idx + + +def visit_binary_product( + fn: Callable[ + [BinaryExpression[Any], ColumnElement[Any], ColumnElement[Any]], None + ], + expr: ColumnElement[Any], +) -> None: + """Produce a traversal of the given expression, delivering + column comparisons to the given function. + + The function is of the form:: + + def my_fn(binary, left, right): ... + + For each binary expression located which has a + comparison operator, the product of "left" and + "right" will be delivered to that function, + in terms of that binary. + + Hence an expression like:: + + and_((a + b) == q + func.sum(e + f), j == r) + + would have the traversal: + + .. sourcecode:: text + + a q + a e + a f + b q + b e + b f + j r + + That is, every combination of "left" and + "right" that doesn't further contain + a binary comparison is passed as pairs. + + """ + stack: List[BinaryExpression[Any]] = [] + + def visit(element: ClauseElement) -> Iterator[ColumnElement[Any]]: + if isinstance(element, ScalarSelect): + # we don't want to dig into correlated subqueries, + # those are just column elements by themselves + yield element + elif element.__visit_name__ == "binary" and operators.is_comparison( + element.operator # type: ignore + ): + stack.insert(0, element) # type: ignore + for l in visit(element.left): # type: ignore + for r in visit(element.right): # type: ignore + fn(stack[0], l, r) + stack.pop(0) + for elem in element.get_children(): + visit(elem) + else: + if isinstance(element, ColumnClause): + yield element + for elem in element.get_children(): + yield from visit(elem) + + list(visit(expr)) + visit = None # type: ignore # remove gc cycles + + +def find_tables( + clause: ClauseElement, + *, + check_columns: bool = False, + include_aliases: bool = False, + include_joins: bool = False, + include_selects: bool = False, + include_crud: bool = False, +) -> List[TableClause]: + """locate Table objects within the given expression.""" + + tables: List[TableClause] = [] + _visitors: Dict[str, _TraverseCallableType[Any]] = {} + + if include_selects: + _visitors["select"] = _visitors["compound_select"] = tables.append + + if include_joins: + _visitors["join"] = tables.append + + if include_aliases: + _visitors["alias"] = _visitors["subquery"] = _visitors[ + "tablesample" + ] = _visitors["lateral"] = tables.append + + if include_crud: + _visitors["insert"] = _visitors["update"] = _visitors["delete"] = ( + lambda ent: tables.append(ent.table) + ) + + if check_columns: + + def visit_column(column): + tables.append(column.table) + + _visitors["column"] = visit_column + + _visitors["table"] = tables.append + + visitors.traverse(clause, {}, _visitors) + return tables + + +def unwrap_order_by(clause: Any) -> Any: + """Break up an 'order by' expression into individual column-expressions, + without DESC/ASC/NULLS FIRST/NULLS LAST""" + + cols = util.column_set() + result = [] + stack = deque([clause]) + + # examples + # column -> ASC/DESC == column + # column -> ASC/DESC -> label == column + # column -> label -> ASC/DESC -> label == column + # scalar_select -> label -> ASC/DESC == scalar_select -> label + + while stack: + t = stack.popleft() + if isinstance(t, ColumnElement) and ( + not isinstance(t, UnaryExpression) + or not operators.is_ordering_modifier(t.modifier) # type: ignore + ): + if isinstance(t, Label) and not isinstance( + t.element, ScalarSelect + ): + t = t.element + + if isinstance(t, Grouping): + t = t.element + + stack.append(t) + continue + elif isinstance(t, _label_reference): + t = t.element + + stack.append(t) + continue + if isinstance(t, (_textual_label_reference)): + continue + if t not in cols: + cols.add(t) + result.append(t) + + else: + for c in t.get_children(): + stack.append(c) + return result + + +def unwrap_label_reference(element): + def replace( + element: ExternallyTraversible, **kw: Any + ) -> Optional[ExternallyTraversible]: + if isinstance(element, _label_reference): + return element.element + elif isinstance(element, _textual_label_reference): + assert False, "can't unwrap a textual label reference" + return None + + return visitors.replacement_traverse(element, {}, replace) + + +def expand_column_list_from_order_by(collist, order_by): + """Given the columns clause and ORDER BY of a selectable, + return a list of column expressions that can be added to the collist + corresponding to the ORDER BY, without repeating those already + in the collist. + + """ + cols_already_present = { + col.element if col._order_by_label_element is not None else col + for col in collist + } + + to_look_for = list(chain(*[unwrap_order_by(o) for o in order_by])) + + return [col for col in to_look_for if col not in cols_already_present] + + +def clause_is_present(clause, search): + """Given a target clause and a second to search within, return True + if the target is plainly present in the search without any + subqueries or aliases involved. + + Basically descends through Joins. + + """ + + for elem in surface_selectables(search): + if clause == elem: # use == here so that Annotated's compare + return True + else: + return False + + +def tables_from_leftmost(clause: FromClause) -> Iterator[FromClause]: + if isinstance(clause, Join): + yield from tables_from_leftmost(clause.left) + yield from tables_from_leftmost(clause.right) + elif isinstance(clause, FromGrouping): + yield from tables_from_leftmost(clause.element) + else: + yield clause + + +def surface_selectables(clause): + stack = [clause] + while stack: + elem = stack.pop() + yield elem + if isinstance(elem, Join): + stack.extend((elem.left, elem.right)) + elif isinstance(elem, FromGrouping): + stack.append(elem.element) + + +def surface_selectables_only(clause: ClauseElement) -> Iterator[ClauseElement]: + stack = [clause] + while stack: + elem = stack.pop() + if isinstance(elem, (TableClause, Alias)): + yield elem + if isinstance(elem, Join): + stack.extend((elem.left, elem.right)) + elif isinstance(elem, FromGrouping): + stack.append(elem.element) + elif isinstance(elem, ColumnClause): + if elem.table is not None: + stack.append(elem.table) + else: + yield elem + elif elem is not None: + yield elem + + +def extract_first_column_annotation(column, annotation_name): + filter_ = (FromGrouping, SelectBase) + + stack = deque([column]) + while stack: + elem = stack.popleft() + if annotation_name in elem._annotations: + return elem._annotations[annotation_name] + for sub in elem.get_children(): + if isinstance(sub, filter_): + continue + stack.append(sub) + return None + + +def selectables_overlap(left: FromClause, right: FromClause) -> bool: + """Return True if left/right have some overlapping selectable""" + + return bool( + set(surface_selectables(left)).intersection(surface_selectables(right)) + ) + + +def bind_values(clause): + """Return an ordered list of "bound" values in the given clause. + + E.g.:: + + >>> expr = and_(table.c.foo == 5, table.c.foo == 7) + >>> bind_values(expr) + [5, 7] + """ + + v = [] + + def visit_bindparam(bind): + v.append(bind.effective_value) + + visitors.traverse(clause, {}, {"bindparam": visit_bindparam}) + return v + + +def _quote_ddl_expr(element): + if isinstance(element, str): + element = element.replace("'", "''") + return "'%s'" % element + else: + return repr(element) + + +class _repr_base: + _LIST: int = 0 + _TUPLE: int = 1 + _DICT: int = 2 + + __slots__ = ("max_chars",) + + max_chars: int + + def trunc(self, value: Any) -> str: + rep = repr(value) + lenrep = len(rep) + if lenrep > self.max_chars: + segment_length = self.max_chars // 2 + rep = ( + rep[0:segment_length] + + ( + " ... (%d characters truncated) ... " + % (lenrep - self.max_chars) + ) + + rep[-segment_length:] + ) + return rep + + +def _repr_single_value(value): + rp = _repr_base() + rp.max_chars = 300 + return rp.trunc(value) + + +class _repr_row(_repr_base): + """Provide a string view of a row.""" + + __slots__ = ("row",) + + def __init__(self, row: Row[Any], max_chars: int = 300): + self.row = row + self.max_chars = max_chars + + def __repr__(self) -> str: + trunc = self.trunc + return "(%s%s)" % ( + ", ".join(trunc(value) for value in self.row), + "," if len(self.row) == 1 else "", + ) + + +class _long_statement(str): + def __str__(self) -> str: + lself = len(self) + if lself > 500: + lleft = 250 + lright = 100 + trunc = lself - lleft - lright + return ( + f"{self[0:lleft]} ... {trunc} " + f"characters truncated ... {self[-lright:]}" + ) + else: + return str.__str__(self) + + +class _repr_params(_repr_base): + """Provide a string view of bound parameters. + + Truncates display to a given number of 'multi' parameter sets, + as well as long values to a given number of characters. + + """ + + __slots__ = "params", "batches", "ismulti", "max_params" + + def __init__( + self, + params: Optional[_AnyExecuteParams], + batches: int, + max_params: int = 100, + max_chars: int = 300, + ismulti: Optional[bool] = None, + ): + self.params = params + self.ismulti = ismulti + self.batches = batches + self.max_chars = max_chars + self.max_params = max_params + + def __repr__(self) -> str: + if self.ismulti is None: + return self.trunc(self.params) + + if isinstance(self.params, list): + typ = self._LIST + + elif isinstance(self.params, tuple): + typ = self._TUPLE + elif isinstance(self.params, dict): + typ = self._DICT + else: + return self.trunc(self.params) + + if self.ismulti: + multi_params = cast( + "_AnyMultiExecuteParams", + self.params, + ) + + if len(self.params) > self.batches: + msg = ( + " ... displaying %i of %i total bound parameter sets ... " + ) + return " ".join( + ( + self._repr_multi( + multi_params[: self.batches - 2], + typ, + )[0:-1], + msg % (self.batches, len(self.params)), + self._repr_multi(multi_params[-2:], typ)[1:], + ) + ) + else: + return self._repr_multi(multi_params, typ) + else: + return self._repr_params( + cast( + "_AnySingleExecuteParams", + self.params, + ), + typ, + ) + + def _repr_multi( + self, + multi_params: _AnyMultiExecuteParams, + typ: int, + ) -> str: + if multi_params: + if isinstance(multi_params[0], list): + elem_type = self._LIST + elif isinstance(multi_params[0], tuple): + elem_type = self._TUPLE + elif isinstance(multi_params[0], dict): + elem_type = self._DICT + else: + assert False, "Unknown parameter type %s" % ( + type(multi_params[0]) + ) + + elements = ", ".join( + self._repr_params(params, elem_type) for params in multi_params + ) + else: + elements = "" + + if typ == self._LIST: + return "[%s]" % elements + else: + return "(%s)" % elements + + def _get_batches(self, params: Iterable[Any]) -> Any: + lparams = list(params) + lenparams = len(lparams) + if lenparams > self.max_params: + lleft = self.max_params // 2 + return ( + lparams[0:lleft], + lparams[-lleft:], + lenparams - self.max_params, + ) + else: + return lparams, None, None + + def _repr_params( + self, + params: _AnySingleExecuteParams, + typ: int, + ) -> str: + if typ is self._DICT: + return self._repr_param_dict( + cast("_CoreSingleExecuteParams", params) + ) + elif typ is self._TUPLE: + return self._repr_param_tuple(cast("Sequence[Any]", params)) + else: + return self._repr_param_list(params) + + def _repr_param_dict(self, params: _CoreSingleExecuteParams) -> str: + trunc = self.trunc + ( + items_first_batch, + items_second_batch, + trunclen, + ) = self._get_batches(params.items()) + + if items_second_batch: + text = "{%s" % ( + ", ".join( + f"{key!r}: {trunc(value)}" + for key, value in items_first_batch + ) + ) + text += f" ... {trunclen} parameters truncated ... " + text += "%s}" % ( + ", ".join( + f"{key!r}: {trunc(value)}" + for key, value in items_second_batch + ) + ) + else: + text = "{%s}" % ( + ", ".join( + f"{key!r}: {trunc(value)}" + for key, value in items_first_batch + ) + ) + return text + + def _repr_param_tuple(self, params: Sequence[Any]) -> str: + trunc = self.trunc + + ( + items_first_batch, + items_second_batch, + trunclen, + ) = self._get_batches(params) + + if items_second_batch: + text = "(%s" % ( + ", ".join(trunc(value) for value in items_first_batch) + ) + text += f" ... {trunclen} parameters truncated ... " + text += "%s)" % ( + ", ".join(trunc(value) for value in items_second_batch), + ) + else: + text = "(%s%s)" % ( + ", ".join(trunc(value) for value in items_first_batch), + "," if len(items_first_batch) == 1 else "", + ) + return text + + def _repr_param_list(self, params: _AnySingleExecuteParams) -> str: + trunc = self.trunc + ( + items_first_batch, + items_second_batch, + trunclen, + ) = self._get_batches(params) + + if items_second_batch: + text = "[%s" % ( + ", ".join(trunc(value) for value in items_first_batch) + ) + text += f" ... {trunclen} parameters truncated ... " + text += "%s]" % ( + ", ".join(trunc(value) for value in items_second_batch) + ) + else: + text = "[%s]" % ( + ", ".join(trunc(value) for value in items_first_batch) + ) + return text + + +def adapt_criterion_to_null(crit: _CE, nulls: Collection[Any]) -> _CE: + """given criterion containing bind params, convert selected elements + to IS NULL. + + """ + + def visit_binary(binary): + if ( + isinstance(binary.left, BindParameter) + and binary.left._identifying_key in nulls + ): + # reverse order if the NULL is on the left side + binary.left = binary.right + binary.right = Null() + binary.operator = operators.is_ + binary.negate = operators.is_not + elif ( + isinstance(binary.right, BindParameter) + and binary.right._identifying_key in nulls + ): + binary.right = Null() + binary.operator = operators.is_ + binary.negate = operators.is_not + + return visitors.cloned_traverse(crit, {}, {"binary": visit_binary}) + + +def splice_joins( + left: Optional[FromClause], + right: Optional[FromClause], + stop_on: Optional[FromClause] = None, +) -> Optional[FromClause]: + if left is None: + return right + + stack: List[Tuple[Optional[FromClause], Optional[Join]]] = [(right, None)] + + adapter = ClauseAdapter(left) + ret = None + while stack: + right, prevright = stack.pop() + if isinstance(right, Join) and right is not stop_on: + right = right._clone() + right.onclause = adapter.traverse(right.onclause) + stack.append((right.left, right)) + else: + right = adapter.traverse(right) + if prevright is not None: + assert right is not None + prevright.left = right + if ret is None: + ret = right + + return ret + + +@overload +def reduce_columns( + columns: Iterable[ColumnElement[Any]], + *clauses: Optional[ClauseElement], + **kw: bool, +) -> Sequence[ColumnElement[Any]]: ... + + +@overload +def reduce_columns( + columns: _SelectIterable, + *clauses: Optional[ClauseElement], + **kw: bool, +) -> Sequence[Union[ColumnElement[Any], TextClause]]: ... + + +def reduce_columns( + columns: _SelectIterable, + *clauses: Optional[ClauseElement], + **kw: bool, +) -> Collection[Union[ColumnElement[Any], TextClause]]: + r"""given a list of columns, return a 'reduced' set based on natural + equivalents. + + the set is reduced to the smallest list of columns which have no natural + equivalent present in the list. A "natural equivalent" means that two + columns will ultimately represent the same value because they are related + by a foreign key. + + \*clauses is an optional list of join clauses which will be traversed + to further identify columns that are "equivalent". + + \**kw may specify 'ignore_nonexistent_tables' to ignore foreign keys + whose tables are not yet configured, or columns that aren't yet present. + + This function is primarily used to determine the most minimal "primary + key" from a selectable, by reducing the set of primary key columns present + in the selectable to just those that are not repeated. + + """ + ignore_nonexistent_tables = kw.pop("ignore_nonexistent_tables", False) + only_synonyms = kw.pop("only_synonyms", False) + + column_set = util.OrderedSet(columns) + cset_no_text: util.OrderedSet[ColumnElement[Any]] = column_set.difference( + c for c in column_set if is_text_clause(c) # type: ignore + ) + + omit = util.column_set() + for col in cset_no_text: + for fk in chain(*[c.foreign_keys for c in col.proxy_set]): + for c in cset_no_text: + if c is col: + continue + try: + fk_col = fk.column + except exc.NoReferencedColumnError: + # TODO: add specific coverage here + # to test/sql/test_selectable ReduceTest + if ignore_nonexistent_tables: + continue + else: + raise + except exc.NoReferencedTableError: + # TODO: add specific coverage here + # to test/sql/test_selectable ReduceTest + if ignore_nonexistent_tables: + continue + else: + raise + if fk_col.shares_lineage(c) and ( + not only_synonyms or c.name == col.name + ): + omit.add(col) + break + + if clauses: + + def visit_binary(binary): + if binary.operator == operators.eq: + cols = util.column_set( + chain( + *[c.proxy_set for c in cset_no_text.difference(omit)] + ) + ) + if binary.left in cols and binary.right in cols: + for c in reversed(cset_no_text): + if c.shares_lineage(binary.right) and ( + not only_synonyms or c.name == binary.left.name + ): + omit.add(c) + break + + for clause in clauses: + if clause is not None: + visitors.traverse(clause, {}, {"binary": visit_binary}) + + return column_set.difference(omit) + + +def criterion_as_pairs( + expression, + consider_as_foreign_keys=None, + consider_as_referenced_keys=None, + any_operator=False, +): + """traverse an expression and locate binary criterion pairs.""" + + if consider_as_foreign_keys and consider_as_referenced_keys: + raise exc.ArgumentError( + "Can only specify one of " + "'consider_as_foreign_keys' or " + "'consider_as_referenced_keys'" + ) + + def col_is(a, b): + # return a is b + return a.compare(b) + + def visit_binary(binary): + if not any_operator and binary.operator is not operators.eq: + return + if not isinstance(binary.left, ColumnElement) or not isinstance( + binary.right, ColumnElement + ): + return + + if consider_as_foreign_keys: + if binary.left in consider_as_foreign_keys and ( + col_is(binary.right, binary.left) + or binary.right not in consider_as_foreign_keys + ): + pairs.append((binary.right, binary.left)) + elif binary.right in consider_as_foreign_keys and ( + col_is(binary.left, binary.right) + or binary.left not in consider_as_foreign_keys + ): + pairs.append((binary.left, binary.right)) + elif consider_as_referenced_keys: + if binary.left in consider_as_referenced_keys and ( + col_is(binary.right, binary.left) + or binary.right not in consider_as_referenced_keys + ): + pairs.append((binary.left, binary.right)) + elif binary.right in consider_as_referenced_keys and ( + col_is(binary.left, binary.right) + or binary.left not in consider_as_referenced_keys + ): + pairs.append((binary.right, binary.left)) + else: + if isinstance(binary.left, Column) and isinstance( + binary.right, Column + ): + if binary.left.references(binary.right): + pairs.append((binary.right, binary.left)) + elif binary.right.references(binary.left): + pairs.append((binary.left, binary.right)) + + pairs: List[Tuple[ColumnElement[Any], ColumnElement[Any]]] = [] + visitors.traverse(expression, {}, {"binary": visit_binary}) + return pairs + + +class ClauseAdapter(visitors.ReplacingExternalTraversal): + """Clones and modifies clauses based on column correspondence. + + E.g.:: + + table1 = Table( + "sometable", + metadata, + Column("col1", Integer), + Column("col2", Integer), + ) + table2 = Table( + "someothertable", + metadata, + Column("col1", Integer), + Column("col2", Integer), + ) + + condition = table1.c.col1 == table2.c.col1 + + make an alias of table1:: + + s = table1.alias("foo") + + calling ``ClauseAdapter(s).traverse(condition)`` converts + condition to read:: + + s.c.col1 == table2.c.col1 + + """ + + __slots__ = ( + "__traverse_options__", + "selectable", + "include_fn", + "exclude_fn", + "equivalents", + "adapt_on_names", + "adapt_from_selectables", + ) + + def __init__( + self, + selectable: Selectable, + equivalents: Optional[_EquivalentColumnMap] = None, + include_fn: Optional[Callable[[ClauseElement], bool]] = None, + exclude_fn: Optional[Callable[[ClauseElement], bool]] = None, + adapt_on_names: bool = False, + anonymize_labels: bool = False, + adapt_from_selectables: Optional[AbstractSet[FromClause]] = None, + ): + self.__traverse_options__ = { + "stop_on": [selectable], + "anonymize_labels": anonymize_labels, + } + self.selectable = selectable + self.include_fn = include_fn + self.exclude_fn = exclude_fn + self.equivalents = util.column_dict(equivalents or {}) + self.adapt_on_names = adapt_on_names + self.adapt_from_selectables = adapt_from_selectables + + if TYPE_CHECKING: + + @overload + def traverse(self, obj: Literal[None]) -> None: ... + + # note this specializes the ReplacingExternalTraversal.traverse() + # method to state + # that we will return the same kind of ExternalTraversal object as + # we were given. This is probably not 100% true, such as it's + # possible for us to swap out Alias for Table at the top level. + # Ideally there could be overloads specific to ColumnElement and + # FromClause but Mypy is not accepting those as compatible with + # the base ReplacingExternalTraversal + @overload + def traverse(self, obj: _ET) -> _ET: ... + + def traverse( + self, obj: Optional[ExternallyTraversible] + ) -> Optional[ExternallyTraversible]: ... + + def _corresponding_column( + self, col, require_embedded, _seen=util.EMPTY_SET + ): + newcol = self.selectable.corresponding_column( + col, require_embedded=require_embedded + ) + if newcol is None and col in self.equivalents and col not in _seen: + for equiv in self.equivalents[col]: + newcol = self._corresponding_column( + equiv, + require_embedded=require_embedded, + _seen=_seen.union([col]), + ) + if newcol is not None: + return newcol + + if ( + self.adapt_on_names + and newcol is None + and isinstance(col, NamedColumn) + ): + newcol = self.selectable.exported_columns.get(col.name) + return newcol + + @util.preload_module("sqlalchemy.sql.functions") + def replace( + self, col: _ET, _include_singleton_constants: bool = False + ) -> Optional[_ET]: + functions = util.preloaded.sql_functions + + # TODO: cython candidate + + if self.include_fn and not self.include_fn(col): # type: ignore + return None + elif self.exclude_fn and self.exclude_fn(col): # type: ignore + return None + + if isinstance(col, FromClause) and not isinstance( + col, functions.FunctionElement + ): + if self.selectable.is_derived_from(col): + if self.adapt_from_selectables: + for adp in self.adapt_from_selectables: + if adp.is_derived_from(col): + break + else: + return None + return self.selectable # type: ignore + elif isinstance(col, Alias) and isinstance( + col.element, TableClause + ): + # we are a SELECT statement and not derived from an alias of a + # table (which nonetheless may be a table our SELECT derives + # from), so return the alias to prevent further traversal + # or + # we are an alias of a table and we are not derived from an + # alias of a table (which nonetheless may be the same table + # as ours) so, same thing + return col + else: + # other cases where we are a selectable and the element + # is another join or selectable that contains a table which our + # selectable derives from, that we want to process + return None + + elif not isinstance(col, ColumnElement): + return None + elif not _include_singleton_constants and col._is_singleton_constant: + # dont swap out NULL, TRUE, FALSE for a label name + # in a SQL statement that's being rewritten, + # leave them as the constant. This is first noted in #6259, + # however the logic to check this moved here as of #7154 so that + # it is made specific to SQL rewriting and not all column + # correspondence + + return None + + if "adapt_column" in col._annotations: + col = col._annotations["adapt_column"] + + if TYPE_CHECKING: + assert isinstance(col, KeyedColumnElement) + + if self.adapt_from_selectables and col not in self.equivalents: + for adp in self.adapt_from_selectables: + if adp.c.corresponding_column(col, False) is not None: + break + else: + return None + + if TYPE_CHECKING: + assert isinstance(col, KeyedColumnElement) + + return self._corresponding_column( # type: ignore + col, require_embedded=True + ) + + +class _ColumnLookup(Protocol): + @overload + def __getitem__(self, key: None) -> None: ... + + @overload + def __getitem__(self, key: ColumnClause[Any]) -> ColumnClause[Any]: ... + + @overload + def __getitem__(self, key: ColumnElement[Any]) -> ColumnElement[Any]: ... + + @overload + def __getitem__(self, key: _ET) -> _ET: ... + + def __getitem__(self, key: Any) -> Any: ... + + +class ColumnAdapter(ClauseAdapter): + """Extends ClauseAdapter with extra utility functions. + + Key aspects of ColumnAdapter include: + + * Expressions that are adapted are stored in a persistent + .columns collection; so that an expression E adapted into + an expression E1, will return the same object E1 when adapted + a second time. This is important in particular for things like + Label objects that are anonymized, so that the ColumnAdapter can + be used to present a consistent "adapted" view of things. + + * Exclusion of items from the persistent collection based on + include/exclude rules, but also independent of hash identity. + This because "annotated" items all have the same hash identity as their + parent. + + * "wrapping" capability is added, so that the replacement of an expression + E can proceed through a series of adapters. This differs from the + visitor's "chaining" feature in that the resulting object is passed + through all replacing functions unconditionally, rather than stopping + at the first one that returns non-None. + + * An adapt_required option, used by eager loading to indicate that + We don't trust a result row column that is not translated. + This is to prevent a column from being interpreted as that + of the child row in a self-referential scenario, see + inheritance/test_basic.py->EagerTargetingTest.test_adapt_stringency + + """ + + __slots__ = ( + "columns", + "adapt_required", + "allow_label_resolve", + "_wrap", + "__weakref__", + ) + + columns: _ColumnLookup + + def __init__( + self, + selectable: Selectable, + equivalents: Optional[_EquivalentColumnMap] = None, + adapt_required: bool = False, + include_fn: Optional[Callable[[ClauseElement], bool]] = None, + exclude_fn: Optional[Callable[[ClauseElement], bool]] = None, + adapt_on_names: bool = False, + allow_label_resolve: bool = True, + anonymize_labels: bool = False, + adapt_from_selectables: Optional[AbstractSet[FromClause]] = None, + ): + super().__init__( + selectable, + equivalents, + include_fn=include_fn, + exclude_fn=exclude_fn, + adapt_on_names=adapt_on_names, + anonymize_labels=anonymize_labels, + adapt_from_selectables=adapt_from_selectables, + ) + + self.columns = util.WeakPopulateDict(self._locate_col) # type: ignore + if self.include_fn or self.exclude_fn: + self.columns = self._IncludeExcludeMapping(self, self.columns) + self.adapt_required = adapt_required + self.allow_label_resolve = allow_label_resolve + self._wrap = None + + class _IncludeExcludeMapping: + def __init__(self, parent, columns): + self.parent = parent + self.columns = columns + + def __getitem__(self, key): + if ( + self.parent.include_fn and not self.parent.include_fn(key) + ) or (self.parent.exclude_fn and self.parent.exclude_fn(key)): + if self.parent._wrap: + return self.parent._wrap.columns[key] + else: + return key + return self.columns[key] + + def wrap(self, adapter): + ac = copy.copy(self) + ac._wrap = adapter + ac.columns = util.WeakPopulateDict(ac._locate_col) # type: ignore + if ac.include_fn or ac.exclude_fn: + ac.columns = self._IncludeExcludeMapping(ac, ac.columns) + + return ac + + @overload + def traverse(self, obj: Literal[None]) -> None: ... + + @overload + def traverse(self, obj: _ET) -> _ET: ... + + def traverse( + self, obj: Optional[ExternallyTraversible] + ) -> Optional[ExternallyTraversible]: + return self.columns[obj] + + def chain(self, visitor: ExternalTraversal) -> ColumnAdapter: + assert isinstance(visitor, ColumnAdapter) + + return super().chain(visitor) + + if TYPE_CHECKING: + + @property + def visitor_iterator(self) -> Iterator[ColumnAdapter]: ... + + adapt_clause = traverse + adapt_list = ClauseAdapter.copy_and_process + + def adapt_check_present( + self, col: ColumnElement[Any] + ) -> Optional[ColumnElement[Any]]: + newcol = self.columns[col] + + if newcol is col and self._corresponding_column(col, True) is None: + return None + + return newcol + + def _locate_col( + self, col: ColumnElement[Any] + ) -> Optional[ColumnElement[Any]]: + # both replace and traverse() are overly complicated for what + # we are doing here and we would do better to have an inlined + # version that doesn't build up as much overhead. the issue is that + # sometimes the lookup does in fact have to adapt the insides of + # say a labeled scalar subquery. However, if the object is an + # Immutable, i.e. Column objects, we can skip the "clone" / + # "copy internals" part since those will be no-ops in any case. + # additionally we want to catch singleton objects null/true/false + # and make sure they are adapted as well here. + + if col._is_immutable: + for vis in self.visitor_iterator: + c = vis.replace(col, _include_singleton_constants=True) + if c is not None: + break + else: + c = col + else: + c = ClauseAdapter.traverse(self, col) + + if self._wrap: + c2 = self._wrap._locate_col(c) + if c2 is not None: + c = c2 + + if self.adapt_required and c is col: + return None + + # allow_label_resolve is consumed by one case for joined eager loading + # as part of its logic to prevent its own columns from being affected + # by .order_by(). Before full typing were applied to the ORM, this + # logic would set this attribute on the incoming object (which is + # typically a column, but we have a test for it being a non-column + # object) if no column were found. While this seemed to + # have no negative effects, this adjustment should only occur on the + # new column which is assumed to be local to an adapted selectable. + if c is not col: + c._allow_label_resolve = self.allow_label_resolve + + return c + + +def _offset_or_limit_clause( + element: _LimitOffsetType, + name: Optional[str] = None, + type_: Optional[_TypeEngineArgument[int]] = None, +) -> ColumnElement[int]: + """Convert the given value to an "offset or limit" clause. + + This handles incoming integers and converts to an expression; if + an expression is already given, it is passed through. + + """ + return coercions.expect( + roles.LimitOffsetRole, element, name=name, type_=type_ + ) + + +def _offset_or_limit_clause_asint_if_possible( + clause: _LimitOffsetType, +) -> _LimitOffsetType: + """Return the offset or limit clause as a simple integer if possible, + else return the clause. + + """ + if clause is None: + return None + if hasattr(clause, "_limit_offset_value"): + value = clause._limit_offset_value + return util.asint(value) + else: + return clause + + +def _make_slice( + limit_clause: _LimitOffsetType, + offset_clause: _LimitOffsetType, + start: int, + stop: int, +) -> Tuple[Optional[ColumnElement[int]], Optional[ColumnElement[int]]]: + """Compute LIMIT/OFFSET in terms of slice start/end""" + + # for calculated limit/offset, try to do the addition of + # values to offset in Python, however if a SQL clause is present + # then the addition has to be on the SQL side. + + # TODO: typing is finding a few gaps in here, see if they can be + # closed up + + if start is not None and stop is not None: + offset_clause = _offset_or_limit_clause_asint_if_possible( + offset_clause + ) + if offset_clause is None: + offset_clause = 0 + + if start != 0: + offset_clause = offset_clause + start # type: ignore + + if offset_clause == 0: + offset_clause = None + else: + assert offset_clause is not None + offset_clause = _offset_or_limit_clause(offset_clause) + + limit_clause = _offset_or_limit_clause(stop - start) + + elif start is None and stop is not None: + limit_clause = _offset_or_limit_clause(stop) + elif start is not None and stop is None: + offset_clause = _offset_or_limit_clause_asint_if_possible( + offset_clause + ) + if offset_clause is None: + offset_clause = 0 + + if start != 0: + offset_clause = offset_clause + start + + if offset_clause == 0: + offset_clause = None + else: + offset_clause = _offset_or_limit_clause(offset_clause) + + return limit_clause, offset_clause diff --git a/venv/Lib/site-packages/sqlalchemy/sql/visitors.py b/venv/Lib/site-packages/sqlalchemy/sql/visitors.py new file mode 100644 index 0000000..5056a70 --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/sql/visitors.py @@ -0,0 +1,1164 @@ +# sql/visitors.py +# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors +# +# +# This module is part of SQLAlchemy and is released under +# the MIT License: https://www.opensource.org/licenses/mit-license.php + +"""Visitor/traversal interface and library functions.""" + +from __future__ import annotations + +from collections import deque +from enum import Enum +import itertools +import operator +import typing +from typing import Any +from typing import Callable +from typing import cast +from typing import ClassVar +from typing import Dict +from typing import Iterable +from typing import Iterator +from typing import List +from typing import Mapping +from typing import Optional +from typing import overload +from typing import Tuple +from typing import Type +from typing import TYPE_CHECKING +from typing import TypeVar +from typing import Union + +from .. import exc +from .. import util +from ..util import langhelpers +from ..util._has_cy import HAS_CYEXTENSION +from ..util.typing import Literal +from ..util.typing import Protocol +from ..util.typing import Self + +if TYPE_CHECKING: + from .annotation import _AnnotationDict + from .elements import ColumnElement + +if typing.TYPE_CHECKING or not HAS_CYEXTENSION: + from ._py_util import prefix_anon_map as prefix_anon_map + from ._py_util import cache_anon_map as anon_map +else: + from sqlalchemy.cyextension.util import ( # noqa: F401,E501 + prefix_anon_map as prefix_anon_map, + ) + from sqlalchemy.cyextension.util import ( # noqa: F401,E501 + cache_anon_map as anon_map, + ) + + +__all__ = [ + "iterate", + "traverse_using", + "traverse", + "cloned_traverse", + "replacement_traverse", + "Visitable", + "ExternalTraversal", + "InternalTraversal", + "anon_map", +] + + +class _CompilerDispatchType(Protocol): + def __call__(_self, self: Visitable, visitor: Any, **kw: Any) -> Any: ... + + +class Visitable: + """Base class for visitable objects. + + :class:`.Visitable` is used to implement the SQL compiler dispatch + functions. Other forms of traversal such as for cache key generation + are implemented separately using the :class:`.HasTraverseInternals` + interface. + + .. versionchanged:: 2.0 The :class:`.Visitable` class was named + :class:`.Traversible` in the 1.4 series; the name is changed back + to :class:`.Visitable` in 2.0 which is what it was prior to 1.4. + + Both names remain importable in both 1.4 and 2.0 versions. + + """ + + __slots__ = () + + __visit_name__: str + + _original_compiler_dispatch: _CompilerDispatchType + + if typing.TYPE_CHECKING: + + def _compiler_dispatch(self, visitor: Any, **kw: Any) -> str: ... + + def __init_subclass__(cls) -> None: + if "__visit_name__" in cls.__dict__: + cls._generate_compiler_dispatch() + super().__init_subclass__() + + @classmethod + def _generate_compiler_dispatch(cls) -> None: + visit_name = cls.__visit_name__ + + if "_compiler_dispatch" in cls.__dict__: + # class has a fixed _compiler_dispatch() method. + # copy it to "original" so that we can get it back if + # sqlalchemy.ext.compiles overrides it. + cls._original_compiler_dispatch = cls._compiler_dispatch + return + + if not isinstance(visit_name, str): + raise exc.InvalidRequestError( + f"__visit_name__ on class {cls.__name__} must be a string " + "at the class level" + ) + + name = "visit_%s" % visit_name + getter = operator.attrgetter(name) + + def _compiler_dispatch( + self: Visitable, visitor: Any, **kw: Any + ) -> str: + """Look for an attribute named "visit_" on the + visitor, and call it with the same kw params. + + """ + try: + meth = getter(visitor) + except AttributeError as err: + return visitor.visit_unsupported_compilation(self, err, **kw) # type: ignore # noqa: E501 + else: + return meth(self, **kw) # type: ignore # noqa: E501 + + cls._compiler_dispatch = ( # type: ignore + cls._original_compiler_dispatch + ) = _compiler_dispatch + + def __class_getitem__(cls, key: Any) -> Any: + # allow generic classes in py3.9+ + return cls + + +class InternalTraversal(Enum): + r"""Defines visitor symbols used for internal traversal. + + The :class:`.InternalTraversal` class is used in two ways. One is that + it can serve as the superclass for an object that implements the + various visit methods of the class. The other is that the symbols + themselves of :class:`.InternalTraversal` are used within + the ``_traverse_internals`` collection. Such as, the :class:`.Case` + object defines ``_traverse_internals`` as :: + + class Case(ColumnElement[_T]): + _traverse_internals = [ + ("value", InternalTraversal.dp_clauseelement), + ("whens", InternalTraversal.dp_clauseelement_tuples), + ("else_", InternalTraversal.dp_clauseelement), + ] + + Above, the :class:`.Case` class indicates its internal state as the + attributes named ``value``, ``whens``, and ``else_``. They each + link to an :class:`.InternalTraversal` method which indicates the type + of datastructure to which each attribute refers. + + Using the ``_traverse_internals`` structure, objects of type + :class:`.InternalTraversible` will have the following methods automatically + implemented: + + * :meth:`.HasTraverseInternals.get_children` + + * :meth:`.HasTraverseInternals._copy_internals` + + * :meth:`.HasCacheKey._gen_cache_key` + + Subclasses can also implement these methods directly, particularly for the + :meth:`.HasTraverseInternals._copy_internals` method, when special steps + are needed. + + .. versionadded:: 1.4 + + """ + + dp_has_cache_key = "HC" + """Visit a :class:`.HasCacheKey` object.""" + + dp_has_cache_key_list = "HL" + """Visit a list of :class:`.HasCacheKey` objects.""" + + dp_clauseelement = "CE" + """Visit a :class:`_expression.ClauseElement` object.""" + + dp_fromclause_canonical_column_collection = "FC" + """Visit a :class:`_expression.FromClause` object in the context of the + ``columns`` attribute. + + The column collection is "canonical", meaning it is the originally + defined location of the :class:`.ColumnClause` objects. Right now + this means that the object being visited is a + :class:`_expression.TableClause` + or :class:`_schema.Table` object only. + + """ + + dp_clauseelement_tuples = "CTS" + """Visit a list of tuples which contain :class:`_expression.ClauseElement` + objects. + + """ + + dp_clauseelement_list = "CL" + """Visit a list of :class:`_expression.ClauseElement` objects. + + """ + + dp_clauseelement_tuple = "CT" + """Visit a tuple of :class:`_expression.ClauseElement` objects. + + """ + + dp_executable_options = "EO" + + dp_with_context_options = "WC" + + dp_fromclause_ordered_set = "CO" + """Visit an ordered set of :class:`_expression.FromClause` objects. """ + + dp_string = "S" + """Visit a plain string value. + + Examples include table and column names, bound parameter keys, special + keywords such as "UNION", "UNION ALL". + + The string value is considered to be significant for cache key + generation. + + """ + + dp_string_list = "SL" + """Visit a list of strings.""" + + dp_anon_name = "AN" + """Visit a potentially "anonymized" string value. + + The string value is considered to be significant for cache key + generation. + + """ + + dp_boolean = "B" + """Visit a boolean value. + + The boolean value is considered to be significant for cache key + generation. + + """ + + dp_operator = "O" + """Visit an operator. + + The operator is a function from the :mod:`sqlalchemy.sql.operators` + module. + + The operator value is considered to be significant for cache key + generation. + + """ + + dp_type = "T" + """Visit a :class:`.TypeEngine` object + + The type object is considered to be significant for cache key + generation. + + """ + + dp_plain_dict = "PD" + """Visit a dictionary with string keys. + + The keys of the dictionary should be strings, the values should + be immutable and hashable. The dictionary is considered to be + significant for cache key generation. + + """ + + dp_dialect_options = "DO" + """Visit a dialect options structure.""" + + dp_string_clauseelement_dict = "CD" + """Visit a dictionary of string keys to :class:`_expression.ClauseElement` + objects. + + """ + + dp_string_multi_dict = "MD" + """Visit a dictionary of string keys to values which may either be + plain immutable/hashable or :class:`.HasCacheKey` objects. + + """ + + dp_annotations_key = "AK" + """Visit the _annotations_cache_key element. + + This is a dictionary of additional information about a ClauseElement + that modifies its role. It should be included when comparing or caching + objects, however generating this key is relatively expensive. Visitors + should check the "_annotations" dict for non-None first before creating + this key. + + """ + + dp_plain_obj = "PO" + """Visit a plain python object. + + The value should be immutable and hashable, such as an integer. + The value is considered to be significant for cache key generation. + + """ + + dp_named_ddl_element = "DD" + """Visit a simple named DDL element. + + The current object used by this method is the :class:`.Sequence`. + + The object is only considered to be important for cache key generation + as far as its name, but not any other aspects of it. + + """ + + dp_prefix_sequence = "PS" + """Visit the sequence represented by :class:`_expression.HasPrefixes` + or :class:`_expression.HasSuffixes`. + + """ + + dp_table_hint_list = "TH" + """Visit the ``_hints`` collection of a :class:`_expression.Select` + object. + + """ + + dp_setup_join_tuple = "SJ" + + dp_memoized_select_entities = "ME" + + dp_statement_hint_list = "SH" + """Visit the ``_statement_hints`` collection of a + :class:`_expression.Select` + object. + + """ + + dp_unknown_structure = "UK" + """Visit an unknown structure. + + """ + + dp_dml_ordered_values = "DML_OV" + """Visit the values() ordered tuple list of an + :class:`_expression.Update` object.""" + + dp_dml_values = "DML_V" + """Visit the values() dictionary of a :class:`.ValuesBase` + (e.g. Insert or Update) object. + + """ + + dp_dml_multi_values = "DML_MV" + """Visit the values() multi-valued list of dictionaries of an + :class:`_expression.Insert` object. + + """ + + dp_propagate_attrs = "PA" + """Visit the propagate attrs dict. This hardcodes to the particular + elements we care about right now.""" + + """Symbols that follow are additional symbols that are useful in + caching applications. + + Traversals for :class:`_expression.ClauseElement` objects only need to use + those symbols present in :class:`.InternalTraversal`. However, for + additional caching use cases within the ORM, symbols dealing with the + :class:`.HasCacheKey` class are added here. + + """ + + dp_ignore = "IG" + """Specify an object that should be ignored entirely. + + This currently applies function call argument caching where some + arguments should not be considered to be part of a cache key. + + """ + + dp_inspectable = "IS" + """Visit an inspectable object where the return value is a + :class:`.HasCacheKey` object.""" + + dp_multi = "M" + """Visit an object that may be a :class:`.HasCacheKey` or may be a + plain hashable object.""" + + dp_multi_list = "MT" + """Visit a tuple containing elements that may be :class:`.HasCacheKey` or + may be a plain hashable object.""" + + dp_has_cache_key_tuples = "HT" + """Visit a list of tuples which contain :class:`.HasCacheKey` + objects. + + """ + + dp_inspectable_list = "IL" + """Visit a list of inspectable objects which upon inspection are + HasCacheKey objects.""" + + +_TraverseInternalsType = List[Tuple[str, InternalTraversal]] +"""a structure that defines how a HasTraverseInternals should be +traversed. + +This structure consists of a list of (attributename, internaltraversal) +tuples, where the "attributename" refers to the name of an attribute on an +instance of the HasTraverseInternals object, and "internaltraversal" refers +to an :class:`.InternalTraversal` enumeration symbol defining what kind +of data this attribute stores, which indicates to the traverser how it should +be handled. + +""" + + +class HasTraverseInternals: + """base for classes that have a "traverse internals" element, + which defines all kinds of ways of traversing the elements of an object. + + Compared to :class:`.Visitable`, which relies upon an external visitor to + define how the object is traversed (i.e. the :class:`.SQLCompiler`), the + :class:`.HasTraverseInternals` interface allows classes to define their own + traversal, that is, what attributes are accessed and in what order. + + """ + + __slots__ = () + + _traverse_internals: _TraverseInternalsType + + _is_immutable: bool = False + + @util.preload_module("sqlalchemy.sql.traversals") + def get_children( + self, *, omit_attrs: Tuple[str, ...] = (), **kw: Any + ) -> Iterable[HasTraverseInternals]: + r"""Return immediate child :class:`.visitors.HasTraverseInternals` + elements of this :class:`.visitors.HasTraverseInternals`. + + This is used for visit traversal. + + \**kw may contain flags that change the collection that is + returned, for example to return a subset of items in order to + cut down on larger traversals, or to return child items from a + different context (such as schema-level collections instead of + clause-level). + + """ + + traversals = util.preloaded.sql_traversals + + try: + traverse_internals = self._traverse_internals + except AttributeError: + # user-defined classes may not have a _traverse_internals + return [] + + dispatch = traversals._get_children.run_generated_dispatch + return itertools.chain.from_iterable( + meth(obj, **kw) + for attrname, obj, meth in dispatch( + self, traverse_internals, "_generated_get_children_traversal" + ) + if attrname not in omit_attrs and obj is not None + ) + + +class _InternalTraversalDispatchType(Protocol): + def __call__(s, self: object, visitor: HasTraversalDispatch) -> Any: ... + + +class HasTraversalDispatch: + r"""Define infrastructure for classes that perform internal traversals + + .. versionadded:: 2.0 + + """ + + __slots__ = () + + _dispatch_lookup: ClassVar[Dict[Union[InternalTraversal, str], str]] = {} + + def dispatch(self, visit_symbol: InternalTraversal) -> Callable[..., Any]: + """Given a method from :class:`.HasTraversalDispatch`, return the + corresponding method on a subclass. + + """ + name = _dispatch_lookup[visit_symbol] + return getattr(self, name, None) # type: ignore + + def run_generated_dispatch( + self, + target: object, + internal_dispatch: _TraverseInternalsType, + generate_dispatcher_name: str, + ) -> Any: + dispatcher: _InternalTraversalDispatchType + try: + dispatcher = target.__class__.__dict__[generate_dispatcher_name] + 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 = self.generate_dispatch( + target.__class__, internal_dispatch, generate_dispatcher_name + ) + return dispatcher(target, self) + + def generate_dispatch( + self, + target_cls: Type[object], + internal_dispatch: _TraverseInternalsType, + generate_dispatcher_name: str, + ) -> _InternalTraversalDispatchType: + dispatcher = self._generate_dispatcher( + internal_dispatch, generate_dispatcher_name + ) + # assert isinstance(target_cls, type) + setattr(target_cls, generate_dispatcher_name, dispatcher) + return dispatcher + + def _generate_dispatcher( + self, internal_dispatch: _TraverseInternalsType, method_name: str + ) -> _InternalTraversalDispatchType: + names = [] + for attrname, visit_sym in internal_dispatch: + meth = self.dispatch(visit_sym) + if meth is not None: + visit_name = _dispatch_lookup[visit_sym] + names.append((attrname, visit_name)) + + code = ( + (" return [\n") + + ( + ", \n".join( + " (%r, self.%s, visitor.%s)" + % (attrname, attrname, visit_name) + for attrname, visit_name in names + ) + ) + + ("\n ]\n") + ) + meth_text = ("def %s(self, visitor):\n" % method_name) + code + "\n" + return cast( + _InternalTraversalDispatchType, + langhelpers._exec_code_in_env(meth_text, {}, method_name), + ) + + +ExtendedInternalTraversal = InternalTraversal + + +def _generate_traversal_dispatch() -> None: + lookup = _dispatch_lookup + + for sym in InternalTraversal: + key = sym.name + if key.startswith("dp_"): + visit_key = key.replace("dp_", "visit_") + sym_name = sym.value + assert sym_name not in lookup, sym_name + lookup[sym] = lookup[sym_name] = visit_key + + +_dispatch_lookup = HasTraversalDispatch._dispatch_lookup +_generate_traversal_dispatch() + + +class ExternallyTraversible(HasTraverseInternals, Visitable): + __slots__ = () + + _annotations: Mapping[Any, Any] = util.EMPTY_DICT + + if typing.TYPE_CHECKING: + + def _annotate(self, values: _AnnotationDict) -> Self: ... + + def get_children( + self, *, omit_attrs: Tuple[str, ...] = (), **kw: Any + ) -> Iterable[ExternallyTraversible]: ... + + def _clone(self, **kw: Any) -> Self: + """clone this element""" + raise NotImplementedError() + + def _copy_internals( + self, *, omit_attrs: Tuple[str, ...] = (), **kw: Any + ) -> None: + """Reassign internal elements to be clones of themselves. + + Called during a copy-and-traverse operation on newly + shallow-copied elements to create a deep copy. + + The given clone function should be used, which may be applying + additional transformations to the element (i.e. replacement + traversal, cloned traversal, annotations). + + """ + raise NotImplementedError() + + +_ET = TypeVar("_ET", bound=ExternallyTraversible) + +_CE = TypeVar("_CE", bound="ColumnElement[Any]") + +_TraverseCallableType = Callable[[_ET], None] + + +class _CloneCallableType(Protocol): + def __call__(self, element: _ET, **kw: Any) -> _ET: ... + + +class _TraverseTransformCallableType(Protocol[_ET]): + def __call__(self, element: _ET, **kw: Any) -> Optional[_ET]: ... + + +_ExtT = TypeVar("_ExtT", bound="ExternalTraversal") + + +class ExternalTraversal(util.MemoizedSlots): + """Base class for visitor objects which can traverse externally using + the :func:`.visitors.traverse` function. + + Direct usage of the :func:`.visitors.traverse` function is usually + preferred. + + """ + + __slots__ = ("_visitor_dict", "_next") + + __traverse_options__: Dict[str, Any] = {} + _next: Optional[ExternalTraversal] + + def traverse_single(self, obj: Visitable, **kw: Any) -> Any: + for v in self.visitor_iterator: + meth = getattr(v, "visit_%s" % obj.__visit_name__, None) + if meth: + return meth(obj, **kw) + + def iterate( + self, obj: Optional[ExternallyTraversible] + ) -> Iterator[ExternallyTraversible]: + """Traverse the given expression structure, returning an iterator + of all elements. + + """ + return iterate(obj, self.__traverse_options__) + + @overload + def traverse(self, obj: Literal[None]) -> None: ... + + @overload + def traverse( + self, obj: ExternallyTraversible + ) -> ExternallyTraversible: ... + + def traverse( + self, obj: Optional[ExternallyTraversible] + ) -> Optional[ExternallyTraversible]: + """Traverse and visit the given expression structure.""" + + return traverse(obj, self.__traverse_options__, self._visitor_dict) + + def _memoized_attr__visitor_dict( + self, + ) -> Dict[str, _TraverseCallableType[Any]]: + visitors = {} + + for name in dir(self): + if name.startswith("visit_"): + visitors[name[6:]] = getattr(self, name) + return visitors + + @property + def visitor_iterator(self) -> Iterator[ExternalTraversal]: + """Iterate through this visitor and each 'chained' visitor.""" + + v: Optional[ExternalTraversal] = self + while v: + yield v + v = getattr(v, "_next", None) + + def chain(self: _ExtT, visitor: ExternalTraversal) -> _ExtT: + """'Chain' an additional ExternalTraversal onto this ExternalTraversal + + The chained visitor will receive all visit events after this one. + + """ + tail = list(self.visitor_iterator)[-1] + tail._next = visitor + return self + + +class CloningExternalTraversal(ExternalTraversal): + """Base class for visitor objects which can traverse using + the :func:`.visitors.cloned_traverse` function. + + Direct usage of the :func:`.visitors.cloned_traverse` function is usually + preferred. + + + """ + + __slots__ = () + + def copy_and_process( + self, list_: List[ExternallyTraversible] + ) -> List[ExternallyTraversible]: + """Apply cloned traversal to the given list of elements, and return + the new list. + + """ + return [self.traverse(x) for x in list_] + + @overload + def traverse(self, obj: Literal[None]) -> None: ... + + @overload + def traverse( + self, obj: ExternallyTraversible + ) -> ExternallyTraversible: ... + + def traverse( + self, obj: Optional[ExternallyTraversible] + ) -> Optional[ExternallyTraversible]: + """Traverse and visit the given expression structure.""" + + return cloned_traverse( + obj, self.__traverse_options__, self._visitor_dict + ) + + +class ReplacingExternalTraversal(CloningExternalTraversal): + """Base class for visitor objects which can traverse using + the :func:`.visitors.replacement_traverse` function. + + Direct usage of the :func:`.visitors.replacement_traverse` function is + usually preferred. + + """ + + __slots__ = () + + def replace( + self, elem: ExternallyTraversible + ) -> Optional[ExternallyTraversible]: + """Receive pre-copied elements during a cloning traversal. + + If the method returns a new element, the element is used + instead of creating a simple copy of the element. Traversal + will halt on the newly returned element if it is re-encountered. + """ + return None + + @overload + def traverse(self, obj: Literal[None]) -> None: ... + + @overload + def traverse( + self, obj: ExternallyTraversible + ) -> ExternallyTraversible: ... + + def traverse( + self, obj: Optional[ExternallyTraversible] + ) -> Optional[ExternallyTraversible]: + """Traverse and visit the given expression structure.""" + + def replace( + element: ExternallyTraversible, + **kw: Any, + ) -> Optional[ExternallyTraversible]: + for v in self.visitor_iterator: + e = cast(ReplacingExternalTraversal, v).replace(element) + if e is not None: + return e + + return None + + return replacement_traverse(obj, self.__traverse_options__, replace) + + +# backwards compatibility +Traversible = Visitable + +ClauseVisitor = ExternalTraversal +CloningVisitor = CloningExternalTraversal +ReplacingCloningVisitor = ReplacingExternalTraversal + + +def iterate( + obj: Optional[ExternallyTraversible], + opts: Mapping[str, Any] = util.EMPTY_DICT, +) -> Iterator[ExternallyTraversible]: + r"""Traverse the given expression structure, returning an iterator. + + Traversal is configured to be breadth-first. + + The central API feature used by the :func:`.visitors.iterate` + function is the + :meth:`_expression.ClauseElement.get_children` method of + :class:`_expression.ClauseElement` objects. This method should return all + the :class:`_expression.ClauseElement` objects which are associated with a + particular :class:`_expression.ClauseElement` object. For example, a + :class:`.Case` structure will refer to a series of + :class:`_expression.ColumnElement` objects within its "whens" and "else\_" + member variables. + + :param obj: :class:`_expression.ClauseElement` structure to be traversed + + :param opts: dictionary of iteration options. This dictionary is usually + empty in modern usage. + + """ + if obj is None: + return + + yield obj + children = obj.get_children(**opts) + + if not children: + return + + stack = deque([children]) + while stack: + t_iterator = stack.popleft() + for t in t_iterator: + yield t + stack.append(t.get_children(**opts)) + + +@overload +def traverse_using( + iterator: Iterable[ExternallyTraversible], + obj: Literal[None], + visitors: Mapping[str, _TraverseCallableType[Any]], +) -> None: ... + + +@overload +def traverse_using( + iterator: Iterable[ExternallyTraversible], + obj: ExternallyTraversible, + visitors: Mapping[str, _TraverseCallableType[Any]], +) -> ExternallyTraversible: ... + + +def traverse_using( + iterator: Iterable[ExternallyTraversible], + obj: Optional[ExternallyTraversible], + visitors: Mapping[str, _TraverseCallableType[Any]], +) -> Optional[ExternallyTraversible]: + """Visit the given expression structure using the given iterator of + objects. + + :func:`.visitors.traverse_using` is usually called internally as the result + of the :func:`.visitors.traverse` function. + + :param iterator: an iterable or sequence which will yield + :class:`_expression.ClauseElement` + structures; the iterator is assumed to be the + product of the :func:`.visitors.iterate` function. + + :param obj: the :class:`_expression.ClauseElement` + that was used as the target of the + :func:`.iterate` function. + + :param visitors: dictionary of visit functions. See :func:`.traverse` + for details on this dictionary. + + .. seealso:: + + :func:`.traverse` + + + """ + for target in iterator: + meth = visitors.get(target.__visit_name__, None) + if meth: + meth(target) + return obj + + +@overload +def traverse( + obj: Literal[None], + opts: Mapping[str, Any], + visitors: Mapping[str, _TraverseCallableType[Any]], +) -> None: ... + + +@overload +def traverse( + obj: ExternallyTraversible, + opts: Mapping[str, Any], + visitors: Mapping[str, _TraverseCallableType[Any]], +) -> ExternallyTraversible: ... + + +def traverse( + obj: Optional[ExternallyTraversible], + opts: Mapping[str, Any], + visitors: Mapping[str, _TraverseCallableType[Any]], +) -> Optional[ExternallyTraversible]: + """Traverse and visit the given expression structure using the default + iterator. + + e.g.:: + + from sqlalchemy.sql import visitors + + stmt = select(some_table).where(some_table.c.foo == "bar") + + + def visit_bindparam(bind_param): + print("found bound value: %s" % bind_param.value) + + + visitors.traverse(stmt, {}, {"bindparam": visit_bindparam}) + + The iteration of objects uses the :func:`.visitors.iterate` function, + which does a breadth-first traversal using a stack. + + :param obj: :class:`_expression.ClauseElement` structure to be traversed + + :param opts: dictionary of iteration options. This dictionary is usually + empty in modern usage. + + :param visitors: dictionary of visit functions. The dictionary should + have strings as keys, each of which would correspond to the + ``__visit_name__`` of a particular kind of SQL expression object, and + callable functions as values, each of which represents a visitor function + for that kind of object. + + """ + return traverse_using(iterate(obj, opts), obj, visitors) + + +@overload +def cloned_traverse( + obj: Literal[None], + opts: Mapping[str, Any], + visitors: Mapping[str, _TraverseCallableType[Any]], +) -> None: ... + + +# a bit of controversy here, as the clone of the lead element +# *could* in theory replace with an entirely different kind of element. +# however this is really not how cloned_traverse is ever used internally +# at least. +@overload +def cloned_traverse( + obj: _ET, + opts: Mapping[str, Any], + visitors: Mapping[str, _TraverseCallableType[Any]], +) -> _ET: ... + + +def cloned_traverse( + obj: Optional[ExternallyTraversible], + opts: Mapping[str, Any], + visitors: Mapping[str, _TraverseCallableType[Any]], +) -> Optional[ExternallyTraversible]: + """Clone the given expression structure, allowing modifications by + visitors for mutable objects. + + Traversal usage is the same as that of :func:`.visitors.traverse`. + The visitor functions present in the ``visitors`` dictionary may also + modify the internals of the given structure as the traversal proceeds. + + The :func:`.cloned_traverse` function does **not** provide objects that are + part of the :class:`.Immutable` interface to the visit methods (this + primarily includes :class:`.ColumnClause`, :class:`.Column`, + :class:`.TableClause` and :class:`.Table` objects). As this traversal is + only intended to allow in-place mutation of objects, :class:`.Immutable` + objects are skipped. The :meth:`.Immutable._clone` method is still called + on each object to allow for objects to replace themselves with a different + object based on a clone of their sub-internals (e.g. a + :class:`.ColumnClause` that clones its subquery to return a new + :class:`.ColumnClause`). + + .. versionchanged:: 2.0 The :func:`.cloned_traverse` function omits + objects that are part of the :class:`.Immutable` interface. + + The central API feature used by the :func:`.visitors.cloned_traverse` + and :func:`.visitors.replacement_traverse` functions, in addition to the + :meth:`_expression.ClauseElement.get_children` + function that is used to achieve + the iteration, is the :meth:`_expression.ClauseElement._copy_internals` + method. + For a :class:`_expression.ClauseElement` + structure to support cloning and replacement + traversals correctly, it needs to be able to pass a cloning function into + its internal members in order to make copies of them. + + .. seealso:: + + :func:`.visitors.traverse` + + :func:`.visitors.replacement_traverse` + + """ + + cloned: Dict[int, ExternallyTraversible] = {} + stop_on = set(opts.get("stop_on", [])) + + def deferred_copy_internals( + obj: ExternallyTraversible, + ) -> ExternallyTraversible: + return cloned_traverse(obj, opts, visitors) + + def clone(elem: ExternallyTraversible, **kw: Any) -> ExternallyTraversible: + if elem in stop_on: + return elem + else: + if id(elem) not in cloned: + if "replace" in kw: + newelem = cast( + Optional[ExternallyTraversible], kw["replace"](elem) + ) + if newelem is not None: + cloned[id(elem)] = newelem + return newelem + + # the _clone method for immutable normally returns "self". + # however, the method is still allowed to return a + # different object altogether; ColumnClause._clone() will + # based on options clone the subquery to which it is associated + # and return the new corresponding column. + cloned[id(elem)] = newelem = elem._clone(clone=clone, **kw) + newelem._copy_internals(clone=clone, **kw) + + # however, visit methods which are tasked with in-place + # mutation of the object should not get access to the immutable + # object. + if not elem._is_immutable: + meth = visitors.get(newelem.__visit_name__, None) + if meth: + meth(newelem) + return cloned[id(elem)] + + if obj is not None: + obj = clone( + obj, deferred_copy_internals=deferred_copy_internals, **opts + ) + clone = None # type: ignore[assignment] # remove gc cycles + return obj + + +@overload +def replacement_traverse( + obj: Literal[None], + opts: Mapping[str, Any], + replace: _TraverseTransformCallableType[Any], +) -> None: ... + + +@overload +def replacement_traverse( + obj: _CE, + opts: Mapping[str, Any], + replace: _TraverseTransformCallableType[Any], +) -> _CE: ... + + +@overload +def replacement_traverse( + obj: ExternallyTraversible, + opts: Mapping[str, Any], + replace: _TraverseTransformCallableType[Any], +) -> ExternallyTraversible: ... + + +def replacement_traverse( + obj: Optional[ExternallyTraversible], + opts: Mapping[str, Any], + replace: _TraverseTransformCallableType[Any], +) -> Optional[ExternallyTraversible]: + """Clone the given expression structure, allowing element + replacement by a given replacement function. + + This function is very similar to the :func:`.visitors.cloned_traverse` + function, except instead of being passed a dictionary of visitors, all + elements are unconditionally passed into the given replace function. + The replace function then has the option to return an entirely new object + which will replace the one given. If it returns ``None``, then the object + is kept in place. + + The difference in usage between :func:`.visitors.cloned_traverse` and + :func:`.visitors.replacement_traverse` is that in the former case, an + already-cloned object is passed to the visitor function, and the visitor + function can then manipulate the internal state of the object. + In the case of the latter, the visitor function should only return an + entirely different object, or do nothing. + + The use case for :func:`.visitors.replacement_traverse` is that of + replacing a FROM clause inside of a SQL structure with a different one, + as is a common use case within the ORM. + + """ + + cloned = {} + stop_on = {id(x) for x in opts.get("stop_on", [])} + + def deferred_copy_internals( + obj: ExternallyTraversible, + ) -> ExternallyTraversible: + return replacement_traverse(obj, opts, replace) + + def clone(elem: ExternallyTraversible, **kw: Any) -> ExternallyTraversible: + if ( + id(elem) in stop_on + or "no_replacement_traverse" in elem._annotations + ): + return elem + else: + newelem = replace(elem) + if newelem is not None: + stop_on.add(id(newelem)) + return newelem # type: ignore + else: + # base "already seen" on id(), not hash, so that we don't + # replace an Annotated element with its non-annotated one, and + # vice versa + id_elem = id(elem) + if id_elem not in cloned: + if "replace" in kw: + newelem = kw["replace"](elem) + if newelem is not None: + cloned[id_elem] = newelem + return newelem # type: ignore + + cloned[id_elem] = newelem = elem._clone(**kw) + newelem._copy_internals(clone=clone, **kw) + return cloned[id_elem] # type: ignore + + if obj is not None: + obj = clone( + obj, deferred_copy_internals=deferred_copy_internals, **opts + ) + clone = None # type: ignore[assignment] # remove gc cycles + return obj