diff options
author | Mike Bayer <mike_mp@zzzcomputing.com> | 2022-03-08 17:14:41 -0500 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2022-03-13 15:29:20 -0400 |
commit | 769fa67d842035dd852ab8b6a26ea3f110a51131 (patch) | |
tree | 5c121caca336071091c6f5ea4c54743c92d6458a /lib/sqlalchemy/sql/traversals.py | |
parent | 77fc8216a74e6b2d0efc6591c6c735687bd10002 (diff) | |
download | sqlalchemy-769fa67d842035dd852ab8b6a26ea3f110a51131.tar.gz |
pep-484: sqlalchemy.sql pass one
sqlalchemy.sql will require many passes to get all
modules even gradually typed. Will have to pick and
choose what modules can be strictly typed vs. which
can be gradual.
in this patch, emphasis is on visitors.py, cache_key.py,
annotations.py for strict typing, compiler.py is on gradual
typing but has much more structure, in particular where it
connects with the outside world.
The work within compiler.py also reached back out to
engine/cursor.py , default.py quite a bit.
References: #6810
Change-Id: I6e8a29f6013fd216e43d45091bc193f8be0368fd
Diffstat (limited to 'lib/sqlalchemy/sql/traversals.py')
-rw-r--r-- | lib/sqlalchemy/sql/traversals.py | 64 |
1 files changed, 37 insertions, 27 deletions
diff --git a/lib/sqlalchemy/sql/traversals.py b/lib/sqlalchemy/sql/traversals.py index 4fa23d370..cf9487f93 100644 --- a/lib/sqlalchemy/sql/traversals.py +++ b/lib/sqlalchemy/sql/traversals.py @@ -15,7 +15,10 @@ import operator import typing from typing import Any from typing import Callable +from typing import Deque from typing import Dict +from typing import Set +from typing import Tuple from typing import Type from typing import TypeVar @@ -23,9 +26,9 @@ from . import operators from .cache_key import HasCacheKey from .visitors import _TraverseInternalsType from .visitors import anon_map -from .visitors import ExtendedInternalTraversal +from .visitors import ExternallyTraversible +from .visitors import HasTraversalDispatch from .visitors import HasTraverseInternals -from .visitors import InternalTraversal from .. import util from ..util import langhelpers @@ -35,6 +38,7 @@ COMPARE_SUCCEEDED = True def compare(obj1, obj2, **kw): + strategy: TraversalComparatorStrategy if kw.get("use_proxies", False): strategy = ColIdentityComparatorStrategy() else: @@ -45,16 +49,18 @@ def compare(obj1, obj2, **kw): def _preconfigure_traversals(target_hierarchy): for cls in util.walk_subclasses(target_hierarchy): - if hasattr(cls, "_traverse_internals"): - cls._generate_cache_attrs() + if hasattr(cls, "_generate_cache_attrs") and hasattr( + cls, "_traverse_internals" + ): + cls._generate_cache_attrs() # type: ignore _copy_internals.generate_dispatch( - cls, - cls._traverse_internals, + cls, # type: ignore + cls._traverse_internals, # type: ignore "_generated_copy_internals_traversal", ) _get_children.generate_dispatch( - cls, - cls._traverse_internals, + cls, # type: ignore + cls._traverse_internals, # type: ignore "_generated_get_children_traversal", ) @@ -125,54 +131,58 @@ class HasShallowCopy(HasTraverseInternals): 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: + def _shallow_from_dict(self, d: Dict[str, Any]) -> None: cls = self.__class__ + shallow_from_dict: Callable[[HasShallowCopy, Dict[str, Any]], None] 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( + shallow_from_dict = self._generate_shallow_from_dict( cls._traverse_internals, "_generated_shallow_from_dict_traversal", ) + cls._generated_shallow_from_dict_traversal = shallow_from_dict # type: ignore # noqa E501 + shallow_from_dict(self, d) def _shallow_to_dict(self) -> Dict[str, Any]: cls = self.__class__ + shallow_to_dict: Callable[[HasShallowCopy], Dict[str, Any]] + 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( + shallow_to_dict = self._generate_shallow_to_dict( cls._traverse_internals, "_generated_shallow_to_dict_traversal" ) + cls._generated_shallow_to_dict_traversal = shallow_to_dict # type: ignore # noqa E501 return shallow_to_dict(self) - def _shallow_copy_to(self: SelfHasShallowCopy, other: SelfHasShallowCopy): + def _shallow_copy_to( + self: SelfHasShallowCopy, other: SelfHasShallowCopy + ) -> None: cls = self.__class__ + shallow_copy: Callable[[SelfHasShallowCopy, SelfHasShallowCopy], None] try: shallow_copy = cls.__dict__["_generated_shallow_copy_traversal"] except KeyError: - shallow_copy = ( - cls._generated_shallow_copy_traversal # type: ignore - ) = self._generate_shallow_copy( + shallow_copy = self._generate_shallow_copy( cls._traverse_internals, "_generated_shallow_copy_traversal" ) + cls._generated_shallow_copy_traversal = shallow_copy # type: ignore # noqa: E501 shallow_copy(self, other) - def _clone(self: SelfHasShallowCopy, **kw) -> SelfHasShallowCopy: + def _clone(self: SelfHasShallowCopy, **kw: Any) -> SelfHasShallowCopy: """Create a shallow copy""" c = self.__class__.__new__(self.__class__) self._shallow_copy_to(c) @@ -246,7 +256,7 @@ class HasCopyInternals(HasTraverseInternals): setattr(self, attrname, result) -class _CopyInternalsTraversal(InternalTraversal): +class _CopyInternalsTraversal(HasTraversalDispatch): """Generate a _copy_internals internal traversal dispatch for classes with a _traverse_internals collection.""" @@ -381,7 +391,7 @@ def _flatten_clauseelement(element): return element -class _GetChildrenTraversal(InternalTraversal): +class _GetChildrenTraversal(HasTraversalDispatch): """Generate a _children_traversal internal traversal dispatch for classes with a _traverse_internals collection.""" @@ -463,13 +473,13 @@ def _resolve_name_for_compare(element, name, anon_map, **kw): return name -class TraversalComparatorStrategy( - ExtendedInternalTraversal, util.MemoizedSlots -): +class TraversalComparatorStrategy(HasTraversalDispatch, util.MemoizedSlots): __slots__ = "stack", "cache", "anon_map" def __init__(self): - self.stack = deque() + self.stack: Deque[ + Tuple[ExternallyTraversible, ExternallyTraversible] + ] = deque() self.cache = set() def _memoized_attr_anon_map(self): @@ -653,7 +663,7 @@ class TraversalComparatorStrategy( if seq1 is None: return seq2 is None - completed = set() + completed: Set[object] = set() for clause in seq1: for other_clause in set(seq2).difference(completed): if self.compare_inner(clause, other_clause, **kw): |