Coverage for wasmtime/component/_component.py: 88%

1from .. import _ffi as ffi 

2from .._wat2wasm import _to_wasm 

3from ._types import ComponentType 

4import ctypes 

5from wasmtime import Engine, wat2wasm, WasmtimeError, Managed, Module 

6import typing 

7from os import PathLike 

8 

9 

10class ExportIndex(Managed["ctypes._Pointer[ffi.wasmtime_component_export_index_t]"]): 

11 

12 def __init__(self) -> None: 

13 raise WasmtimeError("Cannot directly construct an `ExportIndex`") 

14 

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

16 ffi.wasmtime_component_export_index_delete(ptr) 

17 

18 @classmethod 

19 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_export_index_t]") -> "ExportIndex": 

20 if not isinstance(ptr, ctypes.POINTER(ffi.wasmtime_component_export_index_t)): 

21 raise TypeError("wrong pointer type") 

22 ty: "ExportIndex" = cls.__new__(cls) 

23 ty._set_ptr(ptr) 

24 return ty 

25 

26 

27class Component(Managed["ctypes._Pointer[ffi.wasmtime_component_t]"]): 

28 

29 @classmethod 

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

31 """ 

32 Compiles and creates a new `Component` by reading the file at `path` and 

33 then delegating to the `Component` constructor. 

34 """ 

35 

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

37 contents = f.read() 

38 return cls(engine, contents) 

39 

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

41 if not isinstance(engine, Engine): 

42 raise TypeError("expected an Engine") 

43 

44 wasm = _to_wasm(wasm) 

45 

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

47 # figure this out. 

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

49 ptr = ctypes.POINTER(ffi.wasmtime_component_t)() 

50 error = ffi.wasmtime_component_new(engine.ptr(), binary, len(wasm), ctypes.byref(ptr)) 

51 if error: 

52 raise WasmtimeError._from_ptr(error) 

53 self._set_ptr(ptr) 

54 

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

56 ffi.wasmtime_component_delete(ptr) 

57 

58 @classmethod 

59 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_t]") -> "Component": 

60 if not isinstance(ptr, ctypes.POINTER(ffi.wasmtime_component_t)): 

61 raise TypeError("wrong pointer type") 

62 ty: "Component" = cls.__new__(cls) 

63 ty._set_ptr(ptr) 

64 return ty 

65 

66 @classmethod 

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

68 """ 

69 Deserializes bytes previously created by `Component.serialize`. 

70 

71 This constructor for `Component` will deserialize bytes previously created 

72 by a serialized component. This will only succeed if the bytes were 

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

74 same configuration within `Engine`. 

75 """ 

76 

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

78 raise TypeError("expected bytes") 

79 

80 ptr = ctypes.POINTER(ffi.wasmtime_component_t)() 

81 

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

83 # figure this out. 

84 error = ffi.wasmtime_component_deserialize( 

85 engine.ptr(), 

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

87 len(encoded), 

88 ctypes.byref(ptr)) 

89 if error: 

90 raise WasmtimeError._from_ptr(error) 

91 return cls._from_ptr(ptr) 

92 

93 @classmethod 

94 def deserialize_file(cls, engine: Engine, path: str) -> 'Component': 

95 """ 

96 Deserializes bytes previously created by `Component.serialize` that are 

97 stored in a file on the filesystem. 

98 

99 Otherwise this function is the same as `Component.deserialize`. 

100 """ 

101 

102 ptr = ctypes.POINTER(ffi.wasmtime_component_t)() 

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

104 error = ffi.wasmtime_component_deserialize_file( 

105 engine.ptr(), 

106 path_bytes, 

107 ctypes.byref(ptr)) 

108 if error: 

109 raise WasmtimeError._from_ptr(error) 

110 return cls._from_ptr(ptr) 

111 

112 def serialize(self) -> bytearray: 

113 """ 

114 Serializes this component to a binary representation. 

115 

116 This method will serialize this component to an in-memory byte array 

117 which can be cached and later passed to `Component.deserialize` to 

118 recreate this component. 

119 """ 

120 raw = ffi.wasm_byte_vec_t() 

121 err = ffi.wasmtime_component_serialize(self.ptr(), ctypes.byref(raw)) 

122 if err: 

123 raise WasmtimeError._from_ptr(err) 

124 ret = ffi.to_bytes(raw) 

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

126 return ret 

127 

128 def get_export_index(self, name: str, instance : typing.Optional[ExportIndex] = None) -> typing.Optional[ExportIndex]: 

129 """ 

130 Gets an `ExportIndex` from this component pointing to a specific item 

131 in this component. 

132 

133 The returned `ExportIndex` can later be used to lookup exports on an 

134 instance or it can be used to lookup further indexes if it points to an 

135 instance for example. 

136 """ 

137 name_bytes = name.encode('utf-8') 

138 name_buf = ctypes.create_string_buffer(name_bytes) 

139 ret = ffi.wasmtime_component_get_export_index( 

140 self.ptr(), 

141 instance.ptr() if instance is not None else None, 

142 name_buf, 

143 len(name_bytes)) 

144 if not ret: 

145 return None 

146 return ExportIndex._from_ptr(ret) 

147 

148 @property 

149 def type(self) -> ComponentType: 

150 """ 

151 Returns the `ComponentType` corresponding to this component. 

152 """ 

153 ptr = ffi.wasmtime_component_type(self.ptr()) 

154 return ComponentType._from_ptr(ptr)