0
# Component Model
1

2
WebAssembly component model integration through bindgen tooling, enabling rich type communication between Python and WebAssembly components with automatic binding generation, interface definitions, and high-level abstractions for complex data types.
3

4
## Capabilities
5

6
### Bindgen Code Generation
7

8
Automatic Python binding generation from WebAssembly components, providing seamless integration between Python and component model interfaces with type-safe communication and automatic marshalling.
9

10
```python { .api }
11
import wasmtime.bindgen
12

13
def generate(name: str, component: bytes) -> Dict[str, bytes]:
14
    """
15
    Generate Python bindings from a WebAssembly component.
16
    
17
    Parameters:
18
    - name: Name for the generated bindings (used as module name)
19
    - component: Binary WebAssembly component data
20
    
21
    Returns:
22
    Dictionary mapping file names to generated Python code bytes
23
    
24
    Raises:
25
    RuntimeError: If component parsing or binding generation fails
26
    """
27
```
28

29
### Component Runtime
30

31
Runtime system for WebAssembly components providing execution environment, import/export management, and interface implementation for component model applications.
32

33
```python { .api }
34
class Root:
35
    """Generated root component binding class"""
36
    def __init__(self, store: Store, imports: RootImports): ...
37

38
class RootImports:
39
    """Generated import interface definitions"""
40
    def __init__(self, *host_implementations): ...
41

42
class Err:
43
    """Error result type for component operations"""
44
    @property
45
    def value(self): ...
46
```
47

48
### WASI Component Implementations
49

50
Host implementations of WASI component interfaces providing system integration, I/O operations, filesystem access, and environment interaction for component model applications.
51

52
```python { .api }
53
# Stream implementations
54
class WasiStreams:
55
    def drop_input_stream(self, this: int) -> None: ...
56
    def write(self, this: int, buf: bytes) -> Tuple[int, int]: ...
57
    def blocking_write(self, this: int, buf: bytes) -> Tuple[int, int]: ...
58
    def drop_output_stream(self, this: int) -> None: ...
59

60
# Filesystem implementations  
61
class WasiTypes:
62
    def write_via_stream(self, this: int, offset: int): ...
63
    def append_via_stream(self, this: int): ...
64
    def get_type(self, this: int): ...
65
    def drop_descriptor(self, this: int) -> None: ...
66

67
# Environment implementations
68
class WasiEnvironment:
69
    def get_environment(self) -> List[Tuple[str, str]]: ...
70

71
class WasiPreopens:
72
    def get_directories(self) -> List[Tuple[int, str]]: ...
73

74
# Standard I/O implementations
75
class WasiStdin:
76
    def get_stdin(self) -> int: ...
77

78
class WasiStdout:
79
    def get_stdout(self) -> int: ...
80

81
class WasiStderr:  
82
    def get_stderr(self) -> int: ...
83

84
# Random number generation
85
class WasiRandom:
86
    def get_random_bytes(self, len: int) -> bytes: ...
87

88
# Process control
89
class WasiExit:
90
    def exit(self, status) -> None: ...
91

92
# Terminal interfaces
93
class WasiTerminalInput:
94
    def drop_terminal_input(self, this: int) -> None: ...
95

96
class WasiTerminalOutput:
97
    def drop_terminal_output(self, this: int) -> None: ...
98

99
class WasiTerminalStdin:
100
    def get_terminal_stdin(self) -> Optional[int]: ...
101

102
class WasiTerminalStdout:
103
    def get_terminal_stdout(self) -> Optional[int]: ...
104

105
class WasiTerminalStderr:
106
    def get_terminal_stderr(self) -> Optional[int]: ...
107
```
108

109
## Usage Examples
110

111
### Basic Component Binding Generation
112

113
```python
114
import wasmtime.bindgen
115

116
# Load a WebAssembly component
117
with open("my_component.wasm", "rb") as f:
118
    component_bytes = f.read()
119

120
# Generate Python bindings
121
bindings = wasmtime.bindgen.generate("my_component", component_bytes)
122

123
# The result is a dictionary of file names to Python code
124
for filename, code_bytes in bindings.items():
125
    print(f"Generated file: {filename}")
126
    if filename.endswith('.py'):
127
        print("Python code preview:")
128
        print(code_bytes.decode('utf-8')[:200] + "...")
129
    print()
130

131
# Save generated bindings to disk
132
import os
133
output_dir = "generated_bindings"
134
os.makedirs(output_dir, exist_ok=True)
135

136
for filename, code_bytes in bindings.items():
137
    filepath = os.path.join(output_dir, filename)
138
    with open(filepath, 'wb') as f:
139
        f.write(code_bytes)
140
    print(f"Saved: {filepath}")
141
```
142

143
### Component Execution with Custom Host Implementations
144

145
```python
146
import wasmtime
147
import wasmtime.bindgen
148
import sys
149
import os
150
from typing import List, Tuple, Optional
151

152
# Custom host implementations for WASI interfaces
153
class MyWasiStreams(wasmtime.bindgen.WasiStreams):
154
    def __init__(self):
155
        self.output_buffer = []
156
        
157
    def write(self, stream_id: int, buf: bytes) -> Tuple[int, int]:
158
        """Write to stream and return (bytes_written, status)"""
159
        if stream_id == 1:  # stdout
160
            sys.stdout.buffer.write(buf)
161
            sys.stdout.buffer.flush()
162
        elif stream_id == 2:  # stderr
163
            sys.stderr.buffer.write(buf)
164
            sys.stderr.buffer.flush()
165
        else:
166
            # Custom stream - store in buffer
167
            self.output_buffer.append((stream_id, buf))
168
        
169
        return (len(buf), 0)  # 0 = stream open status
170
    
171
    def blocking_write(self, stream_id: int, buf: bytes) -> Tuple[int, int]:
172
        return self.write(stream_id, buf)
173

174
class MyWasiEnvironment(wasmtime.bindgen.WasiEnvironment):
175
    def __init__(self, custom_vars: dict = None):
176
        self.custom_vars = custom_vars or {}
177
        
178
    def get_environment(self) -> List[Tuple[str, str]]:
179
        """Return environment variables as (key, value) pairs"""
180
        env_vars = []
181
        
182
        # Add system environment variables (filtered)
183
        for key, value in os.environ.items():
184
            if not key.startswith(('SECRET_', 'TOKEN_')):
185
                env_vars.append((key, value))
186
        
187
        # Add custom variables
188
        for key, value in self.custom_vars.items():
189
            env_vars.append((key, value))
190
            
191
        return env_vars
192

193
class MyWasiRandom(wasmtime.bindgen.WasiRandom):
194
    def get_random_bytes(self, length: int) -> bytes:
195
        """Generate cryptographically secure random bytes"""
196
        return os.urandom(length)
197

198
# Set up component execution
199
def run_component_with_custom_hosts(component_path: str):
200
    """Run a WebAssembly component with custom host implementations"""
201
    
202
    # Load component
203
    with open(component_path, 'rb') as f:
204
        component_bytes = f.read()
205
    
206
    # Generate bindings
207
    bindings = wasmtime.bindgen.generate("component", component_bytes)
208
    
209
    # Initialize component runtime
210
    root, store = wasmtime.bindgen.init()
211
    
212
    # Create custom host implementations
213
    custom_streams = MyWasiStreams()
214
    custom_env = MyWasiEnvironment({
215
        "COMPONENT_MODE": "python_host",
216
        "MAX_BUFFER_SIZE": "1048576"
217
    })
218
    custom_random = MyWasiRandom()
219
    
220
    # Use default implementations for other interfaces
221
    default_types = wasmtime.bindgen.WasiTypes()
222
    default_preopens = wasmtime.bindgen.WasiPreopens()
223
    default_exit = wasmtime.bindgen.WasiExit()
224
    default_stdin = wasmtime.bindgen.WasiStdin()
225
    default_stdout = wasmtime.bindgen.WasiStdout()
226
    default_stderr = wasmtime.bindgen.WasiStderr()
227
    default_terminal_input = wasmtime.bindgen.WasiTerminalInput()
228
    default_terminal_output = wasmtime.bindgen.WasiTerminalOutput()
229
    default_terminal_stdin = wasmtime.bindgen.WasiTerminalStdin()
230
    default_terminal_stdout = wasmtime.bindgen.WasiTerminalStdout()
231
    default_terminal_stderr = wasmtime.bindgen.WasiTerminalStderr()
232
    
233
    # Create root imports with custom implementations
234
    imports = wasmtime.bindgen.RootImports(
235
        custom_streams,
236
        default_types,
237
        default_preopens,
238
        custom_random,
239
        custom_env,
240
        default_exit,
241
        default_stdin,
242
        default_stdout,
243
        default_stderr,
244
        default_terminal_input,
245
        default_terminal_output,
246
        default_terminal_stdin,
247
        default_terminal_stdout,
248
        default_terminal_stderr
249
    )
250
    
251
    # Create component instance
252
    component = wasmtime.bindgen.Root(store, imports)
253
    
254
    print("Component initialized with custom host implementations")
255
    print(f"Custom environment variables: {len(custom_env.custom_vars)}")
256
    
257
    return component, store
258

259
# Example usage
260
# component, store = run_component_with_custom_hosts("my_component.wasm")
261
```
262

263
### Advanced Component Interface Implementation
264

265
```python
266
import wasmtime.bindgen
267
from typing import Dict, Any, List, Tuple
268
import json
269
import logging
270

271
class AdvancedWasiStreams(wasmtime.bindgen.WasiStreams):
272
    """Advanced stream implementation with logging and buffering"""
273
    
274
    def __init__(self):
275
        self.streams: Dict[int, Dict[str, Any]] = {}
276
        self.next_stream_id = 3  # Start after stdout/stderr
277
        self.logger = logging.getLogger(__name__)
278
        
279
    def create_stream(self, name: str, buffer_size: int = 4096) -> int:
280
        """Create a new custom stream"""
281
        stream_id = self.next_stream_id
282
        self.next_stream_id += 1
283
        
284
        self.streams[stream_id] = {
285
            "name": name,
286
            "buffer": bytearray(),
287
            "buffer_size": buffer_size,
288
            "total_written": 0
289
        }
290
        
291
        self.logger.info(f"Created stream {stream_id}: {name}")
292
        return stream_id
293
    
294
    def write(self, stream_id: int, buf: bytes) -> Tuple[int, int]:
295
        """Write to stream with detailed logging"""
296
        self.logger.debug(f"Writing {len(buf)} bytes to stream {stream_id}")
297
        
298
        if stream_id == 1:  # stdout
299
            import sys
300
            sys.stdout.buffer.write(buf)
301
            sys.stdout.buffer.flush()
302
            return (len(buf), 0)
303
            
304
        elif stream_id == 2:  # stderr
305
            import sys
306
            sys.stderr.buffer.write(buf)
307
            sys.stderr.buffer.flush()
308
            return (len(buf), 0)
309
            
310
        elif stream_id in self.streams:
311
            # Custom stream
312
            stream = self.streams[stream_id]
313
            stream["buffer"].extend(buf)
314
            stream["total_written"] += len(buf)
315
            
316
            # Log buffer status
317
            self.logger.debug(f"Stream {stream_id} ({stream['name']}): "
318
                            f"{len(stream['buffer'])} buffered, "
319
                            f"{stream['total_written']} total")
320
            
321
            # Auto-flush if buffer is full
322
            if len(stream["buffer"]) >= stream["buffer_size"]:
323
                self.flush_stream(stream_id)
324
            
325
            return (len(buf), 0)
326
        else:
327
            self.logger.error(f"Unknown stream ID: {stream_id}")
328
            return (0, 1)  # Error status
329
    
330
    def flush_stream(self, stream_id: int):
331
        """Flush a custom stream buffer"""
332
        if stream_id in self.streams:
333
            stream = self.streams[stream_id]
334
            if stream["buffer"]:
335
                self.logger.info(f"Flushing stream {stream_id} ({stream['name']}): "
336
                               f"{len(stream['buffer'])} bytes")
337
                
338
                # Process buffered data (example: save to file)
339
                filename = f"stream_{stream_id}_{stream['name']}.log"
340
                with open(filename, 'ab') as f:
341
                    f.write(stream["buffer"])
342
                
343
                stream["buffer"].clear()
344

345
class ConfigurableWasiEnvironment(wasmtime.bindgen.WasiEnvironment):
346
    """Environment implementation with configuration file support"""
347
    
348
    def __init__(self, config_file: str = None):
349
        self.config = self.load_config(config_file) if config_file else {}
350
        self.logger = logging.getLogger(__name__)
351
        
352
    def load_config(self, config_file: str) -> dict:
353
        """Load environment configuration from JSON file"""
354
        try:
355
            with open(config_file, 'r') as f:
356
                config = json.load(f)
357
            self.logger.info(f"Loaded environment config from {config_file}")
358
            return config
359
        except Exception as e:
360
            self.logger.error(f"Failed to load config {config_file}: {e}")
361
            return {}
362
    
363
    def get_environment(self) -> List[Tuple[str, str]]:
364
        """Get environment variables with configuration override"""
365
        env_vars = []
366
        
367
        # Start with configured base environment
368
        base_env = self.config.get("base_environment", "inherit")
369
        
370
        if base_env == "inherit":
371
            # Inherit from host process
372
            import os
373
            for key, value in os.environ.items():
374
                if self.is_allowed_var(key):
375
                    env_vars.append((key, value))
376
        elif base_env == "minimal":
377
            # Only essential variables
378
            essential_vars = ["HOME", "PATH", "USER", "SHELL"]
379
            import os
380
            for var in essential_vars:
381
                if var in os.environ:
382
                    env_vars.append((var, os.environ[var]))
383
        
384
        # Add custom variables from config
385
        custom_vars = self.config.get("custom_variables", {})
386
        for key, value in custom_vars.items():
387
            env_vars.append((key, str(value)))
388
        
389
        # Add runtime variables
390
        env_vars.extend([
391
            ("WASM_COMPONENT_MODE", "python_bindgen"),
392
            ("COMPONENT_TIMESTAMP", str(int(__import__('time').time()))),
393
            ("PYTHON_VERSION", __import__('sys').version.split()[0])
394
        ])
395
        
396
        self.logger.debug(f"Providing {len(env_vars)} environment variables")
397
        return env_vars
398
    
399
    def is_allowed_var(self, var_name: str) -> bool:
400
        """Check if environment variable is allowed"""
401
        blocked_patterns = self.config.get("blocked_patterns", [
402
            "SECRET_", "TOKEN_", "PASSWORD_", "KEY_", "PRIVATE_"
403
        ])
404
        
405
        return not any(var_name.startswith(pattern) for pattern in blocked_patterns)
406

407
# Complete component setup with advanced implementations
408
def setup_advanced_component(component_path: str, config_file: str = None):
409
    """Set up component with advanced host implementations"""
410
    
411
    # Configure logging
412
    logging.basicConfig(level=logging.INFO)
413
    logger = logging.getLogger(__name__)
414
    
415
    try:
416
        # Load component
417
        with open(component_path, 'rb') as f:
418
            component_bytes = f.read()
419
        
420
        logger.info(f"Loaded component: {len(component_bytes)} bytes")
421
        
422
        # Generate bindings
423
        logger.info("Generating Python bindings...")
424
        bindings = wasmtime.bindgen.generate("advanced_component", component_bytes)
425
        logger.info(f"Generated {len(bindings)} binding files")
426
        
427
        # Initialize runtime
428
        root, store = wasmtime.bindgen.init()
429
        
430
        # Create advanced host implementations
431
        advanced_streams = AdvancedWasiStreams()
432
        configurable_env = ConfigurableWasiEnvironment(config_file)
433
        
434
        # Create additional custom streams
435
        log_stream = advanced_streams.create_stream("application_log")
436
        metrics_stream = advanced_streams.create_stream("metrics")
437
        
438
        logger.info(f"Created custom streams: log={log_stream}, metrics={metrics_stream}")
439
        
440
        # Set up all WASI interfaces
441
        imports = wasmtime.bindgen.RootImports(
442
            advanced_streams,
443
            wasmtime.bindgen.WasiTypes(),
444
            wasmtime.bindgen.WasiPreopens(),
445
            wasmtime.bindgen.WasiRandom(),
446
            configurable_env,
447
            wasmtime.bindgen.WasiExit(),
448
            wasmtime.bindgen.WasiStdin(),
449
            wasmtime.bindgen.WasiStdout(),
450
            wasmtime.bindgen.WasiStderr(),
451
            wasmtime.bindgen.WasiTerminalInput(),
452
            wasmtime.bindgen.WasiTerminalOutput(),
453
            wasmtime.bindgen.WasiTerminalStdin(),
454
            wasmtime.bindgen.WasiTerminalStdout(),
455
            wasmtime.bindgen.WasiTerminalStderr()
456
        )
457
        
458
        # Create component instance
459
        component = wasmtime.bindgen.Root(store, imports)
460
        
461
        logger.info("Component setup complete")
462
        return component, store, advanced_streams
463
        
464
    except Exception as e:
465
        logger.error(f"Component setup failed: {e}")
466
        raise
467

468
# Example configuration file (config.json):
469
example_config = '''
470
{
471
  "base_environment": "minimal",
472
  "custom_variables": {
473
    "COMPONENT_NAME": "advanced_example",
474
    "LOG_LEVEL": "info",
475
    "MAX_MEMORY": "134217728",
476
    "ENABLE_METRICS": "true"
477
  },
478
  "blocked_patterns": [
479
    "SECRET_",
480
    "TOKEN_",
481
    "PASSWORD_",
482
    "API_KEY_",
483
    "PRIVATE_"
484
  ]
485
}
486
'''
487

488
# Save example config
489
# with open("component_config.json", "w") as f:
490
#     f.write(example_config)
491

492
# Example usage:
493
# component, store, streams = setup_advanced_component("my_component.wasm", "component_config.json")
494
```