Coverage for pydantic_graph/pydantic_graph/_utils.py: 93.85%

1from __future__ import annotations as _annotations (empty)

2 

3import asyncio (empty)

4import types (empty)

5from functools import partial (empty)

6from typing import Any, Callable, TypeVar (empty)

7 

8from typing_extensions import ParamSpec, TypeIs, get_args, get_origin (empty)

9from typing_inspection import typing_objects (empty)

10from typing_inspection.introspection import is_union_origin (empty)

11 

12 

13def get_event_loop(): (empty)

14 try: (empty)

15 event_loop = asyncio.get_event_loop() (empty)

16 except RuntimeError: (empty)

17 event_loop = asyncio.new_event_loop() (empty)

18 asyncio.set_event_loop(event_loop) (empty)

19 return event_loop (empty)

20 

21 

22def get_union_args(tp: Any) -> tuple[Any, ...]: (empty)

23 """Extract the arguments of a Union type if `response_type` is a union, otherwise return an empty tuple.""" 

24 # similar to `pydantic_ai_slim/pydantic_ai/_result.py:get_union_args` 

25 if typing_objects.is_typealiastype(tp): 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true(empty)

26 tp = tp.__value__ 

27 

28 origin = get_origin(tp) (empty)

29 if is_union_origin(origin): (empty)

30 return get_args(tp) (empty)

31 else: 

32 return (tp,) (empty)

33 

34 

35def unpack_annotated(tp: Any) -> tuple[Any, list[Any]]: (empty)

36 """Strip `Annotated` from the type if present. 

37 

38 Returns: 

39 `(tp argument, ())` if not annotated, otherwise `(stripped type, annotations)`. 

40 """ 

41 origin = get_origin(tp) (empty)

42 if typing_objects.is_annotated(origin): (empty)

43 inner_tp, *args = get_args(tp) (empty)

44 return inner_tp, args (empty)

45 else: 

46 return tp, [] (empty)

47 

48 

49def comma_and(items: list[str]) -> str: (empty)

50 """Join with a comma and 'and' for the last item.""" 

51 if len(items) == 1: (empty)

52 return items[0] (empty)

53 else: 

54 # oxford comma ¯\_(ツ)_/¯ 

55 return ', '.join(items[:-1]) + ', and ' + items[-1] (empty)

56 

57 

58def get_parent_namespace(frame: types.FrameType | None) -> dict[str, Any] | None: (empty)

59 """Attempt to get the namespace where the graph was defined. 

60 

61 If the graph is defined with generics `Graph[a, b]` then another frame is inserted, and we have to skip that 

62 to get the correct namespace. 

63 """ 

64 if frame is not None: 64 ↛ exitline 64 didn't return from function 'get_parent_namespace' because the condition on line 64 was always true(empty)

65 if back := frame.f_back: 65 ↛ exitline 65 didn't return from function 'get_parent_namespace' because the condition on line 65 was always true(empty)

66 if back.f_globals.get('__name__') == 'typing': (empty)

67 # If the class calling this function is generic, explicitly parameterizing the class 

68 # results in a `typing._GenericAlias` instance, which proxies instantiation calls to the 

69 # "real" class and thus adding an extra frame to the call. To avoid pulling anything 

70 # from the `typing` module, use the correct frame (the one before): 

71 return get_parent_namespace(back) (empty)

72 else: 

73 return back.f_locals (empty)

74 

75 

76class Unset: (empty)

77 """A singleton to represent an unset value. (empty)

78 

79 Copied from pydantic_ai/_utils.py. 

80 """ 

81 

82 pass (empty)

83 

84 

85UNSET = Unset() (empty)

86T = TypeVar('T') (empty)

87 

88 

89def is_set(t_or_unset: T | Unset) -> TypeIs[T]: (empty)

90 return t_or_unset is not UNSET (empty)

91 

92 

93_P = ParamSpec('_P') (empty)

94_R = TypeVar('_R') (empty)

95 

96 

97async def run_in_executor(func: Callable[_P, _R], *args: _P.args, **kwargs: _P.kwargs) -> _R: (empty)

98 if kwargs: (empty)

99 # noinspection PyTypeChecker 

100 return await asyncio.get_running_loop().run_in_executor(None, partial(func, *args, **kwargs)) (empty)

101 else: 

102 return await asyncio.get_running_loop().run_in_executor(None, func, *args) # type: ignore (empty)