Coverage for wasmtime/component/_types.py: 90%

922 statements  

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

1from abc import abstractmethod 

2from .. import _ffi as ffi, ImportType, ExportType, WasmtimeError, Storelike 

3from ._resources import ResourceAny, ResourceHost 

4from ._resource_type import ResourceType 

5from dataclasses import dataclass 

6import ctypes 

7from ctypes import byref, POINTER 

8from typing import Any, Optional, Dict, List, Union, Tuple, Set 

9from wasmtime import Managed, Engine 

10 

11 

12class ComponentType(Managed["ctypes._Pointer[ffi.wasmtime_component_type_t]"]): 

13 def __init__(self) -> None: 

14 raise WasmtimeError("Cannot directly construct a `ComponentType`") 

15 

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

17 ffi.wasmtime_component_type_delete(ptr) 

18 

19 @classmethod 

20 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_type_t]") -> "ComponentType": 

21 if not isinstance(ptr, POINTER(ffi.wasmtime_component_type_t)): 

22 raise TypeError("wrong pointer type") 

23 ty: "ComponentType" = cls.__new__(cls) 

24 ty._set_ptr(ptr) 

25 return ty 

26 

27 def imports(self, engine: Engine) -> Dict[str, "ComponentItem"]: 

28 """ 

29 Returns a dictionary of the imports of this component type. 

30 """ 

31 n = ffi.wasmtime_component_type_import_count(self.ptr(), engine.ptr()) 

32 items = {} 

33 for i in range(n): 

34 name_ptr = ctypes.POINTER(ctypes.c_char)() 

35 name_len = ctypes.c_size_t() 

36 item = ffi.wasmtime_component_item_t() 

37 found = ffi.wasmtime_component_type_import_nth(self.ptr(), 

38 engine.ptr(), 

39 i, 

40 byref(name_ptr), 

41 byref(name_len), 

42 byref(item)) 

43 assert(found) 

44 name = ctypes.string_at(name_ptr, name_len.value).decode('utf-8') 

45 items[name] = component_item_from_ptr(item) 

46 return items 

47 

48 def exports(self, engine: Engine) -> Dict[str, "ComponentItem"]: 

49 """ 

50 Returns a dictionary of the exports of this component type. 

51 """ 

52 n = ffi.wasmtime_component_type_export_count(self.ptr(), engine.ptr()) 

53 items = {} 

54 for i in range(n): 

55 name_ptr = ctypes.POINTER(ctypes.c_char)() 

56 name_len = ctypes.c_size_t() 

57 item = ffi.wasmtime_component_item_t() 

58 found = ffi.wasmtime_component_type_export_nth(self.ptr(), 

59 engine.ptr(), 

60 i, 

61 byref(name_ptr), 

62 byref(name_len), 

63 byref(item)) 

64 assert(found) 

65 name = ctypes.string_at(name_ptr, name_len.value).decode('utf-8') 

66 items[name] = component_item_from_ptr(item) 

67 return items 

68 

69 

70class ComponentInstanceType(Managed["ctypes._Pointer[ffi.wasmtime_component_instance_type_t]"]): 

71 def __init__(self) -> None: 

72 raise WasmtimeError("Cannot directly construct a `ComponentInstanceType`") 

73 

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

75 ffi.wasmtime_component_instance_type_delete(ptr) 

76 

77 @classmethod 

78 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_instance_type_t]") -> "ComponentInstanceType": 

79 if not isinstance(ptr, POINTER(ffi.wasmtime_component_instance_type_t)): 

80 raise TypeError("wrong pointer type") 

81 ty: "ComponentInstanceType" = cls.__new__(cls) 

82 ty._set_ptr(ptr) 

83 return ty 

84 

85 def exports(self, engine: Engine) -> Dict[str, "ComponentItem"]: 

86 """ 

87 Returns a dictionary of the exports of this component instance type. 

88 """ 

89 n = ffi.wasmtime_component_instance_type_export_count(self.ptr(), engine.ptr()) 

90 items = {} 

91 for i in range(n): 

92 name_ptr = ctypes.POINTER(ctypes.c_char)() 

93 name_len = ctypes.c_size_t() 

94 item = ffi.wasmtime_component_item_t() 

95 found = ffi.wasmtime_component_instance_type_export_nth(self.ptr(), 

96 engine.ptr(), 

97 i, 

98 byref(name_ptr), 

99 byref(name_len), 

100 byref(item)) 

101 name = ctypes.string_at(name_ptr, name_len.value).decode('utf-8') 

102 items[name] = component_item_from_ptr(item) 

103 assert(found) 

104 return items 

105 

106 

107class ModuleType(Managed["ctypes._Pointer[ffi.wasmtime_module_type_t]"]): 

108 def __init__(self) -> None: 

109 raise WasmtimeError("Cannot directly construct a `ModuleType`") 

110 

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

112 ffi.wasmtime_module_type_delete(ptr) 

113 

114 @classmethod 

115 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_module_type_t]") -> "ModuleType": 

116 if not isinstance(ptr, POINTER(ffi.wasmtime_module_type_t)): 

117 raise TypeError("wrong pointer type") 

118 ty: "ModuleType" = cls.__new__(cls) 

119 ty._set_ptr(ptr) 

120 return ty 

121 

122 def imports(self, engine: Engine) -> List[ImportType]: 

123 """ 

124 Returns a list of the imports of this module type. 

125 """ 

126 n = ffi.wasmtime_module_type_import_count(self.ptr(), engine.ptr()) 

127 items = [] 

128 for i in range(n): 

129 item = ffi.wasmtime_module_type_import_nth(self.ptr(), 

130 engine.ptr(), 

131 i) 

132 items.append(ImportType._from_ptr(item)) 

133 return items 

134 

135 def exports(self, engine: Engine) -> List[ExportType]: 

136 """ 

137 Returns a list of the exports of this module type. 

138 """ 

139 n = ffi.wasmtime_module_type_export_count(self.ptr(), engine.ptr()) 

140 items = [] 

141 for i in range(n): 

142 item = ffi.wasmtime_module_type_export_nth(self.ptr(), 

143 engine.ptr(), 

144 i) 

145 items.append(ExportType._from_ptr(item)) 

146 return items 

147 

148 

149class FuncType(Managed["ctypes._Pointer[ffi.wasmtime_component_func_type_t]"]): 

150 _owner: Any 

151 

152 def __init__(self) -> None: 

153 raise WasmtimeError("Cannot directly construct a `FuncType`") 

154 

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

156 if self._owner is None: 

157 ffi.wasmtime_component_func_type_delete(ptr) 

158 

159 @classmethod 

160 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_func_type_t]", owner: Any = None) -> "FuncType": 

161 if not isinstance(ptr, POINTER(ffi.wasmtime_component_func_type_t)): 

162 raise TypeError("wrong pointer type") 

163 ty: "FuncType" = cls.__new__(cls) 

164 ty._set_ptr(ptr) 

165 ty._owner = owner 

166 return ty 

167 

168 @property 

169 def params(self) -> List[Tuple[str, 'ValType']]: 

170 """ 

171 Returns the parameter types of this component function type. 

172 """ 

173 n = ffi.wasmtime_component_func_type_param_count(self.ptr()) 

174 items = [] 

175 for i in range(n): 

176 name_ptr = ctypes.POINTER(ctypes.c_char)() 

177 name_len = ctypes.c_size_t() 

178 valtype_ptr = ffi.wasmtime_component_valtype_t() 

179 found = ffi.wasmtime_component_func_type_param_nth(self.ptr(), 

180 i, 

181 byref(name_ptr), 

182 byref(name_len), 

183 byref(valtype_ptr)) 

184 assert(found) 

185 name = ctypes.string_at(name_ptr, name_len.value).decode('utf-8') 

186 valtype = valtype_from_ptr(valtype_ptr) 

187 items.append((name, valtype)) 

188 return items 

189 

190 @property 

191 def result(self) -> Optional['ValType']: 

192 """ 

193 Returns the result type of this component function type, if any. 

194 """ 

195 valtype_ptr = ffi.wasmtime_component_valtype_t() 

196 has_result = ffi.wasmtime_component_func_type_result(self.ptr(), byref(valtype_ptr)) 

197 if not has_result: 

198 return None 

199 return valtype_from_ptr(valtype_ptr) 

200 

201 

202class ValType: 

203 @abstractmethod 

204 def add_classes(self, s: Set[type]) -> None: 

205 """ 

206 Returns the python class that is created by `convert_from_c` and 

207 accepted by `convert_to_c` 

208 """ 

209 pass 

210 

211 @abstractmethod 

212 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

213 """ 

214 Converts `val` to this type and stores it in `ptr` 

215 """ 

216 pass 

217 

218 @abstractmethod 

219 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

220 """ 

221 Converts `val` to Python 

222 """ 

223 pass 

224 

225 

226@dataclass 

227class Bool(ValType): 

228 def add_classes(self, s: Set[type]) -> None: 

229 s.add(bool) 

230 

231 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

232 if not isinstance(val, bool): 

233 raise TypeError("expected bool for Bool type") 

234 ptr.contents.kind = ffi.WASMTIME_COMPONENT_BOOL 

235 ptr.contents.of.boolean = val 

236 

237 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

238 assert(c.kind == ffi.WASMTIME_COMPONENT_BOOL.value) 

239 return bool(c.of.boolean) 

240 

241 

242@dataclass 

243class S8(ValType): 

244 def add_classes(self, s: Set[type]) -> None: 

245 s.add(int) 

246 

247 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

248 if not isinstance(val, int): 

249 raise TypeError("expected int for S8 type") 

250 ptr.contents.kind = ffi.WASMTIME_COMPONENT_S8 

251 ptr.contents.of.s8 = val 

252 

253 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

254 assert(c.kind == ffi.WASMTIME_COMPONENT_S8.value) 

255 return int(c.of.s8) 

256 

257 

258@dataclass 

259class S16(ValType): 

260 def add_classes(self, s: Set[type]) -> None: 

261 s.add(int) 

262 

263 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

264 if not isinstance(val, int): 

265 raise TypeError("expected int for S16 type") 

266 ptr.contents.kind = ffi.WASMTIME_COMPONENT_S16 

267 ptr.contents.of.s16 = val 

268 

269 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

270 assert(c.kind == ffi.WASMTIME_COMPONENT_S16.value) 

271 return int(c.of.s16) 

272 

273 

274@dataclass 

275class S32(ValType): 

276 def add_classes(self, s: Set[type]) -> None: 

277 s.add(int) 

278 

279 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

280 if not isinstance(val, int): 

281 raise TypeError("expected int for S32 type") 

282 ptr.contents.kind = ffi.WASMTIME_COMPONENT_S32 

283 ptr.contents.of.s32 = val 

284 

285 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

286 assert(c.kind == ffi.WASMTIME_COMPONENT_S32.value) 

287 return int(c.of.s32) 

288 

289 

290@dataclass 

291class S64(ValType): 

292 def add_classes(self, s: Set[type]) -> None: 

293 s.add(int) 

294 

295 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

296 if not isinstance(val, int): 

297 raise TypeError("expected int for S64 type") 

298 ptr.contents.kind = ffi.WASMTIME_COMPONENT_S64 

299 ptr.contents.of.s64 = val 

300 

301 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

302 assert(c.kind == ffi.WASMTIME_COMPONENT_S64.value) 

303 return int(c.of.s64) 

304 

305 

306@dataclass 

307class U8(ValType): 

308 def add_classes(self, s: Set[type]) -> None: 

309 s.add(int) 

310 

311 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

312 if not isinstance(val, int): 

313 raise TypeError("expected int for U8 type") 

314 ptr.contents.kind = ffi.WASMTIME_COMPONENT_U8 

315 ptr.contents.of.u8 = val 

316 

317 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

318 assert(c.kind == ffi.WASMTIME_COMPONENT_U8.value) 

319 return int(c.of.u8) 

320 

321 

322@dataclass 

323class U16(ValType): 

324 def add_classes(self, s: Set[type]) -> None: 

325 s.add(int) 

326 

327 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

328 if not isinstance(val, int): 

329 raise TypeError("expected int for U16 type") 

330 ptr.contents.kind = ffi.WASMTIME_COMPONENT_U16 

331 ptr.contents.of.u16 = val 

332 

333 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

334 assert(c.kind == ffi.WASMTIME_COMPONENT_U16.value) 

335 return int(c.of.u16) 

336 

337 

338@dataclass 

339class U32(ValType): 

340 def add_classes(self, s: Set[type]) -> None: 

341 s.add(int) 

342 

343 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

344 if not isinstance(val, int): 

345 raise TypeError("expected int for U32 type") 

346 ptr.contents.kind = ffi.WASMTIME_COMPONENT_U32 

347 ptr.contents.of.u32 = val 

348 

349 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

350 assert(c.kind == ffi.WASMTIME_COMPONENT_U32.value) 

351 return int(c.of.u32) 

352 

353 

354@dataclass 

355class U64(ValType): 

356 def add_classes(self, s: Set[type]) -> None: 

357 s.add(int) 

358 

359 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

360 if not isinstance(val, int): 

361 raise TypeError("expected int for U64 type") 

362 ptr.contents.kind = ffi.WASMTIME_COMPONENT_U64 

363 ptr.contents.of.u64 = val 

364 

365 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

366 assert(c.kind == ffi.WASMTIME_COMPONENT_U64.value) 

367 return int(c.of.u64) 

368 

369 

370@dataclass 

371class F32(ValType): 

372 def add_classes(self, s: Set[type]) -> None: 

373 s.add(float) 

374 

375 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

376 if not isinstance(val, float): 

377 raise TypeError("expected float for F32 type") 

378 ptr.contents.kind = ffi.WASMTIME_COMPONENT_F32 

379 ptr.contents.of.f32 = val 

380 

381 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

382 assert(c.kind == ffi.WASMTIME_COMPONENT_F32.value) 

383 return float(c.of.f32) 

384 

385 

386@dataclass 

387class F64(ValType): 

388 def add_classes(self, s: Set[type]) -> None: 

389 s.add(float) 

390 

391 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

392 if not isinstance(val, float): 

393 raise TypeError("expected float for F64 type") 

394 ptr.contents.kind = ffi.WASMTIME_COMPONENT_F64 

395 ptr.contents.of.f64 = val 

396 

397 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

398 assert(c.kind == ffi.WASMTIME_COMPONENT_F64.value) 

399 return float(c.of.f64) 

400 

401 

402@dataclass 

403class Char(ValType): 

404 def add_classes(self, s: Set[type]) -> None: 

405 s.add(str) 

406 

407 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

408 if not isinstance(val, str) or len(val) != 1: 

409 raise TypeError("expected single-character string for Char type") 

410 ptr.contents.kind = ffi.WASMTIME_COMPONENT_CHAR 

411 ptr.contents.of.character = ord(val) 

412 

413 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

414 assert(c.kind == ffi.WASMTIME_COMPONENT_CHAR.value) 

415 return chr(c.of.character) 

416 

417 

418 

419@dataclass 

420class String(ValType): 

421 def add_classes(self, s: Set[type]) -> None: 

422 s.add(str) 

423 

424 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

425 if not isinstance(val, str): 

426 raise TypeError("expected string type") 

427 ptr.contents.kind = ffi.WASMTIME_COMPONENT_STRING 

428 ptr.contents.of.string = ffi.str_to_capi(val) 

429 

430 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

431 assert(c.kind == ffi.WASMTIME_COMPONENT_STRING.value) 

432 ret = ffi.to_str(c.of.string) 

433 ffi.wasm_byte_vec_delete(byref(c.of.string)) 

434 return ret 

435 

436 

437@dataclass 

438class ErrorContext(ValType): 

439 def add_classes(self, s: Set[type]) -> None: 

440 raise NotImplementedError("ErrorContext conversion not implemented yet") 

441 

442 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

443 raise NotImplementedError("ErrorContext conversion not implemented yet") 

444 

445 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

446 raise NotImplementedError("ErrorContext conversion not implemented yet") 

447 

448 

449class ListType(Managed["ctypes._Pointer[ffi.wasmtime_component_list_type_t]"], ValType): 

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

451 ffi.wasmtime_component_list_type_delete(ptr) 

452 

453 @classmethod 

454 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_list_type_t]") -> "ListType": 

455 if not isinstance(ptr, POINTER(ffi.wasmtime_component_list_type_t)): 

456 raise TypeError("wrong pointer type") 

457 ty: "ListType" = cls.__new__(cls) 

458 ty._set_ptr(ptr) 

459 return ty 

460 

461 @property 

462 def element(self) -> 'ValType': 

463 """ 

464 Returns the element type of this list type. 

465 """ 

466 valtype_ptr = ffi.wasmtime_component_valtype_t() 

467 ffi.wasmtime_component_list_type_element(self.ptr(), byref(valtype_ptr)) 

468 return valtype_from_ptr(valtype_ptr) 

469 

470 def __eq__(self, other: object) -> bool: 

471 if not isinstance(other, ListType): 

472 return False 

473 return ffi.wasmtime_component_list_type_equal(self.ptr(), other.ptr()) 

474 

475 def _is_bytes(self) -> bool: 

476 return isinstance(self.element, U8) 

477 

478 def add_classes(self, s: Set[type]) -> None: 

479 if self._is_bytes(): 

480 s.add(bytes) 

481 else: 

482 s.add(list) 

483 

484 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

485 element = self.element 

486 if self._is_bytes(): 

487 if not isinstance(val, bytes): 

488 raise TypeError("expected bytes value") 

489 else: 

490 if not isinstance(val, list): 

491 raise TypeError("expected list value") 

492 assert(isinstance(val, (bytes, list))) # here for mypy 

493 raw = ffi.wasmtime_component_vallist_t() 

494 ffi.wasmtime_component_vallist_new_uninit(raw, len(val)) 

495 i = 0 

496 cleanup = True 

497 try: 

498 for e in val: 

499 element.convert_to_c(store, e, ctypes.pointer(raw.data[i])) 

500 i += 1 

501 ptr.contents.kind = ffi.WASMTIME_COMPONENT_LIST 

502 ptr.contents.of.list = raw 

503 cleanup = False 

504 finally: 

505 if cleanup: 

506 for j in range(i, len(val)): 

507 raw.data[j].kind = ffi.WASMTIME_COMPONENT_BOOL 

508 raw.data[j].of.boolean = False 

509 ffi.wasmtime_component_vallist_delete(byref(raw)) 

510 

511 

512 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

513 assert(c.kind == ffi.WASMTIME_COMPONENT_LIST.value) 

514 try: 

515 ret = [] 

516 element = self.element 

517 for i in range(c.of.record.size): 

518 raw_field = c.of.tuple.data[i] 

519 ret.append(element.convert_from_c(raw_field)) 

520 raw_field.kind = ffi.WASMTIME_COMPONENT_BOOL 

521 raw_field.of.boolean = False 

522 if self._is_bytes(): 

523 return bytes(ret) 

524 return ret 

525 finally: 

526 ffi.wasmtime_component_valtuple_delete(byref(c.of.tuple)) 

527 

528 

529class RecordType(Managed["ctypes._Pointer[ffi.wasmtime_component_record_type_t]"], ValType): 

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

531 ffi.wasmtime_component_record_type_delete(ptr) 

532 

533 @classmethod 

534 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_record_type_t]") -> "RecordType": 

535 if not isinstance(ptr, POINTER(ffi.wasmtime_component_record_type_t)): 

536 raise TypeError("wrong pointer type") 

537 ty: "RecordType" = cls.__new__(cls) 

538 ty._set_ptr(ptr) 

539 return ty 

540 

541 @property 

542 def fields(self) -> List[Tuple[str, 'ValType']]: 

543 """ 

544 Returns the fields of this record type. 

545 """ 

546 n = ffi.wasmtime_component_record_type_field_count(self.ptr()) 

547 items = [] 

548 for i in range(n): 

549 name_ptr = ctypes.POINTER(ctypes.c_char)() 

550 name_len = ctypes.c_size_t() 

551 valtype_ptr = ffi.wasmtime_component_valtype_t() 

552 found = ffi.wasmtime_component_record_type_field_nth(self.ptr(), 

553 i, 

554 byref(name_ptr), 

555 byref(name_len), 

556 byref(valtype_ptr)) 

557 assert(found) 

558 name = ctypes.string_at(name_ptr, name_len.value).decode('utf-8') 

559 valtype = valtype_from_ptr(valtype_ptr) 

560 items.append((name, valtype)) 

561 return items 

562 

563 def __eq__(self, other: object) -> bool: 

564 if not isinstance(other, RecordType): 

565 return False 

566 return ffi.wasmtime_component_record_type_equal(self.ptr(), other.ptr()) 

567 

568 def add_classes(self, s: Set[type]) -> None: 

569 s.add(Record) 

570 

571 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

572 fields = self.fields 

573 raw = ffi.wasmtime_component_valrecord_t() 

574 ffi.wasmtime_component_valrecord_new_uninit(raw, len(fields)) 

575 i = 0 

576 cleanup = True 

577 try: 

578 for name, ty in fields: 

579 ty.convert_to_c(store, getattr(val, name), ctypes.pointer(raw.data[i].val)) 

580 raw.data[i].name = ffi.str_to_capi(name) 

581 i += 1 

582 ptr.contents.kind = ffi.WASMTIME_COMPONENT_RECORD 

583 ptr.contents.of.record = raw 

584 cleanup = False 

585 finally: 

586 if cleanup: 

587 for j in range(i, len(fields)): 

588 raw.data[j].val.kind = ffi.WASMTIME_COMPONENT_BOOL 

589 raw.data[j].val.of.boolean = False 

590 ffi.wasmtime_component_valrecord_delete(byref(raw)) 

591 

592 

593 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

594 assert(c.kind == ffi.WASMTIME_COMPONENT_RECORD.value) 

595 try: 

596 ret = Record() 

597 fields = self.fields 

598 for i, (name, ty) in zip(range(c.of.record.size), fields): 

599 raw_field = c.of.record.data[i] 

600 raw_name = ffi.to_str(raw_field.name) 

601 assert(raw_name == name) 

602 val = ty.convert_from_c(raw_field.val) 

603 setattr(ret, name, val) 

604 raw_field.val.kind = ffi.WASMTIME_COMPONENT_BOOL 

605 raw_field.val.of.boolean = False 

606 return ret 

607 finally: 

608 ffi.wasmtime_component_valrecord_delete(c.of.record) 

609 

610 

611class Record: 

612 def __eq__(self, other: Any) -> bool: 

613 for key, value in self.__dict__.items(): 

614 if not hasattr(other, key) or getattr(other, key) != value: 

615 return False 

616 return True 

617 

618 

619class TupleType(Managed["ctypes._Pointer[ffi.wasmtime_component_tuple_type_t]"], ValType): 

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

621 ffi.wasmtime_component_tuple_type_delete(ptr) 

622 

623 @classmethod 

624 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_tuple_type_t]") -> "TupleType": 

625 if not isinstance(ptr, POINTER(ffi.wasmtime_component_tuple_type_t)): 

626 raise TypeError("wrong pointer type") 

627 ty: "TupleType" = cls.__new__(cls) 

628 ty._set_ptr(ptr) 

629 return ty 

630 

631 @property 

632 def elements(self) -> List['ValType']: 

633 """ 

634 Returns the element types of this tuple type. 

635 """ 

636 n = ffi.wasmtime_component_tuple_type_types_count(self.ptr()) 

637 items = [] 

638 for i in range(n): 

639 valtype_ptr = ffi.wasmtime_component_valtype_t() 

640 found = ffi.wasmtime_component_tuple_type_types_nth(self.ptr(), 

641 i, 

642 byref(valtype_ptr)) 

643 assert(found) 

644 valtype = valtype_from_ptr(valtype_ptr) 

645 items.append(valtype) 

646 return items 

647 

648 def __eq__(self, other: object) -> bool: 

649 if not isinstance(other, TupleType): 

650 return False 

651 return ffi.wasmtime_component_tuple_type_equal(self.ptr(), other.ptr()) 

652 

653 def add_classes(self, s: Set[type]) -> None: 

654 s.add(tuple) 

655 

656 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

657 elements = self.elements 

658 if not isinstance(val, tuple): 

659 raise TypeError("expected tuple type") 

660 if len(val) != len(elements): 

661 raise TypeError("tuple length mismatch") 

662 raw = ffi.wasmtime_component_valtuple_t() 

663 ffi.wasmtime_component_valtuple_new_uninit(raw, len(elements)) 

664 i = 0 

665 cleanup = True 

666 try: 

667 for ty, element in zip(elements, val): 

668 ty.convert_to_c(store, element, ctypes.pointer(raw.data[i])) 

669 i += 1 

670 ptr.contents.kind = ffi.WASMTIME_COMPONENT_TUPLE 

671 ptr.contents.of.tuple = raw 

672 cleanup = False 

673 finally: 

674 if cleanup: 

675 for j in range(i, len(elements)): 

676 raw.data[j].kind = ffi.WASMTIME_COMPONENT_BOOL 

677 raw.data[j].of.boolean = False 

678 ffi.wasmtime_component_valtuple_delete(byref(raw)) 

679 

680 

681 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

682 assert(c.kind == ffi.WASMTIME_COMPONENT_TUPLE.value) 

683 try: 

684 ret: List[Any] = [] 

685 for i, ty in zip(range(c.of.record.size), self.elements): 

686 raw_field = c.of.tuple.data[i] 

687 ret.append(ty.convert_from_c(raw_field)) 

688 raw_field.kind = ffi.WASMTIME_COMPONENT_BOOL 

689 raw_field.of.boolean = False 

690 return tuple(ret) 

691 finally: 

692 ffi.wasmtime_component_valtuple_delete(byref(c.of.tuple)) 

693 

694 

695@dataclass 

696class Variant: 

697 tag: str 

698 payload: Optional[Any] = None 

699 

700 

701case_test_cache: Set[type] = set() 

702 

703 

704class VariantLikeType: 

705 @abstractmethod 

706 def _cases(self) -> List[Tuple[str, Optional['ValType']]]: 

707 pass 

708 

709 def add_classes(self, s: Set[type]) -> None: 

710 if self._tagged(): 

711 s.add(Variant) 

712 else: 

713 for _, ty in self._cases(): 

714 if ty is None: 

715 s.add(object) 

716 else: 

717 ty.add_classes(s) 

718 

719 def _tagged(self) -> bool: 

720 t: Set[type] = set() 

721 for _, ty in self._cases(): 

722 case: Set[type] = set() 

723 if ty is None: 

724 case.add(object) 

725 else: 

726 ty.add_classes(case) 

727 if len(case & t) != 0: 

728 return True 

729 t |= case 

730 return False 

731 

732 def _lower(self, store: Storelike, val: Any) -> Tuple[str, Optional['ctypes._Pointer[ffi.wasmtime_component_val_t]']]: 

733 tagged = self._tagged() 

734 if tagged and not isinstance(val, Variant): 

735 raise TypeError("expected Variant type") 

736 for name, ty in self._cases(): 

737 if tagged: 

738 if name != val.tag: 

739 continue 

740 elif ty is None: 

741 if val is not None: 

742 continue 

743 else: 

744 case_test_cache.clear() 

745 ty.add_classes(case_test_cache) 

746 if not isinstance(val, tuple(case_test_cache)): 

747 continue 

748 

749 if ty is None: 

750 return (name, None) 

751 raw = ffi.wasmtime_component_val_t() 

752 if tagged: 

753 payload = val.payload 

754 else: 

755 payload = val 

756 ty.convert_to_c(store, payload, ctypes.pointer(raw)) 

757 return (name, ffi.wasmtime_component_val_new(byref(raw))) 

758 raise ValueError('value not valid for this variant') 

759 

760 def _lift(self, tag: str, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> Any: 

761 tagged = self._tagged() 

762 for name, ty in self._cases(): 

763 if name != tag: 

764 continue 

765 if ty is None: 

766 if tagged: 

767 return Variant(tag) 

768 return None 

769 payload = ty.convert_from_c(ptr.contents) 

770 ptr.contents.kind = ffi.WASMTIME_COMPONENT_BOOL 

771 ptr.contents.of.boolean = False 

772 ffi.wasmtime_component_val_delete(ptr) 

773 if tagged: 

774 return Variant(tag, payload) 

775 return payload 

776 raise ValueError(f"tag {tag} not found in variant cases") 

777 

778 

779class VariantType(Managed["ctypes._Pointer[ffi.wasmtime_component_variant_type_t]"], ValType, VariantLikeType): 

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

781 ffi.wasmtime_component_variant_type_delete(ptr) 

782 

783 @classmethod 

784 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_variant_type_t]") -> "VariantType": 

785 if not isinstance(ptr, POINTER(ffi.wasmtime_component_variant_type_t)): 

786 raise TypeError("wrong pointer type") 

787 ty: "VariantType" = cls.__new__(cls) 

788 ty._set_ptr(ptr) 

789 return ty 

790 

791 @property 

792 def cases(self) -> List[Tuple[str, Optional['ValType']]]: 

793 """ 

794 Returns the cases of this variant type. 

795 """ 

796 n = ffi.wasmtime_component_variant_type_case_count(self.ptr()) 

797 items = [] 

798 for i in range(n): 

799 name_ptr = ctypes.POINTER(ctypes.c_char)() 

800 name_len = ctypes.c_size_t() 

801 has_payload = ctypes.c_bool() 

802 valtype_ptr = ffi.wasmtime_component_valtype_t() 

803 found = ffi.wasmtime_component_variant_type_case_nth(self.ptr(), 

804 i, 

805 byref(name_ptr), 

806 byref(name_len), 

807 byref(has_payload), 

808 byref(valtype_ptr)) 

809 assert(found) 

810 name = ctypes.string_at(name_ptr, name_len.value).decode('utf-8') 

811 if has_payload.value: 

812 valtype = valtype_from_ptr(valtype_ptr) 

813 else: 

814 valtype = None 

815 items.append((name, valtype)) 

816 return items 

817 

818 def __eq__(self, other: object) -> bool: 

819 if not isinstance(other, VariantType): 

820 return False 

821 return ffi.wasmtime_component_variant_type_equal(self.ptr(), other.ptr()) 

822 

823 def _cases(self) -> List[Tuple[str, Optional['ValType']]]: 

824 return self.cases 

825 

826 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

827 name, raw = self._lower(store, val) 

828 ptr.contents.kind = ffi.WASMTIME_COMPONENT_VARIANT 

829 ptr.contents.of.variant.discriminant = ffi.str_to_capi(name) 

830 ptr.contents.of.variant.val = raw 

831 

832 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

833 assert(c.kind == ffi.WASMTIME_COMPONENT_VARIANT.value) 

834 tag = ffi.to_str(c.of.variant.discriminant) 

835 ffi.wasm_byte_vec_delete(byref(c.of.variant.discriminant)) 

836 return self._lift(tag, c.of.variant.val) 

837 

838 

839class EnumType(Managed["ctypes._Pointer[ffi.wasmtime_component_enum_type_t]"], ValType): 

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

841 ffi.wasmtime_component_enum_type_delete(ptr) 

842 

843 @classmethod 

844 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_enum_type_t]") -> "EnumType": 

845 if not isinstance(ptr, POINTER(ffi.wasmtime_component_enum_type_t)): 

846 raise TypeError("wrong pointer type") 

847 ty: "EnumType" = cls.__new__(cls) 

848 ty._set_ptr(ptr) 

849 return ty 

850 

851 @property 

852 def names_count(self) -> int: 

853 """ 

854 Returns the numter of names of this enum type. 

855 """ 

856 return ffi.wasmtime_component_enum_type_names_count(self.ptr()) 

857 

858 def name(self, n: int) -> Optional[str]: 

859 """ 

860 Returns the nth name of this enum type. 

861 """ 

862 name_ptr = ctypes.POINTER(ctypes.c_char)() 

863 name_len = ctypes.c_size_t() 

864 found = ffi.wasmtime_component_enum_type_names_nth(self.ptr(), 

865 n, 

866 byref(name_ptr), 

867 byref(name_len)) 

868 if found: 

869 return ctypes.string_at(name_ptr, name_len.value).decode('utf-8') 

870 return None 

871 

872 @property 

873 def names(self) -> List[str]: 

874 """ 

875 Returns the names of this enum type. 

876 """ 

877 items = [] 

878 for i in range(self.names_count): 

879 name = self.name(i) 

880 assert(name is not None) 

881 items.append(name) 

882 return items 

883 

884 def __eq__(self, other: object) -> bool: 

885 if not isinstance(other, EnumType): 

886 return False 

887 return ffi.wasmtime_component_enum_type_equal(self.ptr(), other.ptr()) 

888 

889 def add_classes(self, s: Set[type]) -> None: 

890 s.add(str) 

891 

892 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

893 if isinstance(val, str): 

894 ptr.contents.kind = ffi.WASMTIME_COMPONENT_ENUM 

895 ptr.contents.of.enumeration = ffi.str_to_capi(val) 

896 else: 

897 raise TypeError("expected str type") 

898 

899 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

900 assert(c.kind == ffi.WASMTIME_COMPONENT_ENUM.value) 

901 ret = ffi.to_str(c.of.enumeration) 

902 ffi.wasm_byte_vec_delete(byref(c.of.enumeration)) 

903 return ret 

904 

905 

906class OptionType(Managed["ctypes._Pointer[ffi.wasmtime_component_option_type_t]"], ValType, VariantLikeType): 

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

908 ffi.wasmtime_component_option_type_delete(ptr) 

909 

910 @classmethod 

911 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_option_type_t]") -> "OptionType": 

912 if not isinstance(ptr, POINTER(ffi.wasmtime_component_option_type_t)): 

913 raise TypeError("wrong pointer type") 

914 ty: "OptionType" = cls.__new__(cls) 

915 ty._set_ptr(ptr) 

916 return ty 

917 

918 @property 

919 def payload(self) -> 'ValType': 

920 """ 

921 Returns the payload type of this option type. 

922 """ 

923 valtype_ptr = ffi.wasmtime_component_valtype_t() 

924 ffi.wasmtime_component_option_type_ty(self.ptr(), byref(valtype_ptr)) 

925 return valtype_from_ptr(valtype_ptr) 

926 

927 def __eq__(self, other: object) -> bool: 

928 if not isinstance(other, OptionType): 

929 return False 

930 return ffi.wasmtime_component_option_type_equal(self.ptr(), other.ptr()) 

931 

932 def _cases(self) -> List[Tuple[str, Optional['ValType']]]: 

933 return [('none', None), ('some', self.payload)] 

934 

935 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

936 _, raw = self._lower(store, val) 

937 ptr.contents.kind = ffi.WASMTIME_COMPONENT_OPTION 

938 ptr.contents.of.option = raw 

939 

940 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

941 assert(c.kind == ffi.WASMTIME_COMPONENT_OPTION.value) 

942 if c.of.option: 

943 tag = 'some' 

944 else: 

945 tag = 'none' 

946 return self._lift(tag, c.of.option) 

947 

948 

949 

950class ResultType(Managed["ctypes._Pointer[ffi.wasmtime_component_result_type_t]"], ValType, VariantLikeType): 

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

952 ffi.wasmtime_component_result_type_delete(ptr) 

953 

954 @classmethod 

955 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_result_type_t]") -> "ResultType": 

956 if not isinstance(ptr, POINTER(ffi.wasmtime_component_result_type_t)): 

957 raise TypeError("wrong pointer type") 

958 ty: "ResultType" = cls.__new__(cls) 

959 ty._set_ptr(ptr) 

960 return ty 

961 

962 @property 

963 def ok(self) -> Optional['ValType']: 

964 """ 

965 Returns the ok type of this result type, if any. 

966 """ 

967 valtype_ptr = ffi.wasmtime_component_valtype_t() 

968 has_ok = ffi.wasmtime_component_result_type_ok(self.ptr(), byref(valtype_ptr)) 

969 if not has_ok: 

970 return None 

971 return valtype_from_ptr(valtype_ptr) 

972 

973 @property 

974 def err(self) -> Optional['ValType']: 

975 """ 

976 Returns the err type of this result type, if any. 

977 """ 

978 valtype_ptr = ffi.wasmtime_component_valtype_t() 

979 has_err = ffi.wasmtime_component_result_type_err(self.ptr(), byref(valtype_ptr)) 

980 if not has_err: 

981 return None 

982 return valtype_from_ptr(valtype_ptr) 

983 

984 def __eq__(self, other: object) -> bool: 

985 if not isinstance(other, ResultType): 

986 return False 

987 return ffi.wasmtime_component_result_type_equal(self.ptr(), other.ptr()) 

988 

989 def _cases(self) -> List[Tuple[str, Optional['ValType']]]: 

990 return [('ok', self.ok), ('err', self.err)] 

991 

992 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

993 name, raw = self._lower(store, val) 

994 ptr.contents.kind = ffi.WASMTIME_COMPONENT_RESULT 

995 ptr.contents.of.result.is_ok = name == 'ok' 

996 ptr.contents.of.result.val = raw 

997 

998 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

999 assert(c.kind == ffi.WASMTIME_COMPONENT_RESULT.value) 

1000 if c.of.result.is_ok: 

1001 tag = 'ok' 

1002 else: 

1003 tag = 'err' 

1004 return self._lift(tag, c.of.result.val) 

1005 

1006 

1007class FlagsType(Managed["ctypes._Pointer[ffi.wasmtime_component_flags_type_t]"], ValType): 

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

1009 ffi.wasmtime_component_flags_type_delete(ptr) 

1010 

1011 @classmethod 

1012 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_flags_type_t]") -> "FlagsType": 

1013 if not isinstance(ptr, POINTER(ffi.wasmtime_component_flags_type_t)): 

1014 raise TypeError("wrong pointer type") 

1015 ty: "FlagsType" = cls.__new__(cls) 

1016 ty._set_ptr(ptr) 

1017 return ty 

1018 

1019 @property 

1020 def names(self) -> List[str]: 

1021 """ 

1022 Returns the names of this flags type. 

1023 """ 

1024 n = ffi.wasmtime_component_flags_type_names_count(self.ptr()) 

1025 items = [] 

1026 for i in range(n): 

1027 name_ptr = ctypes.POINTER(ctypes.c_char)() 

1028 name_len = ctypes.c_size_t() 

1029 found = ffi.wasmtime_component_flags_type_names_nth(self.ptr(), 

1030 i, 

1031 byref(name_ptr), 

1032 byref(name_len)) 

1033 assert(found) 

1034 name = ctypes.string_at(name_ptr, name_len.value).decode('utf-8') 

1035 items.append(name) 

1036 return items 

1037 

1038 def __eq__(self, other: object) -> bool: 

1039 if not isinstance(other, FlagsType): 

1040 return False 

1041 return ffi.wasmtime_component_flags_type_equal(self.ptr(), other.ptr()) 

1042 

1043 def add_classes(self, s: Set[type]) -> None: 

1044 s.add(set) 

1045 

1046 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

1047 if not isinstance(val, set): 

1048 raise TypeError("expected set type for Flags type") 

1049 for s in val: 

1050 if not isinstance(s, str): 

1051 raise TypeError("expected set of strings for Flags type") 

1052 

1053 raw = ffi.wasmtime_component_valflags_t() 

1054 ffi.wasmtime_component_valflags_new_uninit(raw, len(val)) 

1055 for i, s in enumerate(val): 

1056 raw.data[i] = ffi.str_to_capi(s) 

1057 ptr.contents.kind = ffi.WASMTIME_COMPONENT_FLAGS 

1058 ptr.contents.of.flags = raw 

1059 

1060 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

1061 assert(c.kind == ffi.WASMTIME_COMPONENT_FLAGS.value) 

1062 result = set() 

1063 for i in range(c.of.flags.size): 

1064 s = ffi.to_str(c.of.flags.data[i]) 

1065 result.add(s) 

1066 ffi.wasmtime_component_valflags_delete(byref(c.of.flags)) 

1067 return result 

1068 

1069 

1070class FutureType(Managed["ctypes._Pointer[ffi.wasmtime_component_future_type_t]"], ValType): 

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

1072 ffi.wasmtime_component_future_type_delete(ptr) 

1073 

1074 @classmethod 

1075 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_future_type_t]") -> "FutureType": 

1076 if not isinstance(ptr, POINTER(ffi.wasmtime_component_future_type_t)): 

1077 raise TypeError("wrong pointer type") 

1078 ty: "FutureType" = cls.__new__(cls) 

1079 ty._set_ptr(ptr) 

1080 return ty 

1081 

1082 @property 

1083 def payload(self) -> 'ValType': 

1084 """ 

1085 Returns the payload type of this future type. 

1086 """ 

1087 valtype_ptr = ffi.wasmtime_component_valtype_t() 

1088 ffi.wasmtime_component_future_type_ty(self.ptr(), byref(valtype_ptr)) 

1089 return valtype_from_ptr(valtype_ptr) 

1090 

1091 def __eq__(self, other: object) -> bool: 

1092 if not isinstance(other, FutureType): 

1093 return False 

1094 return ffi.wasmtime_component_future_type_equal(self.ptr(), other.ptr()) 

1095 

1096 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

1097 raise NotImplementedError("Future conversion not implemented yet") 

1098 

1099 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

1100 raise NotImplementedError("conversion not implemented yet") 

1101 

1102 

1103class StreamType(Managed["ctypes._Pointer[ffi.wasmtime_component_stream_type_t]"], ValType): 

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

1105 ffi.wasmtime_component_stream_type_delete(ptr) 

1106 

1107 @classmethod 

1108 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_stream_type_t]") -> "StreamType": 

1109 if not isinstance(ptr, POINTER(ffi.wasmtime_component_stream_type_t)): 

1110 raise TypeError("wrong pointer type") 

1111 ty: "StreamType" = cls.__new__(cls) 

1112 ty._set_ptr(ptr) 

1113 return ty 

1114 

1115 @property 

1116 def payload(self) -> 'ValType': 

1117 """ 

1118 Returns the payload type of this stream type. 

1119 """ 

1120 valtype_ptr = ffi.wasmtime_component_valtype_t() 

1121 ffi.wasmtime_component_stream_type_ty(self.ptr(), byref(valtype_ptr)) 

1122 return valtype_from_ptr(valtype_ptr) 

1123 

1124 def __eq__(self, other: object) -> bool: 

1125 if not isinstance(other, StreamType): 

1126 return False 

1127 return ffi.wasmtime_component_stream_type_equal(self.ptr(), other.ptr()) 

1128 

1129 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

1130 raise NotImplementedError("Stream conversion not implemented yet") 

1131 

1132 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

1133 raise NotImplementedError("conversion not implemented yet") 

1134 

1135 

1136@dataclass 

1137class OwnType(ValType): 

1138 ty: ResourceType 

1139 

1140 def add_classes(self, s: Set[type]) -> None: 

1141 s.add(ResourceAny) 

1142 s.add(ResourceHost) 

1143 

1144 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

1145 if isinstance(val, ResourceAny): 

1146 ptr.contents.kind = ffi.WASMTIME_COMPONENT_RESOURCE 

1147 ptr.contents.of.resource = val._consume() 

1148 elif isinstance(val, ResourceHost): 

1149 ptr.contents.kind = ffi.WASMTIME_COMPONENT_RESOURCE 

1150 ptr.contents.of.resource = val.to_any(store)._consume() 

1151 else: 

1152 raise TypeError("expected ResourceAny or ResourceHost for Own type") 

1153 

1154 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

1155 assert(c.kind == ffi.WASMTIME_COMPONENT_RESOURCE.value) 

1156 return ResourceAny._from_ptr(ffi.take_pointer(c.of, 'resource')) 

1157 

1158 

1159@dataclass 

1160class BorrowType(ValType): 

1161 ty: ResourceType 

1162 

1163 def add_classes(self, s: Set[type]) -> None: 

1164 s.add(ResourceAny) 

1165 s.add(ResourceHost) 

1166 

1167 def convert_to_c(self, store: Storelike, val: Any, ptr: 'ctypes._Pointer[ffi.wasmtime_component_val_t]') -> None: 

1168 if isinstance(val, ResourceAny): 

1169 ptr.contents.kind = ffi.WASMTIME_COMPONENT_RESOURCE 

1170 ptr.contents.of.resource = ffi.wasmtime_component_resource_any_clone(val.ptr()) 

1171 elif isinstance(val, ResourceHost): 

1172 ptr.contents.kind = ffi.WASMTIME_COMPONENT_RESOURCE 

1173 ptr.contents.of.resource = val.to_any(store)._consume() 

1174 else: 

1175 raise TypeError("expected ResourceAny or ResourceHost for Own type") 

1176 

1177 def convert_from_c(self, c: 'ffi.wasmtime_component_val_t') -> Any: 

1178 assert(c.kind == ffi.WASMTIME_COMPONENT_RESOURCE.value) 

1179 return ResourceAny._from_ptr(ffi.take_pointer(c.of, 'resource')) 

1180 

1181 

1182ComponentItem = Union[ 

1183 ComponentType, 

1184 ModuleType, 

1185 ComponentInstanceType, 

1186 ResourceType, 

1187 ValType, 

1188 FuncType, 

1189] 

1190 

1191 

1192def component_item_from_ptr(ptr: ffi.wasmtime_component_item_t) -> ComponentItem: 

1193 if ptr.kind == ffi.WASMTIME_COMPONENT_ITEM_COMPONENT.value: 

1194 return ComponentType._from_ptr(ptr.of.component) 

1195 elif ptr.kind == ffi.WASMTIME_COMPONENT_ITEM_MODULE.value: 

1196 return ModuleType._from_ptr(ptr.of.module) 

1197 elif ptr.kind == ffi.WASMTIME_COMPONENT_ITEM_COMPONENT_INSTANCE.value: 

1198 return ComponentInstanceType._from_ptr(ptr.of.component_instance) 

1199 elif ptr.kind == ffi.WASMTIME_COMPONENT_ITEM_RESOURCE.value: 

1200 return ResourceType._from_ptr(ptr.of.resource) 

1201 elif ptr.kind == ffi.WASMTIME_COMPONENT_ITEM_COMPONENT_FUNC.value: 

1202 return FuncType._from_ptr(ptr.of.component_func) 

1203 elif ptr.kind == ffi.WASMTIME_COMPONENT_ITEM_TYPE.value: 

1204 return valtype_from_ptr(ptr.of.type) 

1205 else: 

1206 ffi.wasmtime_component_item_delete(byref(ptr)) 

1207 raise TypeError("unknown component item kind") 

1208 

1209 

1210def valtype_from_ptr(ptr: ffi.wasmtime_component_valtype_t) -> ValType: 

1211 if ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_BOOL.value: 

1212 return Bool() 

1213 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_S8.value: 

1214 return S8() 

1215 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_S16.value: 

1216 return S16() 

1217 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_S32.value: 

1218 return S32() 

1219 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_S64.value: 

1220 return S64() 

1221 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_U8.value: 

1222 return U8() 

1223 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_U16.value: 

1224 return U16() 

1225 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_U32.value: 

1226 return U32() 

1227 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_U64.value: 

1228 return U64() 

1229 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_F32.value: 

1230 return F32() 

1231 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_F64.value: 

1232 return F64() 

1233 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_CHAR.value: 

1234 return Char() 

1235 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_STRING.value: 

1236 return String() 

1237 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_ERROR_CONTEXT.value: 

1238 return ErrorContext() 

1239 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_LIST.value: 

1240 return ListType._from_ptr(ptr.of.list) 

1241 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_RECORD.value: 

1242 return RecordType._from_ptr(ptr.of.record) 

1243 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_TUPLE.value: 

1244 return TupleType._from_ptr(ptr.of.tuple) 

1245 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_VARIANT.value: 

1246 return VariantType._from_ptr(ptr.of.variant) 

1247 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_ENUM.value: 

1248 return EnumType._from_ptr(ptr.of.enum_) 

1249 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_OPTION.value: 

1250 return OptionType._from_ptr(ptr.of.option) 

1251 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_RESULT.value: 

1252 return ResultType._from_ptr(ptr.of.result) 

1253 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_FLAGS.value: 

1254 return FlagsType._from_ptr(ptr.of.flags) 

1255 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_FUTURE.value: 

1256 return FutureType._from_ptr(ptr.of.future) 

1257 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_STREAM.value: 

1258 return StreamType._from_ptr(ptr.of.stream) 

1259 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_OWN.value: 

1260 return OwnType(ResourceType._from_ptr(ptr.of.own)) 

1261 elif ptr.kind == ffi.WASMTIME_COMPONENT_VALTYPE_BORROW.value: 

1262 return BorrowType(ResourceType._from_ptr(ptr.of.borrow)) 

1263 else: 

1264 ffi.wasmtime_component_valtype_delete(byref(ptr)) 

1265 raise TypeError("unknown component value type kind")