Coverage for src / invariant / invocation.py: 89.36%

1"""Shared operation invocation helpers.""" 

2 

3import inspect 

4from collections.abc import Callable 

5from threading import RLock 

6from typing import Any 

7from weakref import WeakKeyDictionary 

8 

9from invariant.cacheable import is_cacheable 

10 

11_SIGNATURES = WeakKeyDictionary() 

12_SIGNATURE_LOCK = RLock() 

13 

14 

15def _signature(op): 

16 # Weak keys never retain closures and their resource payloads between jobs. 

17 try: 

18 with _SIGNATURE_LOCK: 

19 signature = _SIGNATURES.get(op) 

20 if signature is None: 

21 signature = inspect.signature(op) 

22 if len(_SIGNATURES) >= 512: 

23 _SIGNATURES.pop(next(iter(_SIGNATURES))) 

24 _SIGNATURES[op] = signature 

25 return signature 

26 except TypeError: 

27 return inspect.signature(op) 

28 

29 

30def invocation_kwargs(op, op_name, manifest): 

31 sig = _signature(op) 

32 kwargs = {} 

33 for name, param in sig.parameters.items(): 

34 if name in manifest: 

35 kwargs[name] = manifest[name] 

36 elif param.default is inspect.Parameter.empty and param.kind not in { 

37 inspect.Parameter.VAR_KEYWORD, 

38 inspect.Parameter.VAR_POSITIONAL, 

39 }: 

40 raise ValueError(f"Op '{op_name}': missing required parameter '{name}'") 

41 if any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values()): 

42 kwargs.update( 

43 {key: value for key, value in manifest.items() if key not in kwargs} 

44 ) 

45 return kwargs 

46 

47 

48def validate_result(result, op_name): 

49 if not is_cacheable(result): 

50 raise TypeError( 

51 f"Op '{op_name}' returned {type(result).__name__}, which is not a cacheable type" 

52 ) 

53 return result 

54 

55 

56async def invoke_op_async(op, op_name, manifest): 

57 result = op(**invocation_kwargs(op, op_name, manifest)) 

58 if inspect.isawaitable(result): 

59 result = await result 

60 return validate_result(result, op_name) 

61 

62 

63def invoke_op(op: Callable[..., Any], op_name: str, manifest: dict[str, Any]) -> Any: 

64 """Invoke an operation with kwargs dispatch and return validation. 

65 

66 Args: 

67 op: The callable operation to invoke. 

68 op_name: The name of the operation (for error messages). 

69 manifest: The manifest dictionary mapping parameter names to values. 

70 

71 Returns: 

72 The operation result (native type or ICacheable domain type). 

73 

74 Raises: 

75 ValueError: If required parameters are missing. 

76 TypeError: If return value is not cacheable. 

77 """ 

78 result = op(**invocation_kwargs(op, op_name, manifest)) 

79 if inspect.iscoroutine(result): 

80 result.close() 

81 raise TypeError( 

82 f"Op '{op_name}' is asynchronous; use AsyncExecutor with InlineScheduler" 

83 ) 

84 return validate_result(result, op_name) 

85 

86 

87__all__ = ["invoke_op", "invoke_op_async"]