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

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

2 

3import inspect 

4from collections.abc import Callable 

5from typing import Any 

6 

7from invariant.cacheable import is_cacheable 

8 

9 

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

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

12 

13 Args: 

14 op: The callable operation to invoke. 

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

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

17 

18 Returns: 

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

20 

21 Raises: 

22 ValueError: If required parameters are missing. 

23 TypeError: If return value is not cacheable. 

24 """ 

25 sig = inspect.signature(op) 

26 kwargs: dict[str, Any] = {} 

27 

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

29 if name in manifest: 

30 kwargs[name] = manifest[name] 

31 elif ( 

32 param.default is not inspect.Parameter.empty 

33 or param.kind == inspect.Parameter.VAR_KEYWORD 

34 ): 

35 pass 

36 else: 

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

38 

39 has_var_kwargs = any( 

40 p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values() 

41 ) 

42 if has_var_kwargs: 

43 for key, val in manifest.items(): 

44 if key not in kwargs: 

45 kwargs[key] = val 

46 

47 result = op(**kwargs) 

48 

49 if not is_cacheable(result): 

50 raise TypeError( 

51 f"Op '{op_name}' returned {type(result).__name__}, " 

52 f"which is not a cacheable type" 

53 ) 

54 

55 return result 

56 

57 

58__all__ = ["invoke_op"]