|
1 | 1 | import asyncio
|
2 | 2 | import inspect
|
| 3 | +from collections import OrderedDict |
3 | 4 | from functools import lru_cache
|
| 5 | +from itertools import zip_longest |
4 | 6 | from typing import (
|
5 | 7 | Any,
|
6 | 8 | Callable,
|
| 9 | + Generic, |
7 | 10 | Optional,
|
| 11 | + Protocol, |
8 | 12 | TypeVar,
|
| 13 | + Union, |
9 | 14 | get_origin,
|
10 | 15 | )
|
11 | 16 | from typing_extensions import get_args
|
12 | 17 |
|
13 |
| -from strawberry.utils.typing import is_generic_alias |
14 |
| - |
15 | 18 |
|
16 | 19 | def in_async_context() -> bool:
|
17 | 20 | # Based on the way django checks if there's an event loop in the current thread
|
@@ -67,57 +70,60 @@ class IntBarFoo(IntBar, Foo[str]): ...
|
67 | 70 | # {}
|
68 | 71 |
|
69 | 72 | get_specialized_type_var_map(Bar)
|
70 |
| - # {~T: ~T} |
| 73 | + # {} |
71 | 74 |
|
72 | 75 | get_specialized_type_var_map(IntBar)
|
73 |
| - # {~T: int} |
| 76 | + # {~T: int, ~K: int} |
74 | 77 |
|
75 | 78 | get_specialized_type_var_map(IntBarSubclass)
|
76 |
| - # {~T: int} |
| 79 | + # {~T: int, ~K: int} |
77 | 80 |
|
78 | 81 | get_specialized_type_var_map(IntBarFoo)
|
79 | 82 | # {~T: int, ~K: str}
|
80 | 83 | ```
|
81 | 84 | """
|
82 | 85 | from strawberry.types.base import has_object_definition
|
83 | 86 |
|
84 |
| - orig_bases = getattr(cls, "__orig_bases__", None) |
85 |
| - if orig_bases is None: |
86 |
| - # Specialized generic aliases will not have __orig_bases__ |
87 |
| - if get_origin(cls) is not None and is_generic_alias(cls): |
88 |
| - orig_bases = (cls,) |
89 |
| - else: |
90 |
| - # Not a specialized type |
91 |
| - return None |
92 |
| - |
93 |
| - type_var_map = {} |
94 |
| - |
95 |
| - # only get type vars for base generics (ie. Generic[T]) and for strawberry types |
| 87 | + param_args = OrderedDict[TypeVar, Union[None, TypeVar, type]]() |
96 | 88 |
|
97 |
| - orig_bases = [b for b in orig_bases if has_object_definition(b)] |
| 89 | + types: list[type] = [cls] |
| 90 | + while types: |
| 91 | + tp = types.pop(0) |
| 92 | + if (origin := get_origin(tp)) is None or origin in (Generic, Protocol): |
| 93 | + origin = tp |
98 | 94 |
|
99 |
| - for base in orig_bases: |
100 |
| - # Recursively get type var map from base classes |
101 |
| - if base is not cls: |
102 |
| - base_type_var_map = get_specialized_type_var_map(base) |
103 |
| - if base_type_var_map is not None: |
104 |
| - type_var_map.update(base_type_var_map) |
105 |
| - |
106 |
| - args = get_args(base) |
107 |
| - origin = getattr(base, "__origin__", None) |
108 |
| - |
109 |
| - params = origin and getattr(origin, "__parameters__", None) |
110 |
| - if params is None: |
111 |
| - params = getattr(base, "__parameters__", None) |
112 |
| - |
113 |
| - if not params: |
| 95 | + # only get type vars for base generics (i.e. Generic[T]) and for strawberry types |
| 96 | + if not has_object_definition(origin): |
114 | 97 | continue
|
115 | 98 |
|
116 |
| - type_var_map.update( |
117 |
| - {p.__name__: a for p, a in zip(params, args) if not isinstance(a, TypeVar)} |
118 |
| - ) |
119 |
| - |
120 |
| - return type_var_map |
| 99 | + if (type_params := getattr(origin, "__parameters__", None)) is not None: |
| 100 | + args = get_args(tp) |
| 101 | + for type_param, arg in zip_longest(type_params, args): |
| 102 | + if type_param not in param_args: |
| 103 | + param_args[type_param] = arg |
| 104 | + |
| 105 | + if orig_bases := getattr(origin, "__orig_bases__", None): |
| 106 | + types.extend(orig_bases) |
| 107 | + if not param_args: |
| 108 | + return None |
| 109 | + |
| 110 | + resolve = True |
| 111 | + while resolve: |
| 112 | + resolve = False |
| 113 | + for type_param, arg in list(param_args.items()): |
| 114 | + if arg is None or not isinstance(arg, TypeVar): |
| 115 | + continue |
| 116 | + resolved_arg = param_args.get(arg, None) if arg is not type_param else None |
| 117 | + param_args[type_param] = resolved_arg |
| 118 | + |
| 119 | + if resolved_arg: |
| 120 | + resolve = True |
| 121 | + |
| 122 | + return { |
| 123 | + k.__name__: v |
| 124 | + for k, v in reversed(param_args.items()) |
| 125 | + if v is not None and not isinstance(v, TypeVar) |
| 126 | + } |
121 | 127 |
|
122 | 128 |
|
123 | 129 | __all__ = ["get_func_args", "get_specialized_type_var_map", "in_async_context"]
|
0 commit comments