summaryrefslogtreecommitdiff
path: root/lib/sqlalchemy/sql/traversals.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2022-01-18 17:00:16 -0500
committerMike Bayer <mike_mp@zzzcomputing.com>2022-01-21 11:46:51 -0500
commitd46a4c0326bd2e697794514b920e6727d5153324 (patch)
treeb3368bc6d402148d46317b4532db6b92352bd666 /lib/sqlalchemy/sql/traversals.py
parent7d9b811555a88dd2f1cb1520027546b87383e159 (diff)
downloadsqlalchemy-d46a4c0326bd2e697794514b920e6727d5153324.tar.gz
Add new infrastructure to support greater use of __slots__
* Changed AliasedInsp to use __slots__ * Migrated all of strategy_options to use __slots__ for objects. Adds new infrastructure to traversals to support shallow copy, to dict and from dict based on internal traversal attributes. Load / _LoadElement then leverage this to provide clone / generative / getstate without the need for __dict__ or explicit attribute lists. Doing this change revealed that there are lots of things that trigger off of whether or not a class has a __visit_name__ attribute. so to suit that we've gone back to having Visitable, which is a better name than Traversible at this point (I think Traversible is mis-spelled too). Change-Id: I13d04e494339fac9dbda0b8e78153418abebaf72 References: #7527
Diffstat (limited to 'lib/sqlalchemy/sql/traversals.py')
-rw-r--r--lib/sqlalchemy/sql/traversals.py209
1 files changed, 206 insertions, 3 deletions
diff --git a/lib/sqlalchemy/sql/traversals.py b/lib/sqlalchemy/sql/traversals.py
index 2fa3a0408..18fd1d4b8 100644
--- a/lib/sqlalchemy/sql/traversals.py
+++ b/lib/sqlalchemy/sql/traversals.py
@@ -10,12 +10,22 @@ import collections.abc as collections_abc
import itertools
from itertools import zip_longest
import operator
+import typing
+from typing import Any
+from typing import Callable
+from typing import Dict
+from typing import Type
+from typing import TypeVar
from . import operators
+from .cache_key import HasCacheKey
+from .visitors import _TraverseInternalsType
from .visitors import anon_map
+from .visitors import ExtendedInternalTraversal
+from .visitors import HasTraverseInternals
from .visitors import InternalTraversal
from .. import util
-
+from ..util import langhelpers
SKIP_TRAVERSE = util.symbol("skip_traverse")
COMPARE_FAILED = False
@@ -47,11 +57,158 @@ def _preconfigure_traversals(target_hierarchy):
)
+SelfHasShallowCopy = TypeVar("SelfHasShallowCopy", bound="HasShallowCopy")
+
+
+class HasShallowCopy(HasTraverseInternals):
+ """attribute-wide operations that are useful for classes that use
+ __slots__ and therefore can't operate on their attributes in a dictionary.
+
+
+ """
+
+ __slots__ = ()
+
+ if typing.TYPE_CHECKING:
+
+ def _generated_shallow_copy_traversal(
+ self: SelfHasShallowCopy, other: SelfHasShallowCopy
+ ) -> None:
+ ...
+
+ def _generated_shallow_from_dict_traversal(
+ self, d: Dict[str, Any]
+ ) -> None:
+ ...
+
+ def _generated_shallow_to_dict_traversal(self) -> Dict[str, Any]:
+ ...
+
+ @classmethod
+ def _generate_shallow_copy(
+ cls: Type[SelfHasShallowCopy],
+ internal_dispatch: _TraverseInternalsType,
+ method_name: str,
+ ) -> Callable[[SelfHasShallowCopy, SelfHasShallowCopy], None]:
+ code = "\n".join(
+ f" other.{attrname} = self.{attrname}"
+ for attrname, _ in internal_dispatch
+ )
+ meth_text = f"def {method_name}(self, other):\n{code}\n"
+ return langhelpers._exec_code_in_env(meth_text, {}, method_name)
+
+ @classmethod
+ def _generate_shallow_to_dict(
+ cls: Type[SelfHasShallowCopy],
+ internal_dispatch: _TraverseInternalsType,
+ method_name: str,
+ ) -> Callable[[SelfHasShallowCopy], Dict[str, Any]]:
+ code = ",\n".join(
+ f" '{attrname}': self.{attrname}"
+ for attrname, _ in internal_dispatch
+ )
+ meth_text = f"def {method_name}(self):\n return {{{code}}}\n"
+ return langhelpers._exec_code_in_env(meth_text, {}, method_name)
+
+ @classmethod
+ def _generate_shallow_from_dict(
+ cls: Type[SelfHasShallowCopy],
+ internal_dispatch: _TraverseInternalsType,
+ method_name: str,
+ ) -> Callable[[SelfHasShallowCopy, Dict[str, Any]], None]:
+ code = "\n".join(
+ f" self.{attrname} = d['{attrname}']"
+ for attrname, _ in internal_dispatch
+ )
+ meth_text = f"def {method_name}(self, d):\n{code}\n"
+ return langhelpers._exec_code_in_env(meth_text, {}, method_name)
+
+ def _shallow_from_dict(self, d: Dict) -> None:
+ cls = self.__class__
+
+ try:
+ shallow_from_dict = cls.__dict__[
+ "_generated_shallow_from_dict_traversal"
+ ]
+ except KeyError:
+ shallow_from_dict = (
+ cls._generated_shallow_from_dict_traversal # type: ignore
+ ) = self._generate_shallow_from_dict(
+ cls._traverse_internals,
+ "_generated_shallow_from_dict_traversal",
+ )
+
+ shallow_from_dict(self, d)
+
+ def _shallow_to_dict(self) -> Dict[str, Any]:
+ cls = self.__class__
+
+ try:
+ shallow_to_dict = cls.__dict__[
+ "_generated_shallow_to_dict_traversal"
+ ]
+ except KeyError:
+ shallow_to_dict = (
+ cls._generated_shallow_to_dict_traversal # type: ignore
+ ) = self._generate_shallow_to_dict(
+ cls._traverse_internals, "_generated_shallow_to_dict_traversal"
+ )
+
+ return shallow_to_dict(self)
+
+ def _shallow_copy_to(self: SelfHasShallowCopy, other: SelfHasShallowCopy):
+ cls = self.__class__
+
+ try:
+ shallow_copy = cls.__dict__["_generated_shallow_copy_traversal"]
+ except KeyError:
+ shallow_copy = (
+ cls._generated_shallow_copy_traversal # type: ignore
+ ) = self._generate_shallow_copy(
+ cls._traverse_internals, "_generated_shallow_copy_traversal"
+ )
+
+ shallow_copy(self, other)
+
+ def _clone(self: SelfHasShallowCopy, **kw) -> SelfHasShallowCopy:
+ """Create a shallow copy"""
+ c = self.__class__.__new__(self.__class__)
+ self._shallow_copy_to(c)
+ return c
+
+
+SelfGenerativeOnTraversal = TypeVar(
+ "SelfGenerativeOnTraversal", bound="GenerativeOnTraversal"
+)
+
+
+class GenerativeOnTraversal(HasShallowCopy):
+ """Supplies Generative behavior but making use of traversals to shallow
+ copy.
+
+ .. seealso::
+
+ :class:`sqlalchemy.sql.base.Generative`
+
+
+ """
+
+ __slots__ = ()
+
+ def _generate(
+ self: SelfGenerativeOnTraversal,
+ ) -> SelfGenerativeOnTraversal:
+ cls = self.__class__
+ s = cls.__new__(cls)
+ self._shallow_copy_to(s)
+ return s
+
+
def _clone(element, **kw):
return element._clone()
-class HasCopyInternals:
+class HasCopyInternals(HasTraverseInternals):
__slots__ = ()
def _clone(self, **kw):
@@ -304,7 +461,9 @@ def _resolve_name_for_compare(element, name, anon_map, **kw):
return name
-class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
+class TraversalComparatorStrategy(
+ ExtendedInternalTraversal, util.MemoizedSlots
+):
__slots__ = "stack", "cache", "anon_map"
def __init__(self):
@@ -377,6 +536,10 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
continue
dispatch = self.dispatch(left_visit_sym)
+ assert dispatch, (
+ f"{self.__class__} has no dispatch for "
+ f"'{self._dispatch_lookup[left_visit_sym]}'"
+ )
left_child = operator.attrgetter(left_attrname)(left)
right_child = operator.attrgetter(right_attrname)(right)
if left_child is None:
@@ -517,6 +680,46 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
):
return left == right
+ def visit_string_multi_dict(
+ self, attrname, left_parent, left, right_parent, right, **kw
+ ):
+
+ for lk, rk in zip_longest(
+ sorted(left.keys()), sorted(right.keys()), fillvalue=(None, None)
+ ):
+ if lk != rk:
+ return COMPARE_FAILED
+
+ lv, rv = left[lk], right[rk]
+
+ lhc = isinstance(left, HasCacheKey)
+ rhc = isinstance(right, HasCacheKey)
+ if lhc and rhc:
+ if lv._gen_cache_key(
+ self.anon_map[0], []
+ ) != rv._gen_cache_key(self.anon_map[1], []):
+ return COMPARE_FAILED
+ elif lhc != rhc:
+ return COMPARE_FAILED
+ elif lv != rv:
+ return COMPARE_FAILED
+
+ def visit_multi(
+ self, attrname, left_parent, left, right_parent, right, **kw
+ ):
+
+ lhc = isinstance(left, HasCacheKey)
+ rhc = isinstance(right, HasCacheKey)
+ if lhc and rhc:
+ if left._gen_cache_key(
+ self.anon_map[0], []
+ ) != right._gen_cache_key(self.anon_map[1], []):
+ return COMPARE_FAILED
+ elif lhc != rhc:
+ return COMPARE_FAILED
+ else:
+ return left == right
+
def visit_anon_name(
self, attrname, left_parent, left, right_parent, right, **kw
):