Coverage for wasmtime/_module.py: 91%

1from . import _ffi as ffi 

2from ._wat2wasm import _to_wasm 

3import ctypes 

4from wasmtime import Engine, wat2wasm, ImportType, ExportType, WasmtimeError, Managed 

5import typing 

6from os import PathLike 

7 

8 

9class Module(Managed["ctypes._Pointer[ffi.wasmtime_module_t]"]): 

10 

11 @classmethod 

12 def from_file(cls, engine: Engine, path: typing.Union[str, bytes, PathLike]) -> "Module": 

13 """ 

14 Compiles and creates a new `Module` by reading the file at `path` and 

15 then delegating to the `Module` constructor. 

16 """ 

17 

18 with open(path, "rb") as f: 

19 contents = f.read() 

20 return cls(engine, contents) 

21 

22 def __init__(self, engine: Engine, wasm: typing.Union[str, bytes]): 

23 if not isinstance(engine, Engine): 

24 raise TypeError("expected an Engine") 

25 

26 wasm = _to_wasm(wasm) 

27 

28 # TODO: can the copy be avoided here? I can't for the life of me 

29 # figure this out. 

30 binary = (ctypes.c_uint8 * len(wasm)).from_buffer_copy(wasm) 

31 ptr = ctypes.POINTER(ffi.wasmtime_module_t)() 

32 error = ffi.wasmtime_module_new(engine.ptr(), binary, len(wasm), ctypes.byref(ptr)) 

33 if error: 

34 raise WasmtimeError._from_ptr(error) 

35 self._set_ptr(ptr) 

36 

37 def _delete(self, ptr: "ctypes._Pointer[ffi.wasmtime_module_t]") -> None: 

38 ffi.wasmtime_module_delete(ptr) 

39 

40 @classmethod 

41 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_module_t]") -> "Module": 

42 if not isinstance(ptr, ctypes.POINTER(ffi.wasmtime_module_t)): 

43 raise TypeError("wrong pointer type") 

44 ty: "Module" = cls.__new__(cls) 

45 ty._set_ptr(ptr) 

46 return ty 

47 

48 @classmethod 

49 def deserialize(cls, engine: Engine, encoded: typing.Union[bytes, bytearray]) -> 'Module': 

50 """ 

51 Deserializes bytes previously created by `Module.serialize`. 

52 

53 This constructor for `Module` will deserialize bytes previously created 

54 by a serialized module. This will only succeed if the bytes were 

55 previously created by the same version of `wasmtime` as well as the 

56 same configuration within `Engine`. 

57 """ 

58 

59 if not isinstance(encoded, (bytes, bytearray)): 

60 raise TypeError("expected bytes") 

61 

62 ptr = ctypes.POINTER(ffi.wasmtime_module_t)() 

63 

64 # TODO: can the copy be avoided here? I can't for the life of me 

65 # figure this out. 

66 error = ffi.wasmtime_module_deserialize( 

67 engine.ptr(), 

68 (ctypes.c_uint8 * len(encoded)).from_buffer_copy(encoded), 

69 len(encoded), 

70 ctypes.byref(ptr)) 

71 if error: 

72 raise WasmtimeError._from_ptr(error) 

73 return cls._from_ptr(ptr) 

74 

75 @classmethod 

76 def deserialize_file(cls, engine: Engine, path: str) -> 'Module': 

77 """ 

78 Deserializes bytes previously created by `Module.serialize` that are 

79 stored in a file on the filesystem. 

80 

81 Otherwise this function is the same as `Module.deserialize`. 

82 """ 

83 

84 ptr = ctypes.POINTER(ffi.wasmtime_module_t)() 

85 path_bytes = path.encode('utf-8') 

86 error = ffi.wasmtime_module_deserialize_file( 

87 engine.ptr(), 

88 path_bytes, 

89 ctypes.byref(ptr)) 

90 if error: 

91 raise WasmtimeError._from_ptr(error) 

92 return cls._from_ptr(ptr) 

93 

94 @classmethod 

95 def validate(cls, engine: Engine, wasm: typing.Union[bytes, bytearray]) -> None: 

96 """ 

97 Validates whether the list of bytes `wasm` provided is a valid 

98 WebAssembly binary given the configuration in `store` 

99 

100 Raises a `WasmtimeError` if the wasm isn't valid. 

101 """ 

102 

103 if not isinstance(wasm, (bytes, bytearray)): 

104 raise TypeError("expected wasm bytes") 

105 

106 # TODO: can the copy be avoided here? I can't for the life of me 

107 # figure this out. 

108 buf = (ctypes.c_uint8 * len(wasm)).from_buffer_copy(wasm) 

109 error = ffi.wasmtime_module_validate(engine.ptr(), buf, len(wasm)) 

110 

111 if error: 

112 raise WasmtimeError._from_ptr(error) 

113 

114 @property 

115 def imports(self) -> typing.List[ImportType]: 

116 """ 

117 Returns the types of imports that this module has 

118 """ 

119 

120 imports = ImportTypeList() 

121 ffi.wasmtime_module_imports(self.ptr(), ctypes.byref(imports.vec)) 

122 ret = [] 

123 for i in range(0, imports.vec.size): 

124 ret.append(ImportType._from_ptr(imports.vec.data[i], imports)) 

125 return ret 

126 

127 @property 

128 def exports(self) -> typing.List[ExportType]: 

129 """ 

130 Returns the types of the exports that this module has 

131 """ 

132 

133 exports = ExportTypeList() 

134 ffi.wasmtime_module_exports(self.ptr(), ctypes.byref(exports.vec)) 

135 ret = [] 

136 for i in range(0, exports.vec.size): 

137 ret.append(ExportType._from_ptr(exports.vec.data[i], exports)) 

138 return ret 

139 

140 def serialize(self) -> bytearray: 

141 """ 

142 Serializes this module to a binary representation. 

143 

144 This method will serialize this module to an in-memory byte array which 

145 can be cached and later passed to `Module.deserialize` to recreate this 

146 module. 

147 """ 

148 raw = ffi.wasm_byte_vec_t() 

149 err = ffi.wasmtime_module_serialize(self.ptr(), ctypes.byref(raw)) 

150 if err: 

151 raise WasmtimeError._from_ptr(err) 

152 ret = ffi.to_bytes(raw) 

153 ffi.wasm_byte_vec_delete(ctypes.byref(raw)) 

154 return ret 

155 

156 

157class ImportTypeList: 

158 def __init__(self) -> None: 

159 self.vec = ffi.wasm_importtype_vec_t(0, None) 

160 

161 def __del__(self) -> None: 

162 ffi.wasm_importtype_vec_delete(ctypes.byref(self.vec)) 

163 

164 

165class ExportTypeList: 

166 def __init__(self) -> None: 

167 self.vec = ffi.wasm_exporttype_vec_t(0, None) 

168 

169 def __del__(self) -> None: 

170 ffi.wasm_exporttype_vec_delete(ctypes.byref(self.vec))