Coverage for wasmtime/component/_func.py: 100%

49 statements  

« prev     ^ index     » next       coverage.py v7.11.3, created at 2025-12-01 19:40 +0000

1from .. import _ffi as ffi, WasmtimeError, Storelike 

2from ctypes import byref, pointer 

3import ctypes 

4from ._types import ValType, valtype_from_ptr, FuncType 

5from typing import List, Tuple, Optional, Any 

6from ._enter import enter_wasm 

7 

8 

9class Func: 

10 _func: ffi.wasmtime_component_func_t 

11 

12 def __init__(self) -> None: 

13 raise WasmtimeError("Cannot directly construct a `Func`") 

14 

15 @classmethod 

16 def _from_raw(cls, func: ffi.wasmtime_component_func_t) -> "Func": 

17 ty: "Func" = cls.__new__(cls) 

18 ty._func = func 

19 return ty 

20 

21 def __call__(self, store: Storelike, *params: Any) -> Any: 

22 fty = self.type(store) 

23 param_tys = fty.params 

24 result_ty = fty.result 

25 if len(params) != len(param_tys): 

26 raise TypeError("wrong number of parameters provided: given %s, expected %s" % 

27 (len(params), len(param_tys))) 

28 param_capi = (ffi.wasmtime_component_val_t * len(params))() 

29 n = 0 

30 try: 

31 for (_name, ty), val in zip(param_tys, params): 

32 ty.convert_to_c(store, val, pointer(param_capi[n])) 

33 n += 1 

34 result_space = None 

35 result_capi = None 

36 result_len = 0 

37 if result_ty is not None: 

38 result_space = ffi.wasmtime_component_val_t() 

39 result_capi = byref(result_space) 

40 result_len = 1 

41 

42 def run() -> 'ctypes._Pointer[ffi.wasmtime_error_t]': 

43 return ffi.wasmtime_component_func_call(byref(self._func), 

44 store._context(), 

45 param_capi, 

46 n, 

47 result_capi, 

48 result_len) 

49 enter_wasm(run) 

50 

51 if result_space is None: 

52 return None 

53 assert(result_ty is not None) 

54 return result_ty.convert_from_c(result_space) 

55 

56 finally: 

57 for i in range(n): 

58 ffi.wasmtime_component_val_delete(byref(param_capi[i])) 

59 

60 

61 def post_return(self, store: Storelike) -> None: 

62 """ 

63 Performs any necessary post-return operations for this function. 

64 """ 

65 def run() -> 'ctypes._Pointer[ffi.wasmtime_error_t]': 

66 return ffi.wasmtime_component_func_post_return(byref(self._func), store._context()) 

67 enter_wasm(run) 

68 

69 def type(self, store: Storelike) -> FuncType: 

70 """ 

71 Returns the type of this function. 

72 """ 

73 ptr = ffi.wasmtime_component_func_type(byref(self._func), store._context()) 

74 return FuncType._from_ptr(ptr)