Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# util/typing.py
2# Copyright (C) 2022-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: allow-untyped-defs, allow-untyped-calls
9from __future__ import annotations
11import builtins
12from collections import deque
13import collections.abc as collections_abc
14import re
15import sys
16import typing
17from typing import Any
18from typing import Callable
19from typing import Dict
20from typing import ForwardRef
21from typing import Generic
22from typing import Iterable
23from typing import Mapping
24from typing import NewType
25from typing import NoReturn
26from typing import Optional
27from typing import overload
28from typing import Set
29from typing import Tuple
30from typing import Type
31from typing import TYPE_CHECKING
32from typing import TypeVar
33from typing import Union
35import typing_extensions
37from . import compat
39if True: # zimports removes the tailing comments
40 from typing_extensions import Annotated as Annotated # 3.8
41 from typing_extensions import Concatenate as Concatenate # 3.10
42 from typing_extensions import (
43 dataclass_transform as dataclass_transform, # 3.11,
44 )
45 from typing_extensions import Final as Final # 3.8
46 from typing_extensions import final as final # 3.8
47 from typing_extensions import get_args as get_args # 3.10
48 from typing_extensions import get_origin as get_origin # 3.10
49 from typing_extensions import Literal as Literal # 3.8
50 from typing_extensions import NotRequired as NotRequired # 3.11
51 from typing_extensions import ParamSpec as ParamSpec # 3.10
52 from typing_extensions import Protocol as Protocol # 3.8
53 from typing_extensions import SupportsIndex as SupportsIndex # 3.8
54 from typing_extensions import TypeAlias as TypeAlias # 3.10
55 from typing_extensions import TypedDict as TypedDict # 3.8
56 from typing_extensions import TypeGuard as TypeGuard # 3.10
57 from typing_extensions import Self as Self # 3.11
58 from typing_extensions import TypeAliasType as TypeAliasType # 3.12
59 from typing_extensions import Never as Never # 3.11
60 from typing_extensions import LiteralString as LiteralString # 3.11
62_T = TypeVar("_T", bound=Any)
63_KT = TypeVar("_KT")
64_KT_co = TypeVar("_KT_co", covariant=True)
65_KT_contra = TypeVar("_KT_contra", contravariant=True)
66_VT = TypeVar("_VT")
67_VT_co = TypeVar("_VT_co", covariant=True)
69if compat.py310:
70 # why they took until py310 to put this in stdlib is beyond me,
71 # I've been wanting it since py27
72 from types import NoneType as NoneType
73else:
74 NoneType = type(None) # type: ignore
77def is_fwd_none(typ: Any) -> bool:
78 return isinstance(typ, ForwardRef) and typ.__forward_arg__ == "None"
81_AnnotationScanType = Union[
82 Type[Any], str, ForwardRef, NewType, TypeAliasType, "GenericProtocol[Any]"
83]
86class ArgsTypeProtocol(Protocol):
87 """protocol for types that have ``__args__``
89 there's no public interface for this AFAIK
91 """
93 __args__: Tuple[_AnnotationScanType, ...]
96class GenericProtocol(Protocol[_T]):
97 """protocol for generic types.
99 this since Python.typing _GenericAlias is private
101 """
103 __args__: Tuple[_AnnotationScanType, ...]
104 __origin__: Type[_T]
106 # Python's builtin _GenericAlias has this method, however builtins like
107 # list, dict, etc. do not, even though they have ``__origin__`` and
108 # ``__args__``
109 #
110 # def copy_with(self, params: Tuple[_AnnotationScanType, ...]) -> Type[_T]:
111 # ...
114# copied from TypeShed, required in order to implement
115# MutableMapping.update()
116class SupportsKeysAndGetItem(Protocol[_KT, _VT_co]):
117 def keys(self) -> Iterable[_KT]: ...
119 def __getitem__(self, __k: _KT) -> _VT_co: ...
122# work around https://github.com/microsoft/pyright/issues/3025
123_LiteralStar = Literal["*"]
126def de_stringify_annotation(
127 cls: Type[Any],
128 annotation: _AnnotationScanType,
129 originating_module: str,
130 locals_: Mapping[str, Any],
131 *,
132 str_cleanup_fn: Optional[Callable[[str, str], str]] = None,
133 include_generic: bool = False,
134 _already_seen: Optional[Set[Any]] = None,
135) -> Type[Any]:
136 """Resolve annotations that may be string based into real objects.
138 This is particularly important if a module defines "from __future__ import
139 annotations", as everything inside of __annotations__ is a string. We want
140 to at least have generic containers like ``Mapped``, ``Union``, ``List``,
141 etc.
143 """
144 # looked at typing.get_type_hints(), looked at pydantic. We need much
145 # less here, and we here try to not use any private typing internals
146 # or construct ForwardRef objects which is documented as something
147 # that should be avoided.
149 original_annotation = annotation
151 if is_fwd_ref(annotation):
152 annotation = annotation.__forward_arg__
154 if isinstance(annotation, str):
155 if str_cleanup_fn:
156 annotation = str_cleanup_fn(annotation, originating_module)
158 annotation = eval_expression(
159 annotation, originating_module, locals_=locals_, in_class=cls
160 )
162 if (
163 include_generic
164 and is_generic(annotation)
165 and not is_literal(annotation)
166 ):
167 if _already_seen is None:
168 _already_seen = set()
170 if annotation in _already_seen:
171 # only occurs recursively. outermost return type
172 # will always be Type.
173 # the element here will be either ForwardRef or
174 # Optional[ForwardRef]
175 return original_annotation # type: ignore
176 else:
177 _already_seen.add(annotation)
179 elements = tuple(
180 de_stringify_annotation(
181 cls,
182 elem,
183 originating_module,
184 locals_,
185 str_cleanup_fn=str_cleanup_fn,
186 include_generic=include_generic,
187 _already_seen=_already_seen,
188 )
189 for elem in annotation.__args__
190 )
192 return _copy_generic_annotation_with(annotation, elements)
194 return annotation # type: ignore
197def fixup_container_fwd_refs(
198 type_: _AnnotationScanType,
199) -> _AnnotationScanType:
200 """Correct dict['x', 'y'] into dict[ForwardRef('x'), ForwardRef('y')]
201 and similar for list, set
203 """
205 if (
206 is_generic(type_)
207 and get_origin(type_)
208 in (
209 dict,
210 set,
211 list,
212 collections_abc.MutableSet,
213 collections_abc.MutableMapping,
214 collections_abc.MutableSequence,
215 collections_abc.Mapping,
216 collections_abc.Sequence,
217 )
218 # fight, kick and scream to struggle to tell the difference between
219 # dict[] and typing.Dict[] which DO NOT compare the same and DO NOT
220 # behave the same yet there is NO WAY to distinguish between which type
221 # it is using public attributes
222 and not re.match(
223 "typing.(?:Dict|List|Set|.*Mapping|.*Sequence|.*Set)", repr(type_)
224 )
225 ):
226 # compat with py3.10 and earlier
227 return get_origin(type_).__class_getitem__( # type: ignore
228 tuple(
229 [
230 ForwardRef(elem) if isinstance(elem, str) else elem
231 for elem in get_args(type_)
232 ]
233 )
234 )
235 return type_
238def _copy_generic_annotation_with(
239 annotation: GenericProtocol[_T], elements: Tuple[_AnnotationScanType, ...]
240) -> Type[_T]:
241 if hasattr(annotation, "copy_with"):
242 # List, Dict, etc. real generics
243 return annotation.copy_with(elements) # type: ignore
244 else:
245 # Python builtins list, dict, etc.
246 return annotation.__origin__[elements] # type: ignore
249def eval_expression(
250 expression: str,
251 module_name: str,
252 *,
253 locals_: Optional[Mapping[str, Any]] = None,
254 in_class: Optional[Type[Any]] = None,
255) -> Any:
256 try:
257 base_globals: Dict[str, Any] = sys.modules[module_name].__dict__
258 except KeyError as ke:
259 raise NameError(
260 f"Module {module_name} isn't present in sys.modules; can't "
261 f"evaluate expression {expression}"
262 ) from ke
264 try:
265 if in_class is not None:
266 cls_namespace = dict(in_class.__dict__)
267 cls_namespace.setdefault(in_class.__name__, in_class)
269 # see #10899. We want the locals/globals to take precedence
270 # over the class namespace in this context, even though this
271 # is not the usual way variables would resolve.
272 cls_namespace.update(base_globals)
274 annotation = eval(expression, cls_namespace, locals_)
275 else:
276 annotation = eval(expression, base_globals, locals_)
277 except Exception as err:
278 raise NameError(
279 f"Could not de-stringify annotation {expression!r}"
280 ) from err
281 else:
282 return annotation
285def eval_name_only(
286 name: str,
287 module_name: str,
288 *,
289 locals_: Optional[Mapping[str, Any]] = None,
290) -> Any:
291 if "." in name:
292 return eval_expression(name, module_name, locals_=locals_)
294 try:
295 base_globals: Dict[str, Any] = sys.modules[module_name].__dict__
296 except KeyError as ke:
297 raise NameError(
298 f"Module {module_name} isn't present in sys.modules; can't "
299 f"resolve name {name}"
300 ) from ke
302 # name only, just look in globals. eval() works perfectly fine here,
303 # however we are seeking to have this be faster, as this occurs for
304 # every Mapper[] keyword, etc. depending on configuration
305 try:
306 return base_globals[name]
307 except KeyError as ke:
308 # check in builtins as well to handle `list`, `set` or `dict`, etc.
309 try:
310 return builtins.__dict__[name]
311 except KeyError:
312 pass
314 raise NameError(
315 f"Could not locate name {name} in module {module_name}"
316 ) from ke
319def resolve_name_to_real_class_name(name: str, module_name: str) -> str:
320 try:
321 obj = eval_name_only(name, module_name)
322 except NameError:
323 return name
324 else:
325 return getattr(obj, "__name__", name)
328def is_pep593(type_: Optional[Any]) -> bool:
329 return type_ is not None and get_origin(type_) in _type_tuples.Annotated
332def is_non_string_iterable(obj: Any) -> TypeGuard[Iterable[Any]]:
333 return isinstance(obj, collections_abc.Iterable) and not isinstance(
334 obj, (str, bytes)
335 )
338def is_literal(type_: Any) -> bool:
339 return get_origin(type_) in _type_tuples.Literal
342def is_newtype(type_: Optional[_AnnotationScanType]) -> TypeGuard[NewType]:
343 return hasattr(type_, "__supertype__")
345 # doesn't work in 3.8, 3.7 as it passes a closure, not an
346 # object instance
347 # isinstance(type, type_instances.NewType)
350def is_generic(type_: _AnnotationScanType) -> TypeGuard[GenericProtocol[Any]]:
351 return hasattr(type_, "__args__") and hasattr(type_, "__origin__")
354def is_pep695(type_: _AnnotationScanType) -> TypeGuard[TypeAliasType]:
355 # NOTE: a generic TAT does not instance check as TypeAliasType outside of
356 # python 3.10. For sqlalchemy use cases it's fine to consider it a TAT
357 # though.
358 # NOTE: things seems to work also without this additional check
359 if is_generic(type_):
360 if is_pep593(type_):
361 return False
362 return is_pep695(type_.__origin__)
363 return isinstance(type_, _type_instances.TypeAliasType)
366def flatten_newtype(type_: NewType) -> Type[Any]:
367 super_type = type_.__supertype__
368 while is_newtype(super_type):
369 super_type = super_type.__supertype__
370 return super_type # type: ignore[return-value]
373def pep695_values(type_: _AnnotationScanType) -> Set[Any]:
374 """Extracts the value from a TypeAliasType, recursively exploring unions
375 and inner TypeAliasType to flatten them into a single set.
377 Forward references are not evaluated, so no recursive exploration happens
378 into them.
379 """
380 _seen = set()
382 def recursive_value(inner_type):
383 if inner_type in _seen:
384 # recursion are not supported (at least it's flagged as
385 # an error by pyright). Just avoid infinite loop
386 return inner_type
387 _seen.add(inner_type)
388 if not is_pep695(inner_type):
389 return inner_type
390 value = inner_type.__value__
391 if not is_union(value):
392 return value
393 return [recursive_value(t) for t in value.__args__]
395 res = recursive_value(type_)
396 if isinstance(res, list):
397 types = set()
398 stack = deque(res)
399 while stack:
400 t = stack.popleft()
401 if isinstance(t, list):
402 stack.extend(t)
403 else:
404 types.add(None if t is NoneType or is_fwd_none(t) else t)
405 return types
406 else:
407 return {res}
410def is_fwd_ref(
411 type_: _AnnotationScanType,
412 check_generic: bool = False,
413 check_for_plain_string: bool = False,
414) -> TypeGuard[ForwardRef]:
415 if check_for_plain_string and isinstance(type_, str):
416 return True
417 elif isinstance(type_, _type_instances.ForwardRef):
418 return True
419 elif check_generic and is_generic(type_):
420 return any(
421 is_fwd_ref(
422 arg, True, check_for_plain_string=check_for_plain_string
423 )
424 for arg in type_.__args__
425 )
426 else:
427 return False
430@overload
431def de_optionalize_union_types(type_: str) -> str: ...
434@overload
435def de_optionalize_union_types(type_: Type[Any]) -> Type[Any]: ...
438@overload
439def de_optionalize_union_types(
440 type_: _AnnotationScanType,
441) -> _AnnotationScanType: ...
444def de_optionalize_union_types(
445 type_: _AnnotationScanType,
446) -> _AnnotationScanType:
447 """Given a type, filter out ``Union`` types that include ``NoneType``
448 to not include the ``NoneType``.
450 Contains extra logic to work on non-flattened unions, unions that contain
451 ``None`` (seen in py38, 37)
453 """
455 if is_fwd_ref(type_):
456 return _de_optionalize_fwd_ref_union_types(type_, False)
458 elif is_union(type_) and includes_none(type_):
459 if compat.py39:
460 typ = set(type_.__args__)
461 else:
462 # py38, 37 - unions are not automatically flattened, can contain
463 # None rather than NoneType
464 stack_of_unions = deque([type_])
465 typ = set()
466 while stack_of_unions:
467 u_typ = stack_of_unions.popleft()
468 for elem in u_typ.__args__:
469 if is_union(elem):
470 stack_of_unions.append(elem)
471 else:
472 typ.add(elem)
474 typ.discard(None) # type: ignore[arg-type,unused-ignore]
476 typ = {t for t in typ if t is not NoneType and not is_fwd_none(t)}
478 return make_union_type(*typ)
480 else:
481 return type_
484@overload
485def _de_optionalize_fwd_ref_union_types(
486 type_: ForwardRef, return_has_none: Literal[True]
487) -> bool: ...
490@overload
491def _de_optionalize_fwd_ref_union_types(
492 type_: ForwardRef, return_has_none: Literal[False]
493) -> _AnnotationScanType: ...
496def _de_optionalize_fwd_ref_union_types(
497 type_: ForwardRef, return_has_none: bool
498) -> Union[_AnnotationScanType, bool]:
499 """return the non-optional type for Optional[], Union[None, ...], x|None,
500 etc. without de-stringifying forward refs.
502 unfortunately this seems to require lots of hardcoded heuristics
504 """
506 annotation = type_.__forward_arg__
508 mm = re.match(r"^(.+?)\[(.+)\]$", annotation)
509 if mm:
510 g1 = mm.group(1).split(".")[-1]
511 if g1 == "Optional":
512 return True if return_has_none else ForwardRef(mm.group(2))
513 elif g1 == "Union":
514 if "[" in mm.group(2):
515 # cases like "Union[Dict[str, int], int, None]"
516 elements: list[str] = []
517 current: list[str] = []
518 ignore_comma = 0
519 for char in mm.group(2):
520 if char == "[":
521 ignore_comma += 1
522 elif char == "]":
523 ignore_comma -= 1
524 elif ignore_comma == 0 and char == ",":
525 elements.append("".join(current).strip())
526 current.clear()
527 continue
528 current.append(char)
529 else:
530 elements = re.split(r",\s*", mm.group(2))
531 parts = [ForwardRef(elem) for elem in elements if elem != "None"]
532 if return_has_none:
533 return len(elements) != len(parts)
534 else:
535 return make_union_type(*parts) if parts else Never # type: ignore[return-value] # noqa: E501
536 else:
537 return False if return_has_none else type_
539 pipe_tokens = re.split(r"\s*\|\s*", annotation)
540 has_none = "None" in pipe_tokens
541 if return_has_none:
542 return has_none
543 if has_none:
544 anno_str = "|".join(p for p in pipe_tokens if p != "None")
545 return ForwardRef(anno_str) if anno_str else Never # type: ignore[return-value] # noqa: E501
547 return type_
550def make_union_type(*types: _AnnotationScanType) -> Type[Any]:
551 """Make a Union type."""
553 return Union[types] # type: ignore
556def includes_none(type_: Any) -> bool:
557 """Returns if the type annotation ``type_`` allows ``None``.
559 This function supports:
560 * forward refs
561 * unions
562 * pep593 - Annotated
563 * pep695 - TypeAliasType (does not support looking into
564 fw reference of other pep695)
565 * NewType
566 * plain types like ``int``, ``None``, etc
567 """
568 if is_fwd_ref(type_):
569 return _de_optionalize_fwd_ref_union_types(type_, True)
570 if is_union(type_):
571 return any(includes_none(t) for t in get_args(type_))
572 if is_pep593(type_):
573 return includes_none(get_args(type_)[0])
574 if is_pep695(type_):
575 return any(includes_none(t) for t in pep695_values(type_))
576 if is_newtype(type_):
577 return includes_none(type_.__supertype__)
578 try:
579 return type_ in (NoneType, None) or is_fwd_none(type_)
580 except TypeError:
581 # if type_ is Column, mapped_column(), etc. the use of "in"
582 # resolves to ``__eq__()`` which then gives us an expression object
583 # that can't resolve to boolean. just catch it all via exception
584 return False
587def is_a_type(type_: Any) -> bool:
588 return (
589 isinstance(type_, type)
590 or get_origin(type_) is not None
591 or getattr(type_, "__module__", None)
592 in ("typing", "typing_extensions")
593 or type(type_).__mro__[0].__module__ in ("typing", "typing_extensions")
594 )
597def is_union(type_: Any) -> TypeGuard[ArgsTypeProtocol]:
598 return is_origin_of(type_, "Union", "UnionType")
601def is_origin_of_cls(
602 type_: Any, class_obj: Union[Tuple[Type[Any], ...], Type[Any]]
603) -> bool:
604 """return True if the given type has an __origin__ that shares a base
605 with the given class"""
607 origin = get_origin(type_)
608 if origin is None:
609 return False
611 return isinstance(origin, type) and issubclass(origin, class_obj)
614def is_origin_of(
615 type_: Any, *names: str, module: Optional[str] = None
616) -> bool:
617 """return True if the given type has an __origin__ with the given name
618 and optional module."""
620 origin = get_origin(type_)
621 if origin is None:
622 return False
624 return _get_type_name(origin) in names and (
625 module is None or origin.__module__.startswith(module)
626 )
629def _get_type_name(type_: Type[Any]) -> str:
630 if compat.py310:
631 return type_.__name__
632 else:
633 typ_name = getattr(type_, "__name__", None)
634 if typ_name is None:
635 typ_name = getattr(type_, "_name", None)
637 return typ_name # type: ignore
640class DescriptorProto(Protocol):
641 def __get__(self, instance: object, owner: Any) -> Any: ...
643 def __set__(self, instance: Any, value: Any) -> None: ...
645 def __delete__(self, instance: Any) -> None: ...
648_DESC = TypeVar("_DESC", bound=DescriptorProto)
651class DescriptorReference(Generic[_DESC]):
652 """a descriptor that refers to a descriptor.
654 used for cases where we need to have an instance variable referring to an
655 object that is itself a descriptor, which typically confuses typing tools
656 as they don't know when they should use ``__get__`` or not when referring
657 to the descriptor assignment as an instance variable. See
658 sqlalchemy.orm.interfaces.PropComparator.prop
660 """
662 if TYPE_CHECKING:
664 def __get__(self, instance: object, owner: Any) -> _DESC: ...
666 def __set__(self, instance: Any, value: _DESC) -> None: ...
668 def __delete__(self, instance: Any) -> None: ...
671_DESC_co = TypeVar("_DESC_co", bound=DescriptorProto, covariant=True)
674class RODescriptorReference(Generic[_DESC_co]):
675 """a descriptor that refers to a descriptor.
677 same as :class:`.DescriptorReference` but is read-only, so that subclasses
678 can define a subtype as the generically contained element
680 """
682 if TYPE_CHECKING:
684 def __get__(self, instance: object, owner: Any) -> _DESC_co: ...
686 def __set__(self, instance: Any, value: Any) -> NoReturn: ...
688 def __delete__(self, instance: Any) -> NoReturn: ...
691_FN = TypeVar("_FN", bound=Optional[Callable[..., Any]])
694class CallableReference(Generic[_FN]):
695 """a descriptor that refers to a callable.
697 works around mypy's limitation of not allowing callables assigned
698 as instance variables
701 """
703 if TYPE_CHECKING:
705 def __get__(self, instance: object, owner: Any) -> _FN: ...
707 def __set__(self, instance: Any, value: _FN) -> None: ...
709 def __delete__(self, instance: Any) -> None: ...
712class _TypingInstances:
713 def __getattr__(self, key: str) -> tuple[type, ...]:
714 types = tuple(
715 {
716 t
717 for t in [
718 getattr(typing, key, None),
719 getattr(typing_extensions, key, None),
720 ]
721 if t is not None
722 }
723 )
724 if not types:
725 raise AttributeError(key)
726 self.__dict__[key] = types
727 return types
730_type_tuples = _TypingInstances()
731if TYPE_CHECKING:
732 _type_instances = typing_extensions
733else:
734 _type_instances = _type_tuples
736LITERAL_TYPES = _type_tuples.Literal