Package wasmtime

Python bindings for the Wasmtime project

This library binds the Wasmtime project's C API to provide an implementation of a WebAssembly JIT compiler to Python. You can validate, compile, instantiate, and interact with WebAssembly modules via this library.

The API of this library is intended to be very similar to the wasmtime Rust crate, so if you find the docs are lacking here feel free to consult that documentation as well. While not exactly the same the two libraries are intended to be quite similar.

Sub-modules

wasmtime.bindgen
wasmtime.loader

This module is a custom loader for Python which enables importing wasm files directly into Python programs simply through usage of the import

Functions

def wat2wasm(wat: str | bytes) ‑> bytearray
Expand source code
def wat2wasm(wat: typing.Union[str, bytes]) -> bytearray:
    """
    Converts the [WebAssembly Text format][wat] to the binary format.

    This function is intended to be a convenience function for local
    development and you likely don't want to use it extensively in production.
    It's much faster to parse and compile the binary format than it is to
    process the text format.

    Takes a `str` as input, raises an error if it fails to parse, and returns
    a `bytes` if conversion/parsing was successful.

    >>> wat2wasm('(module)')
    bytearray(b'\\x00asm\\x01\\x00\\x00\\x00')

    [wat]: https://webassembly.github.io/spec/core/text/index.html
    """

    if isinstance(wat, str):
        wat = wat.encode('utf8')
    wat_buffer = create_string_buffer(wat)
    wasm = ffi.wasm_byte_vec_t()
    error = ffi.wasmtime_wat2wasm(wat_buffer, len(wat), byref(wasm))
    if error:
        raise WasmtimeError._from_ptr(error)
    else:
        ret = ffi.to_bytes(wasm)
        ffi.wasm_byte_vec_delete(byref(wasm))
        return ret

Converts the WebAssembly Text format to the binary format.

This function is intended to be a convenience function for local development and you likely don't want to use it extensively in production. It's much faster to parse and compile the binary format than it is to process the text format.

Takes a str as input, raises an error if it fails to parse, and returns a bytes if conversion/parsing was successful.

>>> wat2wasm('(module)')
bytearray(b'\x00asm\x01\x00\x00\x00')

Classes

class Caller (ptr: ctypes._Pointer[ffi.wasmtime_caller_t])
Expand source code
class Caller:
    __ptr: "Optional[ctypes._Pointer[ffi.wasmtime_caller_t]]"
    __context: "Optional[ctypes._Pointer[ffi.wasmtime_context_t]]"

    def __init__(self, ptr: "ctypes._Pointer[ffi.wasmtime_caller_t]"):
        self.__ptr = ptr
        self.__context = ffi.wasmtime_caller_context(ptr)

    def __getitem__(self, name: str) -> AsExtern:
        """
        Looks up an export with `name` on the calling module.

        If `name` isn't defined on the calling module, or if the caller has gone
        away for some reason, then this will raise a `KeyError`. For more
        information about when this could fail see the `get` method which
        returns `None` on failure.
        """

        ret = self.get(name)
        if ret is None:
            raise KeyError("failed to find export {}".format(name))
        return ret

    def get(self, name: str) -> Optional[AsExtern]:
        """
        Looks up an export with `name` on the calling module.

        May return `None` if the export isn't found, if it's not a memory (for
        now), or if the caller has gone away and this `Caller` object has
        persisted too long.
        """

        # First convert to a raw name so we can typecheck our argument
        name_bytes = name.encode('utf-8')
        name_buf = ffi.create_string_buffer(name_bytes)

        # Next see if we've been invalidated
        if self.__ptr is None:
            return None

        # And if we're not invalidated we can perform the actual lookup
        item = ffi.wasmtime_extern_t()
        ok = ffi.wasmtime_caller_export_get(self.__ptr, name_buf, len(name_bytes), byref(item))
        if ok:
            return wrap_extern(item)
        else:
            return None

    def _context(self) -> "ctypes._Pointer[ffi.wasmtime_context_t]":
        if self.__context is None:
            raise ValueError("caller is no longer valid")
        return self.__context

    def _invalidate(self) -> None:
        self.__ptr = None
        self.__context = None

Methods

def get(self, name: str) ‑> Func | Table | Memory | SharedMemory | Global | Instance | Module | None
Expand source code
def get(self, name: str) -> Optional[AsExtern]:
    """
    Looks up an export with `name` on the calling module.

    May return `None` if the export isn't found, if it's not a memory (for
    now), or if the caller has gone away and this `Caller` object has
    persisted too long.
    """

    # First convert to a raw name so we can typecheck our argument
    name_bytes = name.encode('utf-8')
    name_buf = ffi.create_string_buffer(name_bytes)

    # Next see if we've been invalidated
    if self.__ptr is None:
        return None

    # And if we're not invalidated we can perform the actual lookup
    item = ffi.wasmtime_extern_t()
    ok = ffi.wasmtime_caller_export_get(self.__ptr, name_buf, len(name_bytes), byref(item))
    if ok:
        return wrap_extern(item)
    else:
        return None

Looks up an export with name on the calling module.

May return None if the export isn't found, if it's not a memory (for now), or if the caller has gone away and this Caller object has persisted too long.

class Config
Expand source code
class Config(Managed["ctypes._Pointer[ffi.wasm_config_t]"]):
    """
    Global configuration, used to create an `Engine`.

    A `Config` houses a number of configuration options which tweaks how wasm
    code is compiled or generated.
    """

    def __init__(self) -> None:
        self._set_ptr(ffi.wasm_config_new())

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

    @setter_property
    def debug_info(self, enable: bool) -> None:
        """
        Configures whether DWARF debug information is emitted for the generated
        code. This can improve profiling and the debugging experience.
        """

        if not isinstance(enable, bool):
            raise TypeError('expected a bool')
        ffi.wasmtime_config_debug_info_set(self.ptr(), enable)

    @setter_property
    def wasm_threads(self, enable: bool) -> None:
        """
        Configures whether the wasm [threads proposal] is enabled.

        [threads proposal]: https://github.com/webassembly/threads
        """

        if not isinstance(enable, bool):
            raise TypeError('expected a bool')
        ffi.wasmtime_config_wasm_threads_set(self.ptr(), enable)

    @setter_property
    def wasm_tail_call(self, enable: bool) -> None:
        """
        Configures whether the wasm [tail call proposal] is enabled.

        [tail call proposal]: https://github.com/WebAssembly/tail-call
        """

        if not isinstance(enable, bool):
            raise TypeError('expected a bool')
        ffi.wasmtime_config_wasm_tail_call_set(self.ptr(), enable)

    @setter_property
    def wasm_reference_types(self, enable: bool) -> None:
        """
        Configures whether the wasm [reference types proposal] is enabled.

        [reference types proposal]: https://github.com/webassembly/reference-types
        """

        if not isinstance(enable, bool):
            raise TypeError('expected a bool')
        ffi.wasmtime_config_wasm_reference_types_set(self.ptr(), enable)

    @setter_property
    def wasm_simd(self, enable: bool) -> None:
        """
        Configures whether the wasm [SIMD proposal] is enabled.

        [SIMD proposal]: https://github.com/webassembly/simd
        """

        if not isinstance(enable, bool):
            raise TypeError('expected a bool')
        ffi.wasmtime_config_wasm_simd_set(self.ptr(), enable)

    @setter_property
    def wasm_bulk_memory(self, enable: bool) -> None:
        """
        Configures whether the wasm [bulk memory proposal] is enabled.

        [bulk memory proposal]: https://github.com/webassembly/bulk-memory
        """

        if not isinstance(enable, bool):
            raise TypeError('expected a bool')
        ffi.wasmtime_config_wasm_bulk_memory_set(self.ptr(), enable)

    @setter_property
    def wasm_multi_value(self, enable: bool) -> None:
        """
        Configures whether the wasm [multi value proposal] is enabled.

        [multi value proposal]: https://github.com/webassembly/multi-value
        """

        if not isinstance(enable, bool):
            raise TypeError('expected a bool')
        ffi.wasmtime_config_wasm_multi_value_set(self.ptr(), enable)

    @setter_property
    def wasm_multi_memory(self, enable: bool) -> None:
        """
        Configures whether the wasm [multi memory proposal] is enabled.

        [multi memory proposal]: https://github.com/webassembly/multi-memory
        """

        if not isinstance(enable, bool):
            raise TypeError('expected a bool')
        ffi.wasmtime_config_wasm_multi_memory_set(self.ptr(), enable)

    @setter_property
    def wasm_memory64(self, enable: bool) -> None:
        """
        Configures whether the wasm [memory64 proposal] is enabled.

        [memory64 proposal]: https://github.com/webassembly/memory64
        """

        if not isinstance(enable, bool):
            raise TypeError('expected a bool')
        ffi.wasmtime_config_wasm_memory64_set(self.ptr(), enable)

    @setter_property
    def wasm_relaxed_simd(self, enable: bool) -> None:
        """
        Configures whether the wasm [relaxed simd proposal] is enabled.

        [relaxed simd proposal]: https://github.com/webassembly/relaxed-simd
        """

        if not isinstance(enable, bool):
            raise TypeError('expected a bool')
        ffi.wasmtime_config_wasm_relaxed_simd_set(self.ptr(), enable)

    @setter_property
    def wasm_relaxed_simd_deterministic(self, enable: bool) -> None:
        """
        Configures whether the wasm [relaxed simd proposal] is deterministic
        in is execution as opposed to having the most optimal implementation for
        the current platform.

        [relaxed simd proposal]: https://github.com/webassembly/relaxed-simd
        """

        if not isinstance(enable, bool):
            raise TypeError('expected a bool')
        ffi.wasmtime_config_wasm_relaxed_simd_deterministic_set(self.ptr(), enable)

    @setter_property
    def strategy(self, strategy: str) -> None:
        """
        Configures the compilation strategy used for wasm code.

        Acceptable values for `strategy` are:

        * `"auto"`
        * `"cranelift"`
        """

        if strategy == "auto":
            ffi.wasmtime_config_strategy_set(self.ptr(), 0)
        elif strategy == "cranelift":
            ffi.wasmtime_config_strategy_set(self.ptr(), 1)
        else:
            raise WasmtimeError("unknown strategy: " + str(strategy))

    @setter_property
    def cranelift_debug_verifier(self, enable: bool) -> None:
        if not isinstance(enable, bool):
            raise TypeError('expected a bool')
        ffi.wasmtime_config_cranelift_debug_verifier_set(self.ptr(), enable)

    @setter_property
    def cranelift_opt_level(self, opt_level: str) -> None:
        if opt_level == "none":
            ffi.wasmtime_config_cranelift_opt_level_set(self.ptr(), 0)
        elif opt_level == "speed":
            ffi.wasmtime_config_cranelift_opt_level_set(self.ptr(), 1)
        elif opt_level == "speed_and_size":
            ffi.wasmtime_config_cranelift_opt_level_set(self.ptr(), 2)
        else:
            raise WasmtimeError("unknown opt level: " + str(opt_level))

    @setter_property
    def profiler(self, profiler: str) -> None:
        if profiler == "none":
            ffi.wasmtime_config_profiler_set(self.ptr(), 0)
        elif profiler == "jitdump":
            ffi.wasmtime_config_profiler_set(self.ptr(), 1)
        else:
            raise WasmtimeError("unknown profiler: " + str(profiler))

    @setter_property
    def cache(self, enabled: typing.Union[bool, str]) -> None:
        """
        Configures whether code caching is enabled for this `Config`.

        The value `True` can be passed in here to enable the default caching
        configuration and location, or a path to a file can be passed in which
        is a path to a TOML configuration file for the cache.

        More information about cache configuration can be found at
        https://bytecodealliance.github.io/wasmtime/cli-cache.html
        """

        if isinstance(enabled, bool):
            if not enabled:
                raise WasmtimeError("caching cannot be explicitly disabled")
            error = ffi.wasmtime_config_cache_config_load(self.ptr(), None)
        elif isinstance(enabled, str):
            error = ffi.wasmtime_config_cache_config_load(self.ptr(),
                                                          c_char_p(enabled.encode('utf-8')))
        else:
            raise TypeError("expected string or bool")
        if error:
            raise WasmtimeError._from_ptr(error)

    @setter_property
    def epoch_interruption(self, enabled: bool) -> None:
        """
        Configures whether wasm execution can be interrupted via epoch
        increments.
        """

        if enabled:
            val = 1
        else:
            val = 0
        ffi.wasmtime_config_epoch_interruption_set(self.ptr(), val)

    @setter_property
    def consume_fuel(self, instances: bool) -> None:
        """
        Configures whether wasm code will consume *fuel* as part of its
        execution.

        Fuel consumption allows WebAssembly to trap when fuel runs out.
        Currently stores start with 0 fuel if this is enabled.
        """
        if not isinstance(instances, bool):
            raise TypeError('expected an bool')
        ffi.wasmtime_config_consume_fuel_set(self.ptr(), instances)

    @setter_property
    def parallel_compilation(self, enable: bool) -> None:
        """
        Configures whether parallel compilation is enabled for functions
        within a module.

        This is enabled by default.
        """
        if not isinstance(enable, bool):
            raise TypeError('expected a bool')
        ffi.wasmtime_config_parallel_compilation_set(self.ptr(), enable)

Global configuration, used to create an Engine.

A Config houses a number of configuration options which tweaks how wasm code is compiled or generated.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Instance variables

prop cache
Expand source code
@setter_property
def cache(self, enabled: typing.Union[bool, str]) -> None:
    """
    Configures whether code caching is enabled for this `Config`.

    The value `True` can be passed in here to enable the default caching
    configuration and location, or a path to a file can be passed in which
    is a path to a TOML configuration file for the cache.

    More information about cache configuration can be found at
    https://bytecodealliance.github.io/wasmtime/cli-cache.html
    """

    if isinstance(enabled, bool):
        if not enabled:
            raise WasmtimeError("caching cannot be explicitly disabled")
        error = ffi.wasmtime_config_cache_config_load(self.ptr(), None)
    elif isinstance(enabled, str):
        error = ffi.wasmtime_config_cache_config_load(self.ptr(),
                                                      c_char_p(enabled.encode('utf-8')))
    else:
        raise TypeError("expected string or bool")
    if error:
        raise WasmtimeError._from_ptr(error)

Configures whether code caching is enabled for this Config.

The value True can be passed in here to enable the default caching configuration and location, or a path to a file can be passed in which is a path to a TOML configuration file for the cache.

More information about cache configuration can be found at https://bytecodealliance.github.io/wasmtime/cli-cache.html

Note that this field can only be set, it cannot be read

prop consume_fuel
Expand source code
@setter_property
def consume_fuel(self, instances: bool) -> None:
    """
    Configures whether wasm code will consume *fuel* as part of its
    execution.

    Fuel consumption allows WebAssembly to trap when fuel runs out.
    Currently stores start with 0 fuel if this is enabled.
    """
    if not isinstance(instances, bool):
        raise TypeError('expected an bool')
    ffi.wasmtime_config_consume_fuel_set(self.ptr(), instances)

Configures whether wasm code will consume fuel as part of its execution.

Fuel consumption allows WebAssembly to trap when fuel runs out. Currently stores start with 0 fuel if this is enabled.

Note that this field can only be set, it cannot be read

prop cranelift_debug_verifier
Expand source code
@setter_property
def cranelift_debug_verifier(self, enable: bool) -> None:
    if not isinstance(enable, bool):
        raise TypeError('expected a bool')
    ffi.wasmtime_config_cranelift_debug_verifier_set(self.ptr(), enable)
prop cranelift_opt_level
Expand source code
@setter_property
def cranelift_opt_level(self, opt_level: str) -> None:
    if opt_level == "none":
        ffi.wasmtime_config_cranelift_opt_level_set(self.ptr(), 0)
    elif opt_level == "speed":
        ffi.wasmtime_config_cranelift_opt_level_set(self.ptr(), 1)
    elif opt_level == "speed_and_size":
        ffi.wasmtime_config_cranelift_opt_level_set(self.ptr(), 2)
    else:
        raise WasmtimeError("unknown opt level: " + str(opt_level))
prop debug_info
Expand source code
@setter_property
def debug_info(self, enable: bool) -> None:
    """
    Configures whether DWARF debug information is emitted for the generated
    code. This can improve profiling and the debugging experience.
    """

    if not isinstance(enable, bool):
        raise TypeError('expected a bool')
    ffi.wasmtime_config_debug_info_set(self.ptr(), enable)

Configures whether DWARF debug information is emitted for the generated code. This can improve profiling and the debugging experience.

Note that this field can only be set, it cannot be read

prop epoch_interruption
Expand source code
@setter_property
def epoch_interruption(self, enabled: bool) -> None:
    """
    Configures whether wasm execution can be interrupted via epoch
    increments.
    """

    if enabled:
        val = 1
    else:
        val = 0
    ffi.wasmtime_config_epoch_interruption_set(self.ptr(), val)

Configures whether wasm execution can be interrupted via epoch increments.

Note that this field can only be set, it cannot be read

prop parallel_compilation
Expand source code
@setter_property
def parallel_compilation(self, enable: bool) -> None:
    """
    Configures whether parallel compilation is enabled for functions
    within a module.

    This is enabled by default.
    """
    if not isinstance(enable, bool):
        raise TypeError('expected a bool')
    ffi.wasmtime_config_parallel_compilation_set(self.ptr(), enable)

Configures whether parallel compilation is enabled for functions within a module.

This is enabled by default.

Note that this field can only be set, it cannot be read

prop profiler
Expand source code
@setter_property
def profiler(self, profiler: str) -> None:
    if profiler == "none":
        ffi.wasmtime_config_profiler_set(self.ptr(), 0)
    elif profiler == "jitdump":
        ffi.wasmtime_config_profiler_set(self.ptr(), 1)
    else:
        raise WasmtimeError("unknown profiler: " + str(profiler))
prop strategy
Expand source code
@setter_property
def strategy(self, strategy: str) -> None:
    """
    Configures the compilation strategy used for wasm code.

    Acceptable values for `strategy` are:

    * `"auto"`
    * `"cranelift"`
    """

    if strategy == "auto":
        ffi.wasmtime_config_strategy_set(self.ptr(), 0)
    elif strategy == "cranelift":
        ffi.wasmtime_config_strategy_set(self.ptr(), 1)
    else:
        raise WasmtimeError("unknown strategy: " + str(strategy))

Configures the compilation strategy used for wasm code.

Acceptable values for strategy are:

  • "auto"
  • "cranelift"

Note that this field can only be set, it cannot be read

prop wasm_bulk_memory
Expand source code
@setter_property
def wasm_bulk_memory(self, enable: bool) -> None:
    """
    Configures whether the wasm [bulk memory proposal] is enabled.

    [bulk memory proposal]: https://github.com/webassembly/bulk-memory
    """

    if not isinstance(enable, bool):
        raise TypeError('expected a bool')
    ffi.wasmtime_config_wasm_bulk_memory_set(self.ptr(), enable)

Configures whether the wasm bulk memory proposal is enabled.

Note that this field can only be set, it cannot be read

prop wasm_memory64
Expand source code
@setter_property
def wasm_memory64(self, enable: bool) -> None:
    """
    Configures whether the wasm [memory64 proposal] is enabled.

    [memory64 proposal]: https://github.com/webassembly/memory64
    """

    if not isinstance(enable, bool):
        raise TypeError('expected a bool')
    ffi.wasmtime_config_wasm_memory64_set(self.ptr(), enable)

Configures whether the wasm memory64 proposal is enabled.

Note that this field can only be set, it cannot be read

prop wasm_multi_memory
Expand source code
@setter_property
def wasm_multi_memory(self, enable: bool) -> None:
    """
    Configures whether the wasm [multi memory proposal] is enabled.

    [multi memory proposal]: https://github.com/webassembly/multi-memory
    """

    if not isinstance(enable, bool):
        raise TypeError('expected a bool')
    ffi.wasmtime_config_wasm_multi_memory_set(self.ptr(), enable)

Configures whether the wasm multi memory proposal is enabled.

Note that this field can only be set, it cannot be read

prop wasm_multi_value
Expand source code
@setter_property
def wasm_multi_value(self, enable: bool) -> None:
    """
    Configures whether the wasm [multi value proposal] is enabled.

    [multi value proposal]: https://github.com/webassembly/multi-value
    """

    if not isinstance(enable, bool):
        raise TypeError('expected a bool')
    ffi.wasmtime_config_wasm_multi_value_set(self.ptr(), enable)

Configures whether the wasm multi value proposal is enabled.

Note that this field can only be set, it cannot be read

prop wasm_reference_types
Expand source code
@setter_property
def wasm_reference_types(self, enable: bool) -> None:
    """
    Configures whether the wasm [reference types proposal] is enabled.

    [reference types proposal]: https://github.com/webassembly/reference-types
    """

    if not isinstance(enable, bool):
        raise TypeError('expected a bool')
    ffi.wasmtime_config_wasm_reference_types_set(self.ptr(), enable)

Configures whether the wasm reference types proposal is enabled.

Note that this field can only be set, it cannot be read

prop wasm_relaxed_simd
Expand source code
@setter_property
def wasm_relaxed_simd(self, enable: bool) -> None:
    """
    Configures whether the wasm [relaxed simd proposal] is enabled.

    [relaxed simd proposal]: https://github.com/webassembly/relaxed-simd
    """

    if not isinstance(enable, bool):
        raise TypeError('expected a bool')
    ffi.wasmtime_config_wasm_relaxed_simd_set(self.ptr(), enable)

Configures whether the wasm relaxed simd proposal is enabled.

Note that this field can only be set, it cannot be read

prop wasm_relaxed_simd_deterministic
Expand source code
@setter_property
def wasm_relaxed_simd_deterministic(self, enable: bool) -> None:
    """
    Configures whether the wasm [relaxed simd proposal] is deterministic
    in is execution as opposed to having the most optimal implementation for
    the current platform.

    [relaxed simd proposal]: https://github.com/webassembly/relaxed-simd
    """

    if not isinstance(enable, bool):
        raise TypeError('expected a bool')
    ffi.wasmtime_config_wasm_relaxed_simd_deterministic_set(self.ptr(), enable)

Configures whether the wasm relaxed simd proposal is deterministic in is execution as opposed to having the most optimal implementation for the current platform.

Note that this field can only be set, it cannot be read

prop wasm_simd
Expand source code
@setter_property
def wasm_simd(self, enable: bool) -> None:
    """
    Configures whether the wasm [SIMD proposal] is enabled.

    [SIMD proposal]: https://github.com/webassembly/simd
    """

    if not isinstance(enable, bool):
        raise TypeError('expected a bool')
    ffi.wasmtime_config_wasm_simd_set(self.ptr(), enable)

Configures whether the wasm SIMD proposal is enabled.

Note that this field can only be set, it cannot be read

prop wasm_tail_call
Expand source code
@setter_property
def wasm_tail_call(self, enable: bool) -> None:
    """
    Configures whether the wasm [tail call proposal] is enabled.

    [tail call proposal]: https://github.com/WebAssembly/tail-call
    """

    if not isinstance(enable, bool):
        raise TypeError('expected a bool')
    ffi.wasmtime_config_wasm_tail_call_set(self.ptr(), enable)

Configures whether the wasm tail call proposal is enabled.

Note that this field can only be set, it cannot be read

prop wasm_threads
Expand source code
@setter_property
def wasm_threads(self, enable: bool) -> None:
    """
    Configures whether the wasm [threads proposal] is enabled.

    [threads proposal]: https://github.com/webassembly/threads
    """

    if not isinstance(enable, bool):
        raise TypeError('expected a bool')
    ffi.wasmtime_config_wasm_threads_set(self.ptr(), enable)

Configures whether the wasm threads proposal is enabled.

Note that this field can only be set, it cannot be read

class DirPerms (*args, **kwds)
Expand source code
class DirPerms(Enum):
    READ_ONLY = ffi.wasi_dir_perms_flags.WASMTIME_WASI_DIR_PERMS_READ.value
    WRITE_ONLY = ffi.wasi_dir_perms_flags.WASMTIME_WASI_DIR_PERMS_WRITE.value
    READ_WRITE = ffi.wasi_dir_perms_flags.WASMTIME_WASI_DIR_PERMS_READ.value | ffi.wasi_dir_perms_flags.WASMTIME_WASI_DIR_PERMS_WRITE.value

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

  • attribute access:

Color.RED

  • value lookup:

Color(1)

  • name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3
>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

  • enum.Enum

Class variables

var READ_ONLY
var READ_WRITE
var WRITE_ONLY
class Engine (config: wasmtime._config.Config | None = None)
Expand source code
class Engine(Managed["ctypes._Pointer[ffi.wasm_engine_t]"]):

    def __init__(self, config: Optional[Config] = None):
        if config is None:
            self._set_ptr(ffi.wasm_engine_new())
        elif not isinstance(config, Config):
            raise TypeError("expected Config")
        else:
            ptr = config._consume()
            self._set_ptr(ffi.wasm_engine_new_with_config(ptr))

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

    def increment_epoch(self) -> None:
        ffi.wasmtime_engine_increment_epoch(self.ptr())

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Methods

def increment_epoch(self) ‑> None
Expand source code
def increment_epoch(self) -> None:
    ffi.wasmtime_engine_increment_epoch(self.ptr())
class ExitTrap (message: str)
Expand source code
class ExitTrap(WasmtimeError):
    """
    A special type of `WasmtimeError` which represents the process exiting via
    WASI's `proc_exit` function call.

    Whenever a WASI program exits via `proc_exit` a trap is raised, but the
    trap will have this type instead of `WasmtimeError`, so you can catch just
    this type instead of all traps (if desired). Exit traps have a `code`
    associated with them which is the exit code provided at exit.

    Note that `ExitTrap` is a subclass of `WasmtimeError`, so if you catch a
    trap you'll also catch `ExitTrap`.
    """
    code: int
    pass

A special type of WasmtimeError which represents the process exiting via WASI's proc_exit function call.

Whenever a WASI program exits via proc_exit a trap is raised, but the trap will have this type instead of WasmtimeError, so you can catch just this type instead of all traps (if desired). Exit traps have a code associated with them which is the exit code provided at exit.

Note that ExitTrap is a subclass of WasmtimeError, so if you catch a trap you'll also catch ExitTrap.

Ancestors

  • wasmtime._error.WasmtimeError
  • builtins.Exception
  • builtins.BaseException
  • wasmtime._managed.Managed
  • typing.Generic

Class variables

var code : int
class ExportType
Expand source code
class ExportType(Managed["ctypes._Pointer[ffi.wasm_exporttype_t]"]):
    _ptr: "ctypes._Pointer[ffi.wasm_exporttype_t]"
    _owner: Optional[Any]

    @classmethod
    def _from_ptr(cls, ptr: 'ctypes._Pointer[ffi.wasm_exporttype_t]', owner: Optional[Any]) -> "ExportType":
        if not isinstance(ptr, POINTER(ffi.wasm_exporttype_t)):
            raise TypeError("wrong pointer type")
        ty: "ExportType" = cls.__new__(cls)
        ty._set_ptr(ptr)
        ty._owner = owner
        return ty

    def _delete(self, ptr: "ctypes._Pointer[ffi.wasm_exporttype_t]") -> None:
        if self._owner is None:
            ffi.wasm_exporttype_delete(ptr)

    @property
    def name(self) -> str:
        """
        Returns the name in the modulethis export type refers to
        """
        return ffi.to_str(ffi.wasm_exporttype_name(self.ptr()).contents)

    @property
    def type(self) -> "AsExternType":
        """
        Returns the type that this export refers to
        """
        ptr = ffi.wasm_exporttype_type(self.ptr())
        return wrap_externtype(ptr, self._owner or self)

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Instance variables

prop name : str
Expand source code
@property
def name(self) -> str:
    """
    Returns the name in the modulethis export type refers to
    """
    return ffi.to_str(ffi.wasm_exporttype_name(self.ptr()).contents)

Returns the name in the modulethis export type refers to

prop type : AsExternType
Expand source code
@property
def type(self) -> "AsExternType":
    """
    Returns the type that this export refers to
    """
    ptr = ffi.wasm_exporttype_type(self.ptr())
    return wrap_externtype(ptr, self._owner or self)

Returns the type that this export refers to

class FilePerms (*args, **kwds)
Expand source code
class FilePerms(Enum):
    READ_ONLY = ffi.wasi_file_perms_flags.WASMTIME_WASI_FILE_PERMS_READ.value
    WRITE_ONLY = ffi.wasi_file_perms_flags.WASMTIME_WASI_FILE_PERMS_WRITE.value
    READ_WRITE = ffi.wasi_file_perms_flags.WASMTIME_WASI_FILE_PERMS_READ.value | ffi.wasi_file_perms_flags.WASMTIME_WASI_FILE_PERMS_WRITE.value

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

  • attribute access:

Color.RED

  • value lookup:

Color(1)

  • name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3
>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

  • enum.Enum

Class variables

var READ_ONLY
var READ_WRITE
var WRITE_ONLY
class Frame
Expand source code
class Frame(Managed["ctypes._Pointer[ffi.wasm_frame_t]"]):
    _owner: Optional[Any]

    @classmethod
    def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasm_frame_t]", owner: Optional[Any]) -> "Frame":
        if not isinstance(ptr, POINTER(ffi.wasm_frame_t)):
            raise TypeError("wrong pointer type")
        ty: "Frame" = cls.__new__(cls)
        ty._set_ptr(ptr)
        ty._owner = owner
        return ty

    def _delete(self, ptr: "ctypes._Pointer[ffi.wasm_frame_t]") -> None:
        if self._owner is None:
            ffi.wasm_frame_delete(ptr)

    @property
    def func_index(self) -> int:
        """
        Returns the function index this frame corresponds to in its wasm module
        """

        return ffi.wasm_frame_func_index(self.ptr())

    @property
    def func_name(self) -> Optional[str]:
        """
        Returns the name of the function this frame corresponds to

        May return `None` if no name can be inferred
        """

        ptr = ffi.wasmtime_frame_func_name(self.ptr())
        if ptr:
            return ffi.to_str(ptr.contents)
        else:
            return None

    @property
    def module_name(self) -> Optional[str]:
        """
        Returns the name of the module this frame corresponds to

        May return `None` if no name can be inferred
        """

        ptr = ffi.wasmtime_frame_module_name(self.ptr())
        if ptr:
            return ffi.to_str(ptr.contents)
        else:
            return None

    @property
    def module_offset(self) -> int:
        """
        Returns the offset of this frame's program counter into the original
        wasm source module.
        """

        return ffi.wasm_frame_module_offset(self.ptr())

    @property
    def func_offset(self) -> int:
        """
        Returns the offset of this frame's program counter into the original
        wasm function.
        """

        return ffi.wasm_frame_func_offset(self.ptr())

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Instance variables

prop func_index : int
Expand source code
@property
def func_index(self) -> int:
    """
    Returns the function index this frame corresponds to in its wasm module
    """

    return ffi.wasm_frame_func_index(self.ptr())

Returns the function index this frame corresponds to in its wasm module

prop func_name : str | None
Expand source code
@property
def func_name(self) -> Optional[str]:
    """
    Returns the name of the function this frame corresponds to

    May return `None` if no name can be inferred
    """

    ptr = ffi.wasmtime_frame_func_name(self.ptr())
    if ptr:
        return ffi.to_str(ptr.contents)
    else:
        return None

Returns the name of the function this frame corresponds to

May return None if no name can be inferred

prop func_offset : int
Expand source code
@property
def func_offset(self) -> int:
    """
    Returns the offset of this frame's program counter into the original
    wasm function.
    """

    return ffi.wasm_frame_func_offset(self.ptr())

Returns the offset of this frame's program counter into the original wasm function.

prop module_name : str | None
Expand source code
@property
def module_name(self) -> Optional[str]:
    """
    Returns the name of the module this frame corresponds to

    May return `None` if no name can be inferred
    """

    ptr = ffi.wasmtime_frame_module_name(self.ptr())
    if ptr:
        return ffi.to_str(ptr.contents)
    else:
        return None

Returns the name of the module this frame corresponds to

May return None if no name can be inferred

prop module_offset : int
Expand source code
@property
def module_offset(self) -> int:
    """
    Returns the offset of this frame's program counter into the original
    wasm source module.
    """

    return ffi.wasm_frame_module_offset(self.ptr())

Returns the offset of this frame's program counter into the original wasm source module.

class Func (store: wasmtime._store.Store | ForwardRef('Caller'),
ty: wasmtime._types.FuncType,
func: Callable,
access_caller: bool = False)
Expand source code
class Func:
    _func: ffi.wasmtime_func_t

    def __init__(self, store: Storelike, ty: FuncType, func: Callable, access_caller: bool = False):
        """
        Creates a new func in `store` with the given `ty` which calls the closure
        given

        The `func` is called with the parameters natively and they'll have native
        Python values rather than being wrapped in `Val`. If `access_caller` is
        set to `True` then the first argument given to `func` is an instance of
        type `Caller` below.
        """

        if not isinstance(store, Store):
            raise TypeError("expected a Store")
        if not isinstance(ty, FuncType):
            raise TypeError("expected a FuncType")
        idx = FUNCTIONS.allocate((func, ty.results, access_caller))
        _func = ffi.wasmtime_func_t()
        ffi.wasmtime_func_new(
            store._context(),
            ty.ptr(),
            trampoline,
            idx,
            finalize,
            byref(_func))
        self._func = _func

    @classmethod
    def _from_raw(cls, func: ffi.wasmtime_func_t) -> "Func":
        ty: "Func" = cls.__new__(cls)
        ty._func = func
        return ty

    def type(self, store: Storelike) -> FuncType:
        """
        Gets the type of this func as a `FuncType`
        """
        ptr = ffi.wasmtime_func_type(store._context(), byref(self._func))
        return FuncType._from_ptr(ptr, None)

    def __call__(self, store: Storelike, *params: Any) -> Any:
        """
        Calls this function with the given parameters

        Parameters can either be a `Val` or a native python value which can be
        converted to a `Val` of the corresponding correct type

        Returns `None` if this func has 0 return types
        Returns a single value if the func has 1 return type
        Returns a list if the func has more than 1 return type

        Note that you can also use the `__call__` method and invoke a `Func` as
        if it were a function directly.
        """

        ty = self.type(store)
        param_tys = ty.params
        if len(params) > len(param_tys):
            raise WasmtimeError("too many parameters provided: given %s, expected %s" %
                                (len(params), len(param_tys)))
        if len(params) < len(param_tys):
            raise WasmtimeError("too few parameters provided: given %s, expected %s" %
                                (len(params), len(param_tys)))

        params_ptr = (ffi.wasmtime_val_t * len(params))()
        params_set = 0
        try:
            for val in params:
                params_ptr[params_set] = Val._convert_to_raw(store, param_tys[params_set], val)
                params_set += 1

            result_tys = ty.results
            results_ptr = (ffi.wasmtime_val_t * len(result_tys))()

            with enter_wasm(store) as trap:
                error = ffi.wasmtime_func_call(
                    store._context(),
                    byref(self._func),
                    params_ptr,
                    len(params),
                    results_ptr,
                    len(result_tys),
                    trap)
                if error:
                    raise WasmtimeError._from_ptr(error)
        finally:
            for i in range(0, params_set):
                ffi.wasmtime_val_unroot(store._context(), byref(params_ptr[i]))

        results = []
        for i in range(0, len(result_tys)):
            results.append(Val._from_raw(store, results_ptr[i]).value)
        if len(results) == 0:
            return None
        elif len(results) == 1:
            return results[0]
        else:
            return results

    def _as_extern(self) -> ffi.wasmtime_extern_t:
        union = ffi.wasmtime_extern_union(func=self._func)
        return ffi.wasmtime_extern_t(ffi.WASMTIME_EXTERN_FUNC, union)

Creates a new func in store with the given ty which calls the closure given

The func is called with the parameters natively and they'll have native Python values rather than being wrapped in Val. If access_caller is set to True then the first argument given to func is an instance of type Caller below.

Methods

def type(self,
store: wasmtime._store.Store | ForwardRef('Caller')) ‑> wasmtime._types.FuncType
Expand source code
def type(self, store: Storelike) -> FuncType:
    """
    Gets the type of this func as a `FuncType`
    """
    ptr = ffi.wasmtime_func_type(store._context(), byref(self._func))
    return FuncType._from_ptr(ptr, None)

Gets the type of this func as a FuncType

class FuncType (params: List[wasmtime._types.ValType], results: List[wasmtime._types.ValType])
Expand source code
class FuncType(Managed["ctypes._Pointer[ffi.wasm_functype_t]"]):
    def __init__(self, params: List[ValType], results: List[ValType]):
        for param in params:
            if not isinstance(param, ValType):
                raise TypeError("expected ValType")
        for result in results:
            if not isinstance(result, ValType):
                raise TypeError("expected ValType")

        params_ffi = ffi.wasm_valtype_vec_t()
        ffi.wasm_valtype_vec_new_uninitialized(byref(params_ffi), len(params))

        results_ffi = ffi.wasm_valtype_vec_t()
        for i, param in enumerate(params):
            params_ffi.data[i] = take_owned_valtype(param)

        ffi.wasm_valtype_vec_new_uninitialized(
            byref(results_ffi), len(results))
        for i, result in enumerate(results):
            results_ffi.data[i] = take_owned_valtype(result)
        ptr = ffi.wasm_functype_new(byref(params_ffi), byref(results_ffi))
        if not ptr:
            raise WasmtimeError("failed to allocate FuncType")
        self._set_ptr(ptr)
        self._owner = None

    def _delete(self, ptr: "ctypes._Pointer[ffi.wasm_functype_t]") -> None:
        if self._owner is None:
            ffi.wasm_functype_delete(ptr)

    @classmethod
    def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasm_functype_t]", owner: Optional[Any]) -> "FuncType":
        if not isinstance(ptr, POINTER(ffi.wasm_functype_t)):
            raise TypeError("wrong pointer type")
        ty: "FuncType" = cls.__new__(cls)
        ty._set_ptr(ptr)
        ty._owner = owner
        return ty

    @property
    def params(self) -> List["ValType"]:
        """
        Returns the list of parameter types for this function type
        """

        ptr = ffi.wasm_functype_params(self.ptr())
        return ValType._from_list(ptr, self)

    @property
    def results(self) -> List["ValType"]:
        """
        Returns the list of result types for this function type
        """

        ptr = ffi.wasm_functype_results(self.ptr())
        return ValType._from_list(ptr, self)

    def _as_extern(self) -> "ctypes._Pointer[ffi.wasm_externtype_t]":
        return ffi.wasm_functype_as_externtype_const(self.ptr())

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Instance variables

prop params : List[ValType]
Expand source code
@property
def params(self) -> List["ValType"]:
    """
    Returns the list of parameter types for this function type
    """

    ptr = ffi.wasm_functype_params(self.ptr())
    return ValType._from_list(ptr, self)

Returns the list of parameter types for this function type

prop results : List[ValType]
Expand source code
@property
def results(self) -> List["ValType"]:
    """
    Returns the list of result types for this function type
    """

    ptr = ffi.wasm_functype_results(self.ptr())
    return ValType._from_list(ptr, self)

Returns the list of result types for this function type

class Global (store: wasmtime._store.Store | ForwardRef('Caller'),
ty: wasmtime._types.GlobalType,
val: Any)
Expand source code
class Global:
    _global: ffi.wasmtime_global_t

    def __init__(self, store: Storelike, ty: GlobalType, val: Any):
        if not isinstance(ty, GlobalType):
            raise TypeError("expected a GlobalType")
        val = Val._convert_to_raw(store, ty.content, val)
        global_ = ffi.wasmtime_global_t()
        error = ffi.wasmtime_global_new(
            store._context(),
            ty.ptr(),
            byref(val),
            byref(global_))
        ffi.wasmtime_val_unroot(store._context(), byref(val))
        if error:
            raise WasmtimeError._from_ptr(error)
        self._global = global_

    @classmethod
    def _from_raw(cls, global_: ffi.wasmtime_global_t) -> "Global":
        ty: "Global" = cls.__new__(cls)
        ty._global = global_
        return ty

    def type(self, store: Storelike) -> GlobalType:
        """
        Gets the type of this global as a `GlobalType`
        """

        ptr = ffi.wasmtime_global_type(store._context(), byref(self._global))
        return GlobalType._from_ptr(ptr, None)

    def value(self, store: Storelike) -> Any:
        """
        Gets the current value of this global

        Returns a native python type
        """
        raw = ffi.wasmtime_val_t()
        ffi.wasmtime_global_get(store._context(), byref(self._global), byref(raw))
        val = Val._from_raw(store, raw)
        if val.value is not None:
            return val.value
        else:
            return val

    def set_value(self, store: Storelike, val: Any) -> None:
        """
        Sets the value of this global to a new value
        """
        val = Val._convert_to_raw(store, self.type(store).content, val)
        error = ffi.wasmtime_global_set(store._context(), byref(self._global), byref(val))
        ffi.wasmtime_val_unroot(store._context(), byref(val))
        if error:
            raise WasmtimeError._from_ptr(error)

    def _as_extern(self) -> ffi.wasmtime_extern_t:
        union = ffi.wasmtime_extern_union(global_=self._global)
        return ffi.wasmtime_extern_t(ffi.WASMTIME_EXTERN_GLOBAL, union)

Methods

def set_value(self,
store: wasmtime._store.Store | ForwardRef('Caller'),
val: Any) ‑> None
Expand source code
def set_value(self, store: Storelike, val: Any) -> None:
    """
    Sets the value of this global to a new value
    """
    val = Val._convert_to_raw(store, self.type(store).content, val)
    error = ffi.wasmtime_global_set(store._context(), byref(self._global), byref(val))
    ffi.wasmtime_val_unroot(store._context(), byref(val))
    if error:
        raise WasmtimeError._from_ptr(error)

Sets the value of this global to a new value

def type(self,
store: wasmtime._store.Store | ForwardRef('Caller')) ‑> wasmtime._types.GlobalType
Expand source code
def type(self, store: Storelike) -> GlobalType:
    """
    Gets the type of this global as a `GlobalType`
    """

    ptr = ffi.wasmtime_global_type(store._context(), byref(self._global))
    return GlobalType._from_ptr(ptr, None)

Gets the type of this global as a GlobalType

def value(self,
store: wasmtime._store.Store | ForwardRef('Caller')) ‑> Any
Expand source code
def value(self, store: Storelike) -> Any:
    """
    Gets the current value of this global

    Returns a native python type
    """
    raw = ffi.wasmtime_val_t()
    ffi.wasmtime_global_get(store._context(), byref(self._global), byref(raw))
    val = Val._from_raw(store, raw)
    if val.value is not None:
        return val.value
    else:
        return val

Gets the current value of this global

Returns a native python type

class GlobalType (valtype: wasmtime._types.ValType, mutable: bool)
Expand source code
class GlobalType(Managed["ctypes._Pointer[ffi.wasm_globaltype_t]"]):
    def __init__(self, valtype: ValType, mutable: bool):
        if mutable:
            mutability = ffi.WASM_VAR
        else:
            mutability = ffi.WASM_CONST
        type_ptr = take_owned_valtype(valtype)
        ptr = ffi.wasm_globaltype_new(type_ptr, mutability)
        if ptr == 0:
            raise WasmtimeError("failed to allocate GlobalType")
        self._set_ptr(ptr)
        self._owner = None

    def _delete(self, ptr: "ctypes._Pointer[ffi.wasm_globaltype_t]") -> None:
        if self._owner is None:
            ffi.wasm_globaltype_delete(ptr)

    @classmethod
    def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasm_globaltype_t]", owner: Optional[Any]) -> "GlobalType":
        if not isinstance(ptr, POINTER(ffi.wasm_globaltype_t)):
            raise TypeError("wrong pointer type")
        ty: "GlobalType" = cls.__new__(cls)
        ty._set_ptr(ptr)
        ty._owner = owner
        return ty

    @property
    def content(self) -> ValType:
        """
        Returns the type this global contains
        """

        ptr = ffi.wasm_globaltype_content(self.ptr())
        return ValType._from_ptr(ptr, self)

    @property
    def mutable(self) -> bool:
        """
        Returns whether this global is mutable or not
        """
        val = ffi.wasm_globaltype_mutability(self.ptr())
        return val == ffi.WASM_VAR.value

    def _as_extern(self) -> "ctypes._Pointer[ffi.wasm_externtype_t]":
        return ffi.wasm_globaltype_as_externtype_const(self.ptr())

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Instance variables

prop content : wasmtime._types.ValType
Expand source code
@property
def content(self) -> ValType:
    """
    Returns the type this global contains
    """

    ptr = ffi.wasm_globaltype_content(self.ptr())
    return ValType._from_ptr(ptr, self)

Returns the type this global contains

prop mutable : bool
Expand source code
@property
def mutable(self) -> bool:
    """
    Returns whether this global is mutable or not
    """
    val = ffi.wasm_globaltype_mutability(self.ptr())
    return val == ffi.WASM_VAR.value

Returns whether this global is mutable or not

class ImportType
Expand source code
class ImportType(Managed["ctypes._Pointer[ffi.wasm_importtype_t]"]):
    _owner: Optional[Any]

    @classmethod
    def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasm_importtype_t]", owner: Optional[Any]) -> "ImportType":
        if not isinstance(ptr, POINTER(ffi.wasm_importtype_t)):
            raise TypeError("wrong pointer type")
        ty: "ImportType" = cls.__new__(cls)
        ty._set_ptr(ptr)
        ty._owner = owner
        return ty

    def _delete(self, ptr: "ctypes._Pointer[ffi.wasm_importtype_t]") -> None:
        if self._owner is None:
            ffi.wasm_importtype_delete(ptr)

    @property
    def module(self) -> str:
        """
        Returns the module this import type refers to
        """

        return ffi.to_str(ffi.wasm_importtype_module(self.ptr()).contents)

    @property
    def name(self) -> Optional[str]:
        """
        Returns the name in the module this import type refers to

        Note that `None` may be returned for the module linking proposal where
        the field name is optional.
        """
        ptr = ffi.wasm_importtype_name(self.ptr())
        if ptr:
            return ffi.to_str(ptr.contents)
        return None

    @property
    def type(self) -> "AsExternType":
        """
        Returns the type that this import refers to
        """
        ptr = ffi.wasm_importtype_type(self.ptr())
        return wrap_externtype(ptr, self._owner or self)

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Instance variables

prop module : str
Expand source code
@property
def module(self) -> str:
    """
    Returns the module this import type refers to
    """

    return ffi.to_str(ffi.wasm_importtype_module(self.ptr()).contents)

Returns the module this import type refers to

prop name : str | None
Expand source code
@property
def name(self) -> Optional[str]:
    """
    Returns the name in the module this import type refers to

    Note that `None` may be returned for the module linking proposal where
    the field name is optional.
    """
    ptr = ffi.wasm_importtype_name(self.ptr())
    if ptr:
        return ffi.to_str(ptr.contents)
    return None

Returns the name in the module this import type refers to

Note that None may be returned for the module linking proposal where the field name is optional.

prop type : AsExternType
Expand source code
@property
def type(self) -> "AsExternType":
    """
    Returns the type that this import refers to
    """
    ptr = ffi.wasm_importtype_type(self.ptr())
    return wrap_externtype(ptr, self._owner or self)

Returns the type that this import refers to

class Instance (store: wasmtime._store.Store | ForwardRef('Caller'),
module: wasmtime._module.Module,
imports: Sequence[ForwardRef('Func') | ForwardRef('Table') | ForwardRef('Memory') | ForwardRef('SharedMemory') | ForwardRef('Global') | ForwardRef('Instance') | ForwardRef('Module')])
Expand source code
class Instance:
    _instance: ffi.wasmtime_instance_t
    _exports: Optional["InstanceExports"]

    def __init__(self, store: Storelike, module: Module, imports: Sequence[AsExtern]):
        """
        Creates a new instance by instantiating the `module` given with the
        `imports` into the `store` provided.

        The `store` must have type `Store`, the `module` must have type
        `Module`, and the `imports` must be an iterable of external values,
        either `Extern`, `Func`, `Table`, `Memory`, or `Global`.

        Raises an error if instantiation fails (e.g. linking or trap) and
        otherwise initializes the new instance.
        """

        imports_ptr = (ffi.wasmtime_extern_t * len(imports))()
        for i, val in enumerate(imports):
            imports_ptr[i] = get_extern_ptr(val)

        instance = ffi.wasmtime_instance_t()
        with enter_wasm(store) as trap:
            error = ffi.wasmtime_instance_new(
                store._context(),
                module.ptr(),
                imports_ptr,
                len(imports),
                byref(instance),
                trap)
            if error:
                raise WasmtimeError._from_ptr(error)
        self._instance = instance
        self._exports = None

    @classmethod
    def _from_raw(cls, instance: ffi.wasmtime_instance_t) -> "Instance":
        ty: "Instance" = cls.__new__(cls)
        ty._exports = None
        ty._instance = instance
        return ty

    def exports(self, store: Storelike) -> "InstanceExports":
        """
        Returns the exports of this module

        The returned type can be indexed both with integers and with strings for
        names of exports.
        """
        if self._exports is None:
            self._exports = InstanceExports(store, self)
        return self._exports

    def _as_extern(self) -> ffi.wasmtime_extern_t:
        union = ffi.wasmtime_extern_union(instance=self._instance)
        return ffi.wasmtime_extern_t(ffi.WASMTIME_EXTERN_INSTANCE, union)

Creates a new instance by instantiating the module given with the imports into the store provided.

The store must have type Store, the module must have type Module, and the imports must be an iterable of external values, either Extern, Func, Table, Memory, or Global.

Raises an error if instantiation fails (e.g. linking or trap) and otherwise initializes the new instance.

Methods

def exports(self,
store: wasmtime._store.Store | ForwardRef('Caller')) ‑> wasmtime._instance.InstanceExports
Expand source code
def exports(self, store: Storelike) -> "InstanceExports":
    """
    Returns the exports of this module

    The returned type can be indexed both with integers and with strings for
    names of exports.
    """
    if self._exports is None:
        self._exports = InstanceExports(store, self)
    return self._exports

Returns the exports of this module

The returned type can be indexed both with integers and with strings for names of exports.

class Limits (min: int, max: int | None)
Expand source code
class Limits:
    def __init__(self, min: int, max: Optional[int]):
        self.min = min
        self.max = max

    def __ffi__(self) -> ffi.wasm_limits_t:
        max = self.max
        if max is None:
            max = 0xffffffff
        return ffi.wasm_limits_t(self.min, max)

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Limits):
            return False
        return self.min == other.min and self.max == other.max

    @classmethod
    def _from_ffi(cls, val: 'ctypes._Pointer[ffi.wasm_limits_t]') -> "Limits":
        min = val.contents.min
        max = val.contents.max
        if max == 0xffffffff:
            return Limits(min, None)
        return Limits(min, max)
class Linker (engine: wasmtime._engine.Engine)
Expand source code
class Linker(Managed["ctypes._Pointer[ffi.wasmtime_linker_t]"]):
    engine: Engine

    def __init__(self, engine: Engine):
        """
        Creates a new linker ready to instantiate modules within the store
        provided.
        """
        self._set_ptr(ffi.wasmtime_linker_new(engine.ptr()))
        self.engine = engine

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

    @setter_property
    def allow_shadowing(self, allow: bool) -> None:
        """
        Configures whether definitions are allowed to shadow one another within
        this linker
        """
        if not isinstance(allow, bool):
            raise TypeError("expected a boolean")
        ffi.wasmtime_linker_allow_shadowing(self.ptr(), allow)

    def define(self, store: Storelike, module: str, name: str, item: AsExtern) -> None:
        """
        Defines a new item, by name, in this linker.

        This method will add a new definition to this linker. The `module` nad
        `name` provided are what to name the `item` within the linker.

        This function will raise an error if `item` comes from the wrong store
        or if shadowing is disallowed and the module/name pair has already been
        defined.
        """
        raw_item = get_extern_ptr(item)
        module_bytes = module.encode('utf-8')
        module_buf = create_string_buffer(module_bytes)
        name_bytes = name.encode('utf-8')
        name_buf = create_string_buffer(name_bytes)
        error = ffi.wasmtime_linker_define(
            self.ptr(),
            store._context(),
            module_buf,
            len(module_bytes),
            name_buf,
            len(name_bytes),
            byref(raw_item))
        if error:
            raise WasmtimeError._from_ptr(error)

    def define_func(self, module: str, name: str, ty: FuncType, func: Callable[..., Any], access_caller: bool = False) -> None:
        """
        Defines a new function, by name, in this linker.

        This method is similar to `define` except that you can directly define a
        function without creating a `Func` itself. This enables
        `Store`-independent functions to be inserted into this linker, meaning
        the linker can be used to instantiate modules in multiple stores.
        """
        module_bytes = module.encode('utf-8')
        module_buf = create_string_buffer(module_bytes)
        name_bytes = name.encode('utf-8')
        name_buf = create_string_buffer(name_bytes)
        if not isinstance(ty, FuncType):
            raise TypeError("expected a FuncType")
        idx = FUNCTIONS.allocate((func, ty.results, access_caller))
        error = ffi.wasmtime_linker_define_func(
            self.ptr(),
            module_buf,
            len(module_bytes),
            name_buf,
            len(name_bytes),
            ty.ptr(),
            trampoline,
            idx,
            finalize)
        if error:
            raise WasmtimeError._from_ptr(error)

    def define_instance(self, store: Storelike, name: str, instance: Instance) -> None:
        """
        Convenience wrapper to define an entire instance in this linker.

        This function will `define` eaech of the exports on the instance into
        this linker, using the name provided as the module name and the export's
        own name as the field name.

        This function will raise an error if `instance` comes from the wrong
        store or if shadowing is disallowed and a name was previously defined.
        """
        if not isinstance(instance, Instance):
            raise TypeError("expected an `Instance`")
        name_bytes = name.encode('utf8')
        name_buf = create_string_buffer(name_bytes)
        error = ffi.wasmtime_linker_define_instance(self.ptr(),
                                                    store._context(),
                                                    name_buf,
                                                    len(name_bytes),
                                                    byref(instance._instance))
        if error:
            raise WasmtimeError._from_ptr(error)

    def define_wasi(self) -> None:
        """
        Defines a WASI instance in this linker.

        The instance provided has been previously constructed and this method
        will define all the appropriate imports and their names into this linker
        to assist with instantiating modules that use WASI.

        This function will raise an error if shadowing is disallowed and a name
        was previously defined.
        """
        error = ffi.wasmtime_linker_define_wasi(self.ptr())
        if error:
            raise WasmtimeError._from_ptr(error)

    def define_module(self, store: Storelike, name: str, module: Module) -> None:
        """
        Defines automatic instantiations of the provided module in this linker.

        The `module` provided is defined under `name` with automatic
        instantiations which respect WASI Commands and Reactors.

        For more information see the Rust documentation at
        https://docs.wasmtime.dev/api/wasmtime/struct.Linker.html#method.module.

        This method will throw an error if shadowing is disallowed and an item
        has previously been defined.
        """
        if not isinstance(module, Module):
            raise TypeError("expected a `Module`")
        name_bytes = name.encode('utf-8')
        name_buf = create_string_buffer(name_bytes)
        error = ffi.wasmtime_linker_module(self.ptr(), store._context(), name_buf, len(name_bytes), module.ptr())
        if error:
            raise WasmtimeError._from_ptr(error)

    def instantiate(self, store: Storelike, module: Module) -> Instance:
        """
        Instantiates a module using this linker's defined set of names.

        This method will attempt to satisfy all the imports of the `module`
        provided with the names defined within this linker. If all names are
        defined then the module is instantiated.

        Raises an error if an import of `module` hasn't been defined in this
        linker or if a trap happens while instantiating the instance.
        """
        trap = POINTER(ffi.wasm_trap_t)()
        instance = ffi.wasmtime_instance_t()
        with enter_wasm(store) as trap:
            error = ffi.wasmtime_linker_instantiate(
                self.ptr(), store._context(), module.ptr(), byref(instance), trap)
            if error:
                raise WasmtimeError._from_ptr(error)
        return Instance._from_raw(instance)

    def get_default(self, store: Storelike, name: str) -> Func:
        """
        Gets the default export for the named module in this linker.

        For more information on this see the Rust documentation at
        https://docs.wasmtime.dev/api/wasmtime/struct.Linker.html#method.get_default.

        Raises an error if the default export wasn't present.
        """
        name_bytes = name.encode('utf-8')
        name_buf = create_string_buffer(name_bytes)
        func = ffi.wasmtime_func_t()
        error = ffi.wasmtime_linker_get_default(self.ptr(), store._context(),
                                                name_buf, len(name_bytes), byref(func))
        if error:
            raise WasmtimeError._from_ptr(error)
        return Func._from_raw(func)

    def get(self, store: Storelike, module: str, name: str) -> AsExtern:
        """
        Gets a singular item defined in this linker.

        Raises an error if this item hasn't been defined or if the item has been
        defined twice with different types.
        """
        module_bytes = module.encode('utf-8')
        module_buf = create_string_buffer(module_bytes)
        name_bytes = name.encode('utf-8')
        name_buf = create_string_buffer(name_bytes)
        item = ffi.wasmtime_extern_t()
        ok = ffi.wasmtime_linker_get(self.ptr(), store._context(),
                                     module_buf, len(module_bytes),
                                     name_buf, len(name_bytes),
                                     byref(item))
        if ok:
            return wrap_extern(item)
        raise WasmtimeError("item not defined in linker")

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Creates a new linker ready to instantiate modules within the store provided.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Class variables

var engine : wasmtime._engine.Engine

Instance variables

prop allow_shadowing
Expand source code
@setter_property
def allow_shadowing(self, allow: bool) -> None:
    """
    Configures whether definitions are allowed to shadow one another within
    this linker
    """
    if not isinstance(allow, bool):
        raise TypeError("expected a boolean")
    ffi.wasmtime_linker_allow_shadowing(self.ptr(), allow)

Configures whether definitions are allowed to shadow one another within this linker

Note that this field can only be set, it cannot be read

Methods

def define(self,
store: wasmtime._store.Store | ForwardRef('Caller'),
module: str,
name: str,
item: ForwardRef('Func') | ForwardRef('Table') | ForwardRef('Memory') | ForwardRef('SharedMemory') | ForwardRef('Global') | ForwardRef('Instance') | ForwardRef('Module'))
Expand source code
def define(self, store: Storelike, module: str, name: str, item: AsExtern) -> None:
    """
    Defines a new item, by name, in this linker.

    This method will add a new definition to this linker. The `module` nad
    `name` provided are what to name the `item` within the linker.

    This function will raise an error if `item` comes from the wrong store
    or if shadowing is disallowed and the module/name pair has already been
    defined.
    """
    raw_item = get_extern_ptr(item)
    module_bytes = module.encode('utf-8')
    module_buf = create_string_buffer(module_bytes)
    name_bytes = name.encode('utf-8')
    name_buf = create_string_buffer(name_bytes)
    error = ffi.wasmtime_linker_define(
        self.ptr(),
        store._context(),
        module_buf,
        len(module_bytes),
        name_buf,
        len(name_bytes),
        byref(raw_item))
    if error:
        raise WasmtimeError._from_ptr(error)

Defines a new item, by name, in this linker.

This method will add a new definition to this linker. The module nad name provided are what to name the item within the linker.

This function will raise an error if item comes from the wrong store or if shadowing is disallowed and the module/name pair has already been defined.

def define_func(self,
module: str,
name: str,
ty: wasmtime._types.FuncType,
func: Callable[..., Any],
access_caller: bool = False) ‑> None
Expand source code
def define_func(self, module: str, name: str, ty: FuncType, func: Callable[..., Any], access_caller: bool = False) -> None:
    """
    Defines a new function, by name, in this linker.

    This method is similar to `define` except that you can directly define a
    function without creating a `Func` itself. This enables
    `Store`-independent functions to be inserted into this linker, meaning
    the linker can be used to instantiate modules in multiple stores.
    """
    module_bytes = module.encode('utf-8')
    module_buf = create_string_buffer(module_bytes)
    name_bytes = name.encode('utf-8')
    name_buf = create_string_buffer(name_bytes)
    if not isinstance(ty, FuncType):
        raise TypeError("expected a FuncType")
    idx = FUNCTIONS.allocate((func, ty.results, access_caller))
    error = ffi.wasmtime_linker_define_func(
        self.ptr(),
        module_buf,
        len(module_bytes),
        name_buf,
        len(name_bytes),
        ty.ptr(),
        trampoline,
        idx,
        finalize)
    if error:
        raise WasmtimeError._from_ptr(error)

Defines a new function, by name, in this linker.

This method is similar to define except that you can directly define a function without creating a Func itself. This enables Store-independent functions to be inserted into this linker, meaning the linker can be used to instantiate modules in multiple stores.

def define_instance(self,
store: wasmtime._store.Store | ForwardRef('Caller'),
name: str,
instance: wasmtime._instance.Instance) ‑> None
Expand source code
def define_instance(self, store: Storelike, name: str, instance: Instance) -> None:
    """
    Convenience wrapper to define an entire instance in this linker.

    This function will `define` eaech of the exports on the instance into
    this linker, using the name provided as the module name and the export's
    own name as the field name.

    This function will raise an error if `instance` comes from the wrong
    store or if shadowing is disallowed and a name was previously defined.
    """
    if not isinstance(instance, Instance):
        raise TypeError("expected an `Instance`")
    name_bytes = name.encode('utf8')
    name_buf = create_string_buffer(name_bytes)
    error = ffi.wasmtime_linker_define_instance(self.ptr(),
                                                store._context(),
                                                name_buf,
                                                len(name_bytes),
                                                byref(instance._instance))
    if error:
        raise WasmtimeError._from_ptr(error)

Convenience wrapper to define an entire instance in this linker.

This function will define eaech of the exports on the instance into this linker, using the name provided as the module name and the export's own name as the field name.

This function will raise an error if instance comes from the wrong store or if shadowing is disallowed and a name was previously defined.

def define_module(self,
store: wasmtime._store.Store | ForwardRef('Caller'),
name: str,
module: wasmtime._module.Module) ‑> None
Expand source code
def define_module(self, store: Storelike, name: str, module: Module) -> None:
    """
    Defines automatic instantiations of the provided module in this linker.

    The `module` provided is defined under `name` with automatic
    instantiations which respect WASI Commands and Reactors.

    For more information see the Rust documentation at
    https://docs.wasmtime.dev/api/wasmtime/struct.Linker.html#method.module.

    This method will throw an error if shadowing is disallowed and an item
    has previously been defined.
    """
    if not isinstance(module, Module):
        raise TypeError("expected a `Module`")
    name_bytes = name.encode('utf-8')
    name_buf = create_string_buffer(name_bytes)
    error = ffi.wasmtime_linker_module(self.ptr(), store._context(), name_buf, len(name_bytes), module.ptr())
    if error:
        raise WasmtimeError._from_ptr(error)

Defines automatic instantiations of the provided module in this linker.

The module provided is defined under name with automatic instantiations which respect WASI Commands and Reactors.

For more information see the Rust documentation at https://docs.wasmtime.dev/api/wasmtime/struct.Linker.html#method.module.

This method will throw an error if shadowing is disallowed and an item has previously been defined.

def define_wasi(self) ‑> None
Expand source code
def define_wasi(self) -> None:
    """
    Defines a WASI instance in this linker.

    The instance provided has been previously constructed and this method
    will define all the appropriate imports and their names into this linker
    to assist with instantiating modules that use WASI.

    This function will raise an error if shadowing is disallowed and a name
    was previously defined.
    """
    error = ffi.wasmtime_linker_define_wasi(self.ptr())
    if error:
        raise WasmtimeError._from_ptr(error)

Defines a WASI instance in this linker.

The instance provided has been previously constructed and this method will define all the appropriate imports and their names into this linker to assist with instantiating modules that use WASI.

This function will raise an error if shadowing is disallowed and a name was previously defined.

def get(self,
store: wasmtime._store.Store | ForwardRef('Caller'),
module: str,
name: str) ‑> Func | Table | Memory | SharedMemory | Global | Instance | Module
Expand source code
def get(self, store: Storelike, module: str, name: str) -> AsExtern:
    """
    Gets a singular item defined in this linker.

    Raises an error if this item hasn't been defined or if the item has been
    defined twice with different types.
    """
    module_bytes = module.encode('utf-8')
    module_buf = create_string_buffer(module_bytes)
    name_bytes = name.encode('utf-8')
    name_buf = create_string_buffer(name_bytes)
    item = ffi.wasmtime_extern_t()
    ok = ffi.wasmtime_linker_get(self.ptr(), store._context(),
                                 module_buf, len(module_bytes),
                                 name_buf, len(name_bytes),
                                 byref(item))
    if ok:
        return wrap_extern(item)
    raise WasmtimeError("item not defined in linker")

Gets a singular item defined in this linker.

Raises an error if this item hasn't been defined or if the item has been defined twice with different types.

def get_default(self,
store: wasmtime._store.Store | ForwardRef('Caller'),
name: str) ‑> wasmtime._func.Func
Expand source code
def get_default(self, store: Storelike, name: str) -> Func:
    """
    Gets the default export for the named module in this linker.

    For more information on this see the Rust documentation at
    https://docs.wasmtime.dev/api/wasmtime/struct.Linker.html#method.get_default.

    Raises an error if the default export wasn't present.
    """
    name_bytes = name.encode('utf-8')
    name_buf = create_string_buffer(name_bytes)
    func = ffi.wasmtime_func_t()
    error = ffi.wasmtime_linker_get_default(self.ptr(), store._context(),
                                            name_buf, len(name_bytes), byref(func))
    if error:
        raise WasmtimeError._from_ptr(error)
    return Func._from_raw(func)

Gets the default export for the named module in this linker.

For more information on this see the Rust documentation at https://docs.wasmtime.dev/api/wasmtime/struct.Linker.html#method.get_default.

Raises an error if the default export wasn't present.

def instantiate(self,
store: wasmtime._store.Store | ForwardRef('Caller'),
module: wasmtime._module.Module) ‑> wasmtime._instance.Instance
Expand source code
def instantiate(self, store: Storelike, module: Module) -> Instance:
    """
    Instantiates a module using this linker's defined set of names.

    This method will attempt to satisfy all the imports of the `module`
    provided with the names defined within this linker. If all names are
    defined then the module is instantiated.

    Raises an error if an import of `module` hasn't been defined in this
    linker or if a trap happens while instantiating the instance.
    """
    trap = POINTER(ffi.wasm_trap_t)()
    instance = ffi.wasmtime_instance_t()
    with enter_wasm(store) as trap:
        error = ffi.wasmtime_linker_instantiate(
            self.ptr(), store._context(), module.ptr(), byref(instance), trap)
        if error:
            raise WasmtimeError._from_ptr(error)
    return Instance._from_raw(instance)

Instantiates a module using this linker's defined set of names.

This method will attempt to satisfy all the imports of the module provided with the names defined within this linker. If all names are defined then the module is instantiated.

Raises an error if an import of module hasn't been defined in this linker or if a trap happens while instantiating the instance.

class Memory (store: wasmtime._store.Store | ForwardRef('Caller'),
ty: wasmtime._types.MemoryType)
Expand source code
class Memory:
    _memory: ffi.wasmtime_memory_t

    def __init__(self, store: Storelike, ty: MemoryType):
        """
        Creates a new memory in `store` with the given `ty`
        """

        mem = ffi.wasmtime_memory_t()
        error = ffi.wasmtime_memory_new(store._context(), ty.ptr(), byref(mem))
        if error:
            raise WasmtimeError._from_ptr(error)
        self._memory = mem

    @classmethod
    def _from_raw(cls, mem: ffi.wasmtime_memory_t) -> "Memory":
        ty: "Memory" = cls.__new__(cls)
        ty._memory = mem
        return ty

    def type(self, store: Storelike) -> MemoryType:
        """
        Gets the type of this memory as a `MemoryType`
        """

        ptr = ffi.wasmtime_memory_type(store._context(), byref(self._memory))
        return MemoryType._from_ptr(ptr, None)

    def grow(self, store: Storelike, delta: int) -> int:
        """
        Grows this memory by the given number of pages
        """

        if delta < 0:
            raise WasmtimeError("cannot grow by negative amount")
        prev = ffi.c_uint64(0)
        error = ffi.wasmtime_memory_grow(store._context(), byref(self._memory), delta, byref(prev))
        if error:
            raise WasmtimeError._from_ptr(error)
        return prev.value

    def size(self, store: Storelike) -> int:
        """
        Returns the size, in WebAssembly pages, of this memory.
        """

        return ffi.wasmtime_memory_size(store._context(), byref(self._memory))

    def data_ptr(self, store: Storelike) -> "ctypes._Pointer[c_ubyte]":
        """
        Returns the raw pointer in memory where this wasm memory lives.

        Remember that all accesses to wasm memory should be bounds-checked
        against the `data_len` method.
        """
        return ffi.wasmtime_memory_data(store._context(), byref(self._memory))

    def get_buffer_ptr(self, store: Storelike,
                       size: typing.Optional[int] = None,
                       offset: int = 0) -> ctypes.Array:
        """
        return raw pointer to buffer suitable for creating zero-copy writable NumPy Buffer Protocol
        this method is also used internally by `read()` and `write()`

        np_mem = np.frombuffer(memory.get_buffer_ptr(store), dtype=np.uint8)
        np_mem[start:end] = A # write
        B = np_mem[start:end] # read
        """
        if size is None:
            size = self.data_len(store)
        ptr_type = ctypes.c_ubyte * size
        return ptr_type.from_address(ctypes.addressof(self.data_ptr(store).contents) + offset)

    def read(
            self,
            store: Storelike,
            start: typing.Optional[int] = 0,
            stop: typing.Optional[int] = None) -> bytearray:
        """
        Reads this memory starting from `start` and up to `stop`
        and returns a copy of the contents as a `bytearray`.

        The indexing behavior of this method is similar to `list[start:stop]`
        where negative starts can be used to read from the end, for example.
        """
        size = self.data_len(store)
        key = slice(start, stop, None)
        start, stop, _ = key.indices(size)
        val_size = stop - start
        if val_size <= 0:
            # return bytearray of size zero
            return bytearray(0)
        src_ptr = self.get_buffer_ptr(store, val_size, start)
        return bytearray(src_ptr)

    def write(
            self,
            store: Storelike,
            value: typing.Union[bytearray, bytes],
            start: typing.Optional[int] = None) -> int:
        """
        write a bytearray value into a possibly large slice of memory
        negative start is allowed in a way similat to list slice mylist[-10:]
        if value is not bytearray it will be used to construct an intermediate bytearray (copyied twice)
        return number of bytes written
        raises IndexError when trying to write outside the memory range
        this happens when start offset is >= size or when end side of value is >= size
        """
        size = self.data_len(store)
        key = slice(start, None)
        start = key.indices(size)[0]
        if start >= size:
            raise IndexError("index out of range")
        # value must be bytearray ex. cast bytes() to bytearray
        if not isinstance(value, bytearray):
            value = bytearray(value)
        val_size = len(value)
        if val_size == 0:
            return val_size
        # stop is exclusive
        stop = start + val_size
        if stop > size:
            raise IndexError("index out of range")
        ptr_type = ctypes.c_ubyte * val_size
        src_ptr = ptr_type.from_buffer(value)
        dst_ptr = self.get_buffer_ptr(store, val_size, start)
        ctypes.memmove(dst_ptr, src_ptr, val_size)
        return val_size

    def data_len(self, store: Storelike) -> int:
        """
        Returns the raw byte length of this memory.
        """

        return ffi.wasmtime_memory_data_size(store._context(), byref(self._memory))

    def _as_extern(self) -> ffi.wasmtime_extern_t:
        union = ffi.wasmtime_extern_union(memory=self._memory)
        return ffi.wasmtime_extern_t(ffi.WASMTIME_EXTERN_MEMORY, union)

Creates a new memory in store with the given ty

Methods

def data_len(self,
store: wasmtime._store.Store | ForwardRef('Caller')) ‑> int
Expand source code
def data_len(self, store: Storelike) -> int:
    """
    Returns the raw byte length of this memory.
    """

    return ffi.wasmtime_memory_data_size(store._context(), byref(self._memory))

Returns the raw byte length of this memory.

def data_ptr(self,
store: wasmtime._store.Store | ForwardRef('Caller')) ‑> ctypes._Pointer[c_ubyte]
Expand source code
def data_ptr(self, store: Storelike) -> "ctypes._Pointer[c_ubyte]":
    """
    Returns the raw pointer in memory where this wasm memory lives.

    Remember that all accesses to wasm memory should be bounds-checked
    against the `data_len` method.
    """
    return ffi.wasmtime_memory_data(store._context(), byref(self._memory))

Returns the raw pointer in memory where this wasm memory lives.

Remember that all accesses to wasm memory should be bounds-checked against the data_len method.

def get_buffer_ptr(self,
store: wasmtime._store.Store | ForwardRef('Caller'),
size: int | None = None,
offset: int = 0) ‑> _ctypes.Array
Expand source code
def get_buffer_ptr(self, store: Storelike,
                   size: typing.Optional[int] = None,
                   offset: int = 0) -> ctypes.Array:
    """
    return raw pointer to buffer suitable for creating zero-copy writable NumPy Buffer Protocol
    this method is also used internally by `read()` and `write()`

    np_mem = np.frombuffer(memory.get_buffer_ptr(store), dtype=np.uint8)
    np_mem[start:end] = A # write
    B = np_mem[start:end] # read
    """
    if size is None:
        size = self.data_len(store)
    ptr_type = ctypes.c_ubyte * size
    return ptr_type.from_address(ctypes.addressof(self.data_ptr(store).contents) + offset)

return raw pointer to buffer suitable for creating zero-copy writable NumPy Buffer Protocol this method is also used internally by read() and write()

np_mem = np.frombuffer(memory.get_buffer_ptr(store), dtype=np.uint8) np_mem[start:end] = A # write B = np_mem[start:end] # read

def grow(self,
store: wasmtime._store.Store | ForwardRef('Caller'),
delta: int) ‑> int
Expand source code
def grow(self, store: Storelike, delta: int) -> int:
    """
    Grows this memory by the given number of pages
    """

    if delta < 0:
        raise WasmtimeError("cannot grow by negative amount")
    prev = ffi.c_uint64(0)
    error = ffi.wasmtime_memory_grow(store._context(), byref(self._memory), delta, byref(prev))
    if error:
        raise WasmtimeError._from_ptr(error)
    return prev.value

Grows this memory by the given number of pages

def read(self,
store: wasmtime._store.Store | ForwardRef('Caller'),
start: int | None = 0,
stop: int | None = None) ‑> bytearray
Expand source code
def read(
        self,
        store: Storelike,
        start: typing.Optional[int] = 0,
        stop: typing.Optional[int] = None) -> bytearray:
    """
    Reads this memory starting from `start` and up to `stop`
    and returns a copy of the contents as a `bytearray`.

    The indexing behavior of this method is similar to `list[start:stop]`
    where negative starts can be used to read from the end, for example.
    """
    size = self.data_len(store)
    key = slice(start, stop, None)
    start, stop, _ = key.indices(size)
    val_size = stop - start
    if val_size <= 0:
        # return bytearray of size zero
        return bytearray(0)
    src_ptr = self.get_buffer_ptr(store, val_size, start)
    return bytearray(src_ptr)

Reads this memory starting from start and up to stop and returns a copy of the contents as a bytearray.

The indexing behavior of this method is similar to list[start:stop] where negative starts can be used to read from the end, for example.

def size(self,
store: wasmtime._store.Store | ForwardRef('Caller')) ‑> int
Expand source code
def size(self, store: Storelike) -> int:
    """
    Returns the size, in WebAssembly pages, of this memory.
    """

    return ffi.wasmtime_memory_size(store._context(), byref(self._memory))

Returns the size, in WebAssembly pages, of this memory.

def type(self,
store: wasmtime._store.Store | ForwardRef('Caller')) ‑> wasmtime._types.MemoryType
Expand source code
def type(self, store: Storelike) -> MemoryType:
    """
    Gets the type of this memory as a `MemoryType`
    """

    ptr = ffi.wasmtime_memory_type(store._context(), byref(self._memory))
    return MemoryType._from_ptr(ptr, None)

Gets the type of this memory as a MemoryType

def write(self,
store: wasmtime._store.Store | ForwardRef('Caller'),
value: bytearray | bytes,
start: int | None = None) ‑> int
Expand source code
def write(
        self,
        store: Storelike,
        value: typing.Union[bytearray, bytes],
        start: typing.Optional[int] = None) -> int:
    """
    write a bytearray value into a possibly large slice of memory
    negative start is allowed in a way similat to list slice mylist[-10:]
    if value is not bytearray it will be used to construct an intermediate bytearray (copyied twice)
    return number of bytes written
    raises IndexError when trying to write outside the memory range
    this happens when start offset is >= size or when end side of value is >= size
    """
    size = self.data_len(store)
    key = slice(start, None)
    start = key.indices(size)[0]
    if start >= size:
        raise IndexError("index out of range")
    # value must be bytearray ex. cast bytes() to bytearray
    if not isinstance(value, bytearray):
        value = bytearray(value)
    val_size = len(value)
    if val_size == 0:
        return val_size
    # stop is exclusive
    stop = start + val_size
    if stop > size:
        raise IndexError("index out of range")
    ptr_type = ctypes.c_ubyte * val_size
    src_ptr = ptr_type.from_buffer(value)
    dst_ptr = self.get_buffer_ptr(store, val_size, start)
    ctypes.memmove(dst_ptr, src_ptr, val_size)
    return val_size

write a bytearray value into a possibly large slice of memory negative start is allowed in a way similat to list slice mylist[-10:] if value is not bytearray it will be used to construct an intermediate bytearray (copyied twice) return number of bytes written raises IndexError when trying to write outside the memory range this happens when start offset is >= size or when end side of value is >= size

class MemoryType (limits: wasmtime._types.Limits, is_64: bool = False, shared: bool = False)
Expand source code
class MemoryType(Managed["ctypes._Pointer[ffi.wasm_memorytype_t]"]):
    def __init__(self, limits: Limits, is_64: bool = False, shared: bool = False):
        if not isinstance(limits, Limits):
            raise TypeError("expected Limits")
        if is_64:
            maximum = 0x10000000000000000
        else:
            maximum = 0x100000000
        if limits.min >= maximum:
            raise WasmtimeError("minimum size too large")
        if limits.max and limits.max >= maximum:
            raise WasmtimeError("maximum size too large")
        ptr = ffi.wasmtime_memorytype_new(limits.min,
                                          limits.max is not None,
                                          limits.max if limits.max else 0,
                                          is_64,
                                          shared)
        if not ptr:
            raise WasmtimeError("failed to allocate MemoryType")
        self._set_ptr(ptr)
        self._owner = None

    def _delete(self, ptr: "ctypes._Pointer[ffi.wasm_memorytype_t]") -> None:
        if self._owner is None:
            ffi.wasm_memorytype_delete(ptr)

    @classmethod
    def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasm_memorytype_t]", owner: Optional[Any]) -> "MemoryType":
        if not isinstance(ptr, POINTER(ffi.wasm_memorytype_t)):
            raise TypeError("wrong pointer type")
        ty: "MemoryType" = cls.__new__(cls)
        ty._set_ptr(ptr)
        ty._owner = owner
        return ty

    @property
    def limits(self) -> Limits:
        """
        Returns the limits on the size of this table
        """
        minimum = ffi.wasmtime_memorytype_minimum(self.ptr())
        maximum = ffi.c_uint64(0)
        has_max = ffi.wasmtime_memorytype_maximum(self.ptr(), byref(maximum))
        return Limits(minimum, maximum.value if has_max else None)

    @property
    def is_64(self) -> bool:
        """
        Returns whether or not this is a 64-bit memory
        """
        return ffi.wasmtime_memorytype_is64(self.ptr())
    
    @property
    def is_shared(self) -> bool:
        """
        Returns whether or not this is a shared memory
        """
        return ffi.wasmtime_memorytype_isshared(self.ptr())


    def _as_extern(self) -> "ctypes._Pointer[ffi.wasm_externtype_t]":
        return ffi.wasm_memorytype_as_externtype_const(self.ptr())

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Instance variables

prop is_64 : bool
Expand source code
@property
def is_64(self) -> bool:
    """
    Returns whether or not this is a 64-bit memory
    """
    return ffi.wasmtime_memorytype_is64(self.ptr())

Returns whether or not this is a 64-bit memory

prop is_shared : bool
Expand source code
@property
def is_shared(self) -> bool:
    """
    Returns whether or not this is a shared memory
    """
    return ffi.wasmtime_memorytype_isshared(self.ptr())

Returns whether or not this is a shared memory

prop limits : wasmtime._types.Limits
Expand source code
@property
def limits(self) -> Limits:
    """
    Returns the limits on the size of this table
    """
    minimum = ffi.wasmtime_memorytype_minimum(self.ptr())
    maximum = ffi.c_uint64(0)
    has_max = ffi.wasmtime_memorytype_maximum(self.ptr(), byref(maximum))
    return Limits(minimum, maximum.value if has_max else None)

Returns the limits on the size of this table

class Module (engine: wasmtime._engine.Engine, wasm: str | bytes)
Expand source code
class Module(Managed["ctypes._Pointer[ffi.wasmtime_module_t]"]):

    @classmethod
    def from_file(cls, engine: Engine, path: typing.Union[str, bytes, PathLike]) -> "Module":
        """
        Compiles and creates a new `Module` by reading the file at `path` and
        then delegating to the `Module` constructor.
        """

        with open(path, "rb") as f:
            contents = f.read()
        return cls(engine, contents)

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

        if not isinstance(engine, Engine):
            raise TypeError("expected an Engine")

        # If this looks like a string, parse it as the text format. Note that
        # in python 2 strings and bytes are basically the same, so we skip this
        # if the first byte in the string is 0, meaning this is actually a wasm
        # module.
        if isinstance(wasm, str) and len(wasm) > 0 and ord(wasm[0]) != 0:
            wasm = wat2wasm(wasm)
        if isinstance(wasm, bytes) and len(wasm) > 0 and wasm[0] != 0:
            wasm = wat2wasm(wasm)

        if not isinstance(wasm, (bytes, bytearray)):
            raise TypeError("expected wasm bytes")

        # TODO: can the copy be avoided here? I can't for the life of me
        # figure this out.
        binary = (c_uint8 * len(wasm)).from_buffer_copy(wasm)
        ptr = POINTER(ffi.wasmtime_module_t)()
        error = ffi.wasmtime_module_new(engine.ptr(), binary, len(wasm), byref(ptr))
        if error:
            raise WasmtimeError._from_ptr(error)
        self._set_ptr(ptr)

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

    @classmethod
    def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_module_t]") -> "Module":
        if not isinstance(ptr, POINTER(ffi.wasmtime_module_t)):
            raise TypeError("wrong pointer type")
        ty: "Module" = cls.__new__(cls)
        ty._set_ptr(ptr)
        return ty

    @classmethod
    def deserialize(cls, engine: Engine, encoded: typing.Union[bytes, bytearray]) -> 'Module':
        """
        Deserializes bytes previously created by `Module.serialize`.

        This constructor for `Module` will deserialize bytes previously created
        by a serialized module. This will only succeed if the bytes were
        previously created by the same version of `wasmtime` as well as the
        same configuration within `Engine`.
        """

        if not isinstance(encoded, (bytes, bytearray)):
            raise TypeError("expected bytes")

        ptr = POINTER(ffi.wasmtime_module_t)()

        # TODO: can the copy be avoided here? I can't for the life of me
        # figure this out.
        error = ffi.wasmtime_module_deserialize(
            engine.ptr(),
            (c_uint8 * len(encoded)).from_buffer_copy(encoded),
            len(encoded),
            byref(ptr))
        if error:
            raise WasmtimeError._from_ptr(error)
        ret: "Module" = cls.__new__(cls)
        ret._set_ptr(ptr)
        return ret

    @classmethod
    def deserialize_file(cls, engine: Engine, path: str) -> 'Module':
        """
        Deserializes bytes previously created by `Module.serialize` that are
        stored in a file on the filesystem.

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

        ptr = POINTER(ffi.wasmtime_module_t)()
        path_bytes = path.encode('utf-8')
        error = ffi.wasmtime_module_deserialize_file(
            engine.ptr(),
            path_bytes,
            byref(ptr))
        if error:
            raise WasmtimeError._from_ptr(error)
        ret: "Module" = cls.__new__(cls)
        ret._set_ptr(ptr)
        return ret

    @classmethod
    def validate(cls, engine: Engine, wasm: typing.Union[bytes, bytearray]) -> None:
        """
        Validates whether the list of bytes `wasm` provided is a valid
        WebAssembly binary given the configuration in `store`

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

        if not isinstance(wasm, (bytes, bytearray)):
            raise TypeError("expected wasm bytes")

        # TODO: can the copy be avoided here? I can't for the life of me
        # figure this out.
        buf = (c_uint8 * len(wasm)).from_buffer_copy(wasm)
        error = ffi.wasmtime_module_validate(engine.ptr(), buf, len(wasm))

        if error:
            raise WasmtimeError._from_ptr(error)

#     @property
#     def type(self) -> ModuleType:
#         """
#         Gets the type of this module as a `ModuleType`
#         """

#         ptr = ffi.wasmtime_module_type(self.ptr())
#         return ModuleType._from_ptr(ptr, None)

    @property
    def imports(self) -> typing.List[ImportType]:
        """
        Returns the types of imports that this module has
        """

        imports = ImportTypeList()
        ffi.wasmtime_module_imports(self.ptr(), byref(imports.vec))
        ret = []
        for i in range(0, imports.vec.size):
            ret.append(ImportType._from_ptr(imports.vec.data[i], imports))
        return ret

    @property
    def exports(self) -> typing.List[ExportType]:
        """
        Returns the types of the exports that this module has
        """

        exports = ExportTypeList()
        ffi.wasmtime_module_exports(self.ptr(), byref(exports.vec))
        ret = []
        for i in range(0, exports.vec.size):
            ret.append(ExportType._from_ptr(exports.vec.data[i], exports))
        return ret

    def serialize(self) -> bytearray:
        """
        Serializes this module to a binary representation.

        This method will serialize this module to an in-memory byte array which
        can be cached and later passed to `Module.deserialize` to recreate this
        module.
        """
        raw = ffi.wasm_byte_vec_t()
        err = ffi.wasmtime_module_serialize(self.ptr(), byref(raw))
        if err:
            raise WasmtimeError._from_ptr(err)
        ret = ffi.to_bytes(raw)
        ffi.wasm_byte_vec_delete(byref(raw))
        return ret

    def _as_extern(self) -> ffi.wasmtime_extern_t:
        union = ffi.wasmtime_extern_union(module=self.ptr())
        return ffi.wasmtime_extern_t(ffi.WASMTIME_EXTERN_MODULE, union)

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Static methods

def deserialize(engine: wasmtime._engine.Engine, encoded: bytearray | bytes) ‑> wasmtime._module.Module

Deserializes bytes previously created by Module.serialize().

This constructor for Module will deserialize bytes previously created by a serialized module. This will only succeed if the bytes were previously created by the same version of wasmtime as well as the same configuration within Engine.

def deserialize_file(engine: wasmtime._engine.Engine, path: str) ‑> wasmtime._module.Module

Deserializes bytes previously created by Module.serialize() that are stored in a file on the filesystem.

Otherwise this function is the same as Module.deserialize().

def from_file(engine: wasmtime._engine.Engine, path: str | bytes | os.PathLike) ‑> wasmtime._module.Module

Compiles and creates a new Module by reading the file at path and then delegating to the Module constructor.

def validate(engine: wasmtime._engine.Engine, wasm: bytearray | bytes) ‑> None

Validates whether the list of bytes wasm provided is a valid WebAssembly binary given the configuration in store

Raises a WasmtimeError if the wasm isn't valid.

Instance variables

prop exports : List[wasmtime._types.ExportType]
Expand source code
@property
def exports(self) -> typing.List[ExportType]:
    """
    Returns the types of the exports that this module has
    """

    exports = ExportTypeList()
    ffi.wasmtime_module_exports(self.ptr(), byref(exports.vec))
    ret = []
    for i in range(0, exports.vec.size):
        ret.append(ExportType._from_ptr(exports.vec.data[i], exports))
    return ret

Returns the types of the exports that this module has

prop imports : List[wasmtime._types.ImportType]
Expand source code
@property
def imports(self) -> typing.List[ImportType]:
    """
    Returns the types of imports that this module has
    """

    imports = ImportTypeList()
    ffi.wasmtime_module_imports(self.ptr(), byref(imports.vec))
    ret = []
    for i in range(0, imports.vec.size):
        ret.append(ImportType._from_ptr(imports.vec.data[i], imports))
    return ret

Returns the types of imports that this module has

Methods

def serialize(self) ‑> bytearray
Expand source code
def serialize(self) -> bytearray:
    """
    Serializes this module to a binary representation.

    This method will serialize this module to an in-memory byte array which
    can be cached and later passed to `Module.deserialize` to recreate this
    module.
    """
    raw = ffi.wasm_byte_vec_t()
    err = ffi.wasmtime_module_serialize(self.ptr(), byref(raw))
    if err:
        raise WasmtimeError._from_ptr(err)
    ret = ffi.to_bytes(raw)
    ffi.wasm_byte_vec_delete(byref(raw))
    return ret

Serializes this module to a binary representation.

This method will serialize this module to an in-memory byte array which can be cached and later passed to Module.deserialize() to recreate this module.

class SharedMemory (engine: wasmtime._engine.Engine, ty: wasmtime._types.MemoryType)
Expand source code
class SharedMemory(Managed["ctypes._Pointer[ffi.wasmtime_sharedmemory_t]"]):
    def __init__(self, engine: Engine, ty: MemoryType):
        """
        Creates a new shared memory in `store` with the given `ty`
        """

        sharedmemory_ptr = POINTER(ffi.wasmtime_sharedmemory_t)()
        error = ffi.wasmtime_sharedmemory_new(engine.ptr(), ty.ptr(), byref(sharedmemory_ptr))
        if error:
            raise WasmtimeError._from_ptr(error)
        self._set_ptr(sharedmemory_ptr)

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

    @classmethod
    def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_sharedmemory_t]") -> "SharedMemory":
        if not isinstance(ptr, POINTER(ffi.wasmtime_sharedmemory_t)):
            raise TypeError("wrong shared memory pointer type provided to _from_ptr")
        
        ty: "SharedMemory" = cls.__new__(cls)
        ty._set_ptr(ptr)
        return ty

    def type(self) -> MemoryType:
        """
        Gets the type of this memory as a `MemoryType`
        """

        ptr = ffi.wasmtime_sharedmemory_type(self.ptr())
        return MemoryType._from_ptr(ptr, None)

    def grow(self, delta: int) -> int:
        """
        Grows this memory by the given number of pages
        """

        if delta < 0:
            raise WasmtimeError("cannot grow by negative amount")
        prev = ffi.c_uint64(0)
        error = ffi.wasmtime_sharedmemory_grow(self.ptr(), delta, byref(prev))
        if error:
            raise WasmtimeError._from_ptr(error)
        return prev.value

    def size(self) -> int:
        """
        Returns the size, in WebAssembly pages, of this shared memory.
        """

        return ffi.wasmtime_sharedmemory_size(byref(self.ptr()))

    def data_ptr(self) -> "ctypes._Pointer[c_ubyte]":
        """
        Returns the raw pointer in memory where this wasm shared memory lives.

        Remember that all accesses to wasm shared memory should be bounds-checked
        against the `data_len` method.
        """
        return ffi.wasmtime_sharedmemory_data(self.ptr())

    def data_len(self) -> int:
        """
        Returns the raw byte length of this memory.
        """

        return ffi.wasmtime_sharedmemory_data_size(self.ptr())

    def _as_extern(self) -> ffi.wasmtime_extern_t:
        union = ffi.wasmtime_extern_union(sharedmemory=pointer(self.ptr()))
        return ffi.wasmtime_extern_t(ffi.WASMTIME_EXTERN_SHAREDMEMORY, union)

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Creates a new shared memory in store with the given ty

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Methods

def data_len(self) ‑> int
Expand source code
def data_len(self) -> int:
    """
    Returns the raw byte length of this memory.
    """

    return ffi.wasmtime_sharedmemory_data_size(self.ptr())

Returns the raw byte length of this memory.

def data_ptr(self) ‑> ctypes._Pointer[c_ubyte]
Expand source code
def data_ptr(self) -> "ctypes._Pointer[c_ubyte]":
    """
    Returns the raw pointer in memory where this wasm shared memory lives.

    Remember that all accesses to wasm shared memory should be bounds-checked
    against the `data_len` method.
    """
    return ffi.wasmtime_sharedmemory_data(self.ptr())

Returns the raw pointer in memory where this wasm shared memory lives.

Remember that all accesses to wasm shared memory should be bounds-checked against the data_len method.

def grow(self, delta: int) ‑> int
Expand source code
def grow(self, delta: int) -> int:
    """
    Grows this memory by the given number of pages
    """

    if delta < 0:
        raise WasmtimeError("cannot grow by negative amount")
    prev = ffi.c_uint64(0)
    error = ffi.wasmtime_sharedmemory_grow(self.ptr(), delta, byref(prev))
    if error:
        raise WasmtimeError._from_ptr(error)
    return prev.value

Grows this memory by the given number of pages

def size(self) ‑> int
Expand source code
def size(self) -> int:
    """
    Returns the size, in WebAssembly pages, of this shared memory.
    """

    return ffi.wasmtime_sharedmemory_size(byref(self.ptr()))

Returns the size, in WebAssembly pages, of this shared memory.

def type(self) ‑> wasmtime._types.MemoryType
Expand source code
def type(self) -> MemoryType:
    """
    Gets the type of this memory as a `MemoryType`
    """

    ptr = ffi.wasmtime_sharedmemory_type(self.ptr())
    return MemoryType._from_ptr(ptr, None)

Gets the type of this memory as a MemoryType

class Store (engine: wasmtime._engine.Engine | None = None, data: Any | None = None)
Expand source code
class Store(Managed["ctypes._Pointer[ffi.wasmtime_store_t]"]):
    __context: "typing.Optional[ctypes._Pointer[ffi.wasmtime_context_t]]"

    def __init__(self, engine: typing.Optional[Engine] = None, data: typing.Optional[typing.Any] = None):

        if engine is None:
            engine = Engine()
        elif not isinstance(engine, Engine):
            raise TypeError("expected an Engine")
        data_id = ffi.c_void_p(0)
        finalize = cast(0, CFUNCTYPE(None, c_void_p))
        if data:
            data_id = value._intern(data)
            finalize = value._externref_finalizer
        self._set_ptr(ffi.wasmtime_store_new(engine.ptr(), data_id, finalize))
        self.__context = ffi.wasmtime_store_context(self.ptr())
        self.engine = engine

    def _delete(self, ptr: "ctypes._Pointer[ffi.wasmtime_store_t]") -> None:
        ffi.wasmtime_store_delete(ptr)
        self.__context = None

    def _context(self) -> "ctypes._Pointer[ffi.wasmtime_context_t]":
        if self.__context is None:
            raise ValueError('already closed')
        return self.__context

    def data(self) -> typing.Optional[typing.Any]:
        """
        TODO
        """
        data = ffi.wasmtime_context_get_data(self._context())
        if data:
            return value._unintern(data)
        else:
            return None

    def gc(self) -> None:
        """
        Runs a GC over `externref` values that have been passed into this Store,
        cleaning out anything that is no longer referenced.

        This is not required to be run manually, but can be done so if you'd
        like more precise control over when unreferenced `externref` values are
        deallocated.
        """
        ffi.wasmtime_context_gc(self._context())

    def set_fuel(self, fuel: int) -> None:
        """
        Sets the amount of fuel in this store to `fuel`.

        This is only relevant when `Config.consume_fuel` is configured.

        This is a required call to ensure that the store has fuel to
        execute WebAssembly since otherwise stores start with zero fuel.

        Raises a `WasmtimeError` if this store's configuration is not configured
        to consume fuel.
        """
        err = ffi.wasmtime_context_set_fuel(self._context(), fuel)
        if err:
            raise WasmtimeError._from_ptr(err)

    def get_fuel(self) -> int:
        """
        Returns the amount of fuel left in the store.

        This is only relevant when `Config.consume_fuel` is configured.

        Raises a `WasmtimeError` if this store's configuration is not configured
        to consume fuel or if the store doesn't have enough fuel remaining.
        """
        remaining = c_uint64(0)
        err = ffi.wasmtime_context_get_fuel(self._context(), byref(remaining))
        if err:
            raise WasmtimeError._from_ptr(err)
        return remaining.value

    def set_wasi(self, wasi: "WasiConfig") -> None:
        """
        TODO
        """
        error = ffi.wasmtime_context_set_wasi(self._context(), wasi._consume())
        if error:
            raise WasmtimeError._from_ptr(error)

    def set_epoch_deadline(self, ticks_after_current: int) -> None:
        """
        Configures the relative epoch deadline, after the current engine's
        epoch, after which WebAssembly code will trap.
        """
        ffi.wasmtime_context_set_epoch_deadline(self._context(), ticks_after_current)

    def set_limits(self,
                   memory_size: int = -1,
                   table_elements: int = -1,
                   instances: int = -1,
                   tables: int = -1,
                   memories: int = -1) -> None:
        """
        Configures the limits of various items within this store.

        * `memory_size` - the maximum size, in bytes, that linear memory is
          allowed to consume within this store. Setting this to a lower value
          will cause instantiation to fail if a module needs more memory.
          Additionally the `memory.grow` instruction will return -1 once this
          threshold is reached.

        * `table_elements` - the maximum number of elements that can be stored
          within tables in this store. Currently each table element takes 8
          bytes.

        * `instances` - the maximum number of WebAssembly instances that can
          be created.

        * `tables` - the maximum number of WebAssembly tables that can
          be created.

        * `memories` - the maximum number of WebAssembly linear memories that
          can be created.

        If any limit is negative then the limit will not be set as a part of
        this invocation and it will be ignored.
        """
        ffi.wasmtime_store_limiter(self.ptr(), memory_size, table_elements, instances, tables, memories)

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Methods

def data(self) ‑> Any | None
Expand source code
def data(self) -> typing.Optional[typing.Any]:
    """
    TODO
    """
    data = ffi.wasmtime_context_get_data(self._context())
    if data:
        return value._unintern(data)
    else:
        return None

TODO

def gc(self) ‑> None
Expand source code
def gc(self) -> None:
    """
    Runs a GC over `externref` values that have been passed into this Store,
    cleaning out anything that is no longer referenced.

    This is not required to be run manually, but can be done so if you'd
    like more precise control over when unreferenced `externref` values are
    deallocated.
    """
    ffi.wasmtime_context_gc(self._context())

Runs a GC over externref values that have been passed into this Store, cleaning out anything that is no longer referenced.

This is not required to be run manually, but can be done so if you'd like more precise control over when unreferenced externref values are deallocated.

def get_fuel(self) ‑> int
Expand source code
def get_fuel(self) -> int:
    """
    Returns the amount of fuel left in the store.

    This is only relevant when `Config.consume_fuel` is configured.

    Raises a `WasmtimeError` if this store's configuration is not configured
    to consume fuel or if the store doesn't have enough fuel remaining.
    """
    remaining = c_uint64(0)
    err = ffi.wasmtime_context_get_fuel(self._context(), byref(remaining))
    if err:
        raise WasmtimeError._from_ptr(err)
    return remaining.value

Returns the amount of fuel left in the store.

This is only relevant when Config.consume_fuel is configured.

Raises a WasmtimeError if this store's configuration is not configured to consume fuel or if the store doesn't have enough fuel remaining.

def set_epoch_deadline(self, ticks_after_current: int) ‑> None
Expand source code
def set_epoch_deadline(self, ticks_after_current: int) -> None:
    """
    Configures the relative epoch deadline, after the current engine's
    epoch, after which WebAssembly code will trap.
    """
    ffi.wasmtime_context_set_epoch_deadline(self._context(), ticks_after_current)

Configures the relative epoch deadline, after the current engine's epoch, after which WebAssembly code will trap.

def set_fuel(self, fuel: int) ‑> None
Expand source code
def set_fuel(self, fuel: int) -> None:
    """
    Sets the amount of fuel in this store to `fuel`.

    This is only relevant when `Config.consume_fuel` is configured.

    This is a required call to ensure that the store has fuel to
    execute WebAssembly since otherwise stores start with zero fuel.

    Raises a `WasmtimeError` if this store's configuration is not configured
    to consume fuel.
    """
    err = ffi.wasmtime_context_set_fuel(self._context(), fuel)
    if err:
        raise WasmtimeError._from_ptr(err)

Sets the amount of fuel in this store to fuel.

This is only relevant when Config.consume_fuel is configured.

This is a required call to ensure that the store has fuel to execute WebAssembly since otherwise stores start with zero fuel.

Raises a WasmtimeError if this store's configuration is not configured to consume fuel.

def set_limits(self,
memory_size: int = -1,
table_elements: int = -1,
instances: int = -1,
tables: int = -1,
memories: int = -1) ‑> None
Expand source code
def set_limits(self,
               memory_size: int = -1,
               table_elements: int = -1,
               instances: int = -1,
               tables: int = -1,
               memories: int = -1) -> None:
    """
    Configures the limits of various items within this store.

    * `memory_size` - the maximum size, in bytes, that linear memory is
      allowed to consume within this store. Setting this to a lower value
      will cause instantiation to fail if a module needs more memory.
      Additionally the `memory.grow` instruction will return -1 once this
      threshold is reached.

    * `table_elements` - the maximum number of elements that can be stored
      within tables in this store. Currently each table element takes 8
      bytes.

    * `instances` - the maximum number of WebAssembly instances that can
      be created.

    * `tables` - the maximum number of WebAssembly tables that can
      be created.

    * `memories` - the maximum number of WebAssembly linear memories that
      can be created.

    If any limit is negative then the limit will not be set as a part of
    this invocation and it will be ignored.
    """
    ffi.wasmtime_store_limiter(self.ptr(), memory_size, table_elements, instances, tables, memories)

Configures the limits of various items within this store.

  • memory_size - the maximum size, in bytes, that linear memory is allowed to consume within this store. Setting this to a lower value will cause instantiation to fail if a module needs more memory. Additionally the memory.grow instruction will return -1 once this threshold is reached.

  • table_elements - the maximum number of elements that can be stored within tables in this store. Currently each table element takes 8 bytes.

  • instances - the maximum number of WebAssembly instances that can be created.

  • tables - the maximum number of WebAssembly tables that can be created.

  • memories - the maximum number of WebAssembly linear memories that can be created.

If any limit is negative then the limit will not be set as a part of this invocation and it will be ignored.

def set_wasi(self,
wasi: WasiConfig)
Expand source code
def set_wasi(self, wasi: "WasiConfig") -> None:
    """
    TODO
    """
    error = ffi.wasmtime_context_set_wasi(self._context(), wasi._consume())
    if error:
        raise WasmtimeError._from_ptr(error)

TODO

class Table (store: wasmtime._store.Store, ty: wasmtime._types.TableType, init: Any)
Expand source code
class Table:
    _table: ffi.wasmtime_table_t

    def __init__(self, store: Store, ty: TableType, init: Any):
        """
        Creates a new table within `store` with the specified `ty`.
        """

        init_val = Val._convert_to_raw(store, ty.element, init)

        table = ffi.wasmtime_table_t()
        error = ffi.wasmtime_table_new(store._context(), ty.ptr(), byref(init_val), byref(table))
        ffi.wasmtime_val_unroot(store._context(), byref(init_val))
        if error:
            raise WasmtimeError._from_ptr(error)
        self._table = table

    @classmethod
    def _from_raw(cls, table: ffi.wasmtime_table_t) -> "Table":
        ty: "Table" = cls.__new__(cls)
        ty._table = table
        return ty

    def type(self, store: Storelike) -> TableType:
        """
        Gets the type of this table as a `TableType`
        """

        ptr = ffi.wasmtime_table_type(store._context(), byref(self._table))
        return TableType._from_ptr(ptr, None)

    def size(self, store: Storelike) -> int:
        """
        Gets the size, in elements, of this table
        """
        return ffi.wasmtime_table_size(store._context(), byref(self._table))

    def grow(self, store: Storelike, amt: int, init: Any) -> int:
        """
        Grows this table by the specified number of slots, using the specified
        initializer for all new table slots.

        Raises a `WasmtimeError` if the table could not be grown.
        Returns the previous size of the table otherwise.
        """
        init_val = Val._convert_to_raw(store, self.type(store).element, init)
        prev = c_uint64(0)
        error = ffi.wasmtime_table_grow(store._context(), byref(self._table), c_uint64(amt), byref(init_val), byref(prev))
        ffi.wasmtime_val_unroot(store._context(), byref(init_val))
        if error:
            raise WasmtimeError._from_ptr(error)
        return prev.value

    def get(self, store: Store, idx: int) -> Optional[Any]:
        """
        Gets an individual element within this table.

        Returns `None` for null references in the table (i.e. a null `funcref`
        or a null `externref).

        Returns a `Func` for non-null `funcref` table elements.

        Returns the wrapped extern data for non-null `externref` table elements.

        Returns `None` if `idx` is out of bounds.
        """
        raw = ffi.wasmtime_val_t()
        ok = ffi.wasmtime_table_get(store._context(), byref(self._table), idx, byref(raw))
        if not ok:
            return None
        val = Val._from_raw(store, raw)
        if val.value:
            return val.value
        else:
            return val

    def set(self, store: Store, idx: int, val: Any) -> None:
        """
        Sets an individual element within this table.

        `idx` must be an integer index.

        The `val` specified must be convertible into this table's element
        type. I.e. for a `funcref` table, `val` must either be a `Func` or
        `None`, and for an `externref` table, `val` may be any arbitrary
        external data.

        Raises a `WasmtimeError` if `idx` is out of bounds.
        """
        value = Val._convert_to_raw(store, self.type(store).element, val)
        error = ffi.wasmtime_table_set(store._context(), byref(self._table), idx, byref(value))
        ffi.wasmtime_val_unroot(store._context(), byref(value))
        if error:
            raise WasmtimeError._from_ptr(error)

    def _as_extern(self) -> ffi.wasmtime_extern_t:
        union = ffi.wasmtime_extern_union(table=self._table)
        return ffi.wasmtime_extern_t(ffi.WASMTIME_EXTERN_TABLE, union)

Creates a new table within store with the specified ty.

Methods

def get(self, store: wasmtime._store.Store, idx: int) ‑> Any | None
Expand source code
def get(self, store: Store, idx: int) -> Optional[Any]:
    """
    Gets an individual element within this table.

    Returns `None` for null references in the table (i.e. a null `funcref`
    or a null `externref).

    Returns a `Func` for non-null `funcref` table elements.

    Returns the wrapped extern data for non-null `externref` table elements.

    Returns `None` if `idx` is out of bounds.
    """
    raw = ffi.wasmtime_val_t()
    ok = ffi.wasmtime_table_get(store._context(), byref(self._table), idx, byref(raw))
    if not ok:
        return None
    val = Val._from_raw(store, raw)
    if val.value:
        return val.value
    else:
        return val

Gets an individual element within this table.

Returns None for null references in the table (i.e. a null funcref or a null `externref).

Returns a Func for non-null funcref table elements.

Returns the wrapped extern data for non-null externref table elements.

Returns None if idx is out of bounds.

def grow(self,
store: wasmtime._store.Store | ForwardRef('Caller'),
amt: int,
init: Any) ‑> int
Expand source code
def grow(self, store: Storelike, amt: int, init: Any) -> int:
    """
    Grows this table by the specified number of slots, using the specified
    initializer for all new table slots.

    Raises a `WasmtimeError` if the table could not be grown.
    Returns the previous size of the table otherwise.
    """
    init_val = Val._convert_to_raw(store, self.type(store).element, init)
    prev = c_uint64(0)
    error = ffi.wasmtime_table_grow(store._context(), byref(self._table), c_uint64(amt), byref(init_val), byref(prev))
    ffi.wasmtime_val_unroot(store._context(), byref(init_val))
    if error:
        raise WasmtimeError._from_ptr(error)
    return prev.value

Grows this table by the specified number of slots, using the specified initializer for all new table slots.

Raises a WasmtimeError if the table could not be grown. Returns the previous size of the table otherwise.

def set(self, store: wasmtime._store.Store, idx: int, val: Any) ‑> None
Expand source code
def set(self, store: Store, idx: int, val: Any) -> None:
    """
    Sets an individual element within this table.

    `idx` must be an integer index.

    The `val` specified must be convertible into this table's element
    type. I.e. for a `funcref` table, `val` must either be a `Func` or
    `None`, and for an `externref` table, `val` may be any arbitrary
    external data.

    Raises a `WasmtimeError` if `idx` is out of bounds.
    """
    value = Val._convert_to_raw(store, self.type(store).element, val)
    error = ffi.wasmtime_table_set(store._context(), byref(self._table), idx, byref(value))
    ffi.wasmtime_val_unroot(store._context(), byref(value))
    if error:
        raise WasmtimeError._from_ptr(error)

Sets an individual element within this table.

idx must be an integer index.

The val specified must be convertible into this table's element type. I.e. for a funcref table, val must either be a Func or None, and for an externref table, val may be any arbitrary external data.

Raises a WasmtimeError if idx is out of bounds.

def size(self,
store: wasmtime._store.Store | ForwardRef('Caller')) ‑> int
Expand source code
def size(self, store: Storelike) -> int:
    """
    Gets the size, in elements, of this table
    """
    return ffi.wasmtime_table_size(store._context(), byref(self._table))

Gets the size, in elements, of this table

def type(self,
store: wasmtime._store.Store | ForwardRef('Caller')) ‑> wasmtime._types.TableType
Expand source code
def type(self, store: Storelike) -> TableType:
    """
    Gets the type of this table as a `TableType`
    """

    ptr = ffi.wasmtime_table_type(store._context(), byref(self._table))
    return TableType._from_ptr(ptr, None)

Gets the type of this table as a TableType

class TableType (valtype: wasmtime._types.ValType, limits: wasmtime._types.Limits)
Expand source code
class TableType(Managed["ctypes._Pointer[ffi.wasm_tabletype_t]"]):
    def __init__(self, valtype: ValType, limits: Limits):
        if not isinstance(limits, Limits):
            raise TypeError("expected Limits")
        type_ptr = take_owned_valtype(valtype)
        ptr = ffi.wasm_tabletype_new(type_ptr, byref(limits.__ffi__()))
        if not ptr:
            raise WasmtimeError("failed to allocate TableType")
        self._set_ptr(ptr)
        self._owner = None

    def _delete(self, ptr: "ctypes._Pointer[ffi.wasm_tabletype_t]") -> None:
        if self._owner is None:
            ffi.wasm_tabletype_delete(ptr)

    @classmethod
    def _from_ptr(cls, ptr: 'ctypes._Pointer[ffi.wasm_tabletype_t]', owner: Optional[Any]) -> "TableType":
        ty: "TableType" = cls.__new__(cls)
        if not isinstance(ptr, POINTER(ffi.wasm_tabletype_t)):
            raise TypeError("wrong pointer type")
        ty._set_ptr(ptr)
        ty._owner = owner
        return ty

    @property
    def element(self) -> ValType:
        """
        Returns the type of this table's elements
        """
        ptr = ffi.wasm_tabletype_element(self.ptr())
        return ValType._from_ptr(ptr, self)

    @property
    def limits(self) -> Limits:
        """
        Returns the limits on the size of thi stable
        """
        val = ffi.wasm_tabletype_limits(self.ptr())
        return Limits._from_ffi(val)

    def _as_extern(self) -> "ctypes._Pointer[ffi.wasm_externtype_t]":
        return ffi.wasm_tabletype_as_externtype_const(self.ptr())

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Instance variables

prop element : wasmtime._types.ValType
Expand source code
@property
def element(self) -> ValType:
    """
    Returns the type of this table's elements
    """
    ptr = ffi.wasm_tabletype_element(self.ptr())
    return ValType._from_ptr(ptr, self)

Returns the type of this table's elements

prop limits : wasmtime._types.Limits
Expand source code
@property
def limits(self) -> Limits:
    """
    Returns the limits on the size of thi stable
    """
    val = ffi.wasm_tabletype_limits(self.ptr())
    return Limits._from_ffi(val)

Returns the limits on the size of thi stable

class Trap (message: str)
Expand source code
class Trap(Exception, Managed["ctypes._Pointer[ffi.wasm_trap_t]"]):

    def __init__(self, message: str):
        """
        Creates a new trap with the given `message`
        """

        vec = message.encode('utf-8')
        self._set_ptr(ffi.wasmtime_trap_new(ffi.create_string_buffer(vec), len(vec)))

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

    @classmethod
    def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasm_trap_t]") -> "Trap":
        if not isinstance(ptr, POINTER(ffi.wasm_trap_t)):
            raise TypeError("wrong pointer type")
        trap: Trap = cls.__new__(cls)
        trap._set_ptr(ptr)
        return trap

    @property
    def message(self) -> str:
        """
        Returns the message for this trap
        """

        message = ffi.wasm_byte_vec_t()
        ffi.wasm_trap_message(self.ptr(), byref(message))
        # subtract one to chop off the trailing nul byte
        message.size -= 1
        ret = ffi.to_str(message)
        message.size += 1
        ffi.wasm_byte_vec_delete(byref(message))
        return ret

    @property
    def frames(self) -> List["Frame"]:
        frames = FrameList(self)
        ffi.wasm_trap_trace(self.ptr(), byref(frames.vec))
        ret = []
        for i in range(0, frames.vec.size):
            ret.append(Frame._from_ptr(frames.vec.data[i], frames))
        return ret

    @property
    def trap_code(self) -> Optional[TrapCode]:
        """
        Returns an optional `TrapCode` that corresponds to why this trap
        happened.

        Note that `None` may be returned for manually created traps which do
        not have an associated code with them.
        """
        code = ffi.wasmtime_trap_code_t()
        if ffi.wasmtime_trap_code(self.ptr(), byref(code)):
            return TrapCode(code.value)
        return None

    def __str__(self) -> str:
        return self.message

Common base class for all non-exit exceptions.

Creates a new trap with the given message

Ancestors

  • builtins.Exception
  • builtins.BaseException
  • wasmtime._managed.Managed
  • typing.Generic

Instance variables

prop frames : List[Frame]
Expand source code
@property
def frames(self) -> List["Frame"]:
    frames = FrameList(self)
    ffi.wasm_trap_trace(self.ptr(), byref(frames.vec))
    ret = []
    for i in range(0, frames.vec.size):
        ret.append(Frame._from_ptr(frames.vec.data[i], frames))
    return ret
prop message : str
Expand source code
@property
def message(self) -> str:
    """
    Returns the message for this trap
    """

    message = ffi.wasm_byte_vec_t()
    ffi.wasm_trap_message(self.ptr(), byref(message))
    # subtract one to chop off the trailing nul byte
    message.size -= 1
    ret = ffi.to_str(message)
    message.size += 1
    ffi.wasm_byte_vec_delete(byref(message))
    return ret

Returns the message for this trap

prop trap_code : wasmtime._trap.TrapCode | None
Expand source code
@property
def trap_code(self) -> Optional[TrapCode]:
    """
    Returns an optional `TrapCode` that corresponds to why this trap
    happened.

    Note that `None` may be returned for manually created traps which do
    not have an associated code with them.
    """
    code = ffi.wasmtime_trap_code_t()
    if ffi.wasmtime_trap_code(self.ptr(), byref(code)):
        return TrapCode(code.value)
    return None

Returns an optional TrapCode that corresponds to why this trap happened.

Note that None may be returned for manually created traps which do not have an associated code with them.

class TrapCode (*args, **kwds)
Expand source code
class TrapCode(Enum):
    # The current stack space was exhausted.
    STACK_OVERFLOW = 0
    # An out-of-bounds memory access.
    MEMORY_OUT_OF_BOUNDS = 1
    # A wasm atomic operation was presented with a not-naturally-aligned linear-memory address.
    HEAP_MISALIGNED = 2
    # An out-of-bounds access to a table.
    TABLE_OUT_OF_BOUNDS = 3
    # Indirect call to a null table entry.
    INDIRECT_CALL_TO_NULL = 4
    # Signature mismatch on indirect call.
    BAD_SIGNATURE = 5
    # An integer arithmetic operation caused an overflow.
    INTEGER_OVERFLOW = 6
    # An integer division by zero.
    INTEGER_DIVISION_BY_ZERO = 7
    # Failed float-to-int conversion.
    BAD_CONVERSION_TO_INTEGER = 8
    # Code that was supposed to have been unreachable was reached.
    UNREACHABLE = 9
    # Execution has potentially run too long and may be interrupted.
    INTERRUPT = 10

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

  • attribute access:

Color.RED

  • value lookup:

Color(1)

  • name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3
>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

  • enum.Enum

Class variables

var BAD_CONVERSION_TO_INTEGER
var BAD_SIGNATURE
var HEAP_MISALIGNED
var INDIRECT_CALL_TO_NULL
var INTEGER_DIVISION_BY_ZERO
var INTEGER_OVERFLOW
var INTERRUPT
var MEMORY_OUT_OF_BOUNDS
var STACK_OVERFLOW
var TABLE_OUT_OF_BOUNDS
var UNREACHABLE
class Val (kind: ctypes.c_ubyte, val: Any)
Expand source code
class Val:
    # We can't let the extern values we wrap `externref`s around be GC'd, so we
    # pin them in `_id_to_extern`. Additionally, we might make multiple
    # `externref`s to the same extern value, so we count how many references
    # we've created in `_id_to_ref_count`, and only remove a value's entry from
    # `_id_to_extern` once the ref count is zero.
    _id_to_extern: typing.Dict[int, typing.Any] = {}
    _id_to_ref_count: typing.Dict[int, int] = {}

    _kind: wasmtime_valkind_t
    _val: typing.Any

    @classmethod
    def i32(cls, val: int) -> "Val":
        """
        Create a new 32-bit integer value
        """
        if not isinstance(val, int):
            raise TypeError("expected an integer")
        val = wasmtime_valunion_t(i32=val).i32
        return Val(WASMTIME_I32, val)

    @classmethod
    def i64(cls, val: int) -> "Val":
        """
        Create a new 64-bit integer value
        """
        if not isinstance(val, int):
            raise TypeError("expected an integer")
        val = wasmtime_valunion_t(i64=val).i64
        return Val(WASMTIME_I64, val)

    @classmethod
    def f32(cls, val: float) -> "Val":
        """
        Create a new 32-bit float value
        """
        if not isinstance(val, float):
            raise TypeError("expected a float")
        val = wasmtime_valunion_t(f32=val).f32
        return Val(WASMTIME_F32, val)

    @classmethod
    def f64(cls, val: float) -> "Val":
        """
        Create a new 64-bit float value
        """
        if not isinstance(val, float):
            raise TypeError("expected a float")
        val = wasmtime_valunion_t(f64=val).f64
        return Val(WASMTIME_F64, val)

    @classmethod
    def externref(cls, extern: typing.Optional[typing.Any]) -> "Val":
        return Val(WASMTIME_EXTERNREF, extern)

    @classmethod
    def funcref(cls, f: "typing.Optional[wasmtime.Func]") -> "Val":
        return Val(WASMTIME_FUNCREF, f)

    @classmethod
    def ref_null(cls, ty: ValType) -> "Val":
        """
        Create a null reference value of the given type.

        Raise an exception if `ty` is not a reference type.
        """
        if ty == ValType.externref():
            return Val.externref(None)
        if ty == ValType.funcref():
            return Val.funcref(None)
        raise WasmtimeError("Invalid reference type for `ref_null`: %s" % ty)

    def __init__(self, kind: wasmtime_valkind_t, val: typing.Any):
        self._kind = kind
        self._val = val

    def __eq__(self, rhs: typing.Any) -> typing.Any:
        if isinstance(rhs, Val):
            return self._kind == rhs._kind and self._val == rhs._val
        return self._val == rhs

    @classmethod
    def _convert_to_raw(cls, store: 'Storelike', ty: ValType, val: Any) -> wasmtime_val_t:
        if isinstance(val, Val):
            if ty != val.type:
                raise TypeError("wrong type of `Val` provided")
            return val._new_raw(store)
        if ty == ValType.externref():
            print('convert to externref', val);
            return Val.externref(val)._new_raw(store)
        elif isinstance(val, int):
            if ty == ValType.i32():
                return Val.i32(val)._new_raw(store)
            if ty == ValType.i64():
                return Val.i64(val)._new_raw(store)
        elif isinstance(val, float):
            if ty == ValType.f32():
                return Val.f32(val)._new_raw(store)
            if ty == ValType.f64():
                return Val.f64(val)._new_raw(store)
        elif isinstance(val, wasmtime.Func):
            return Val.funcref(val)._new_raw(store)
        elif val is None:
            if ty == ValType.externref():
                return Val.externref(None)._new_raw(store)
            if ty == ValType.funcref():
                return Val.funcref(None)._new_raw(store)
        raise TypeError("don't know how to convert %r to %s" % (val, ty))

    def _new_raw(self, store: 'Storelike') -> wasmtime_val_t:
        ret = wasmtime_val_t(kind = self._kind)
        if ret.kind == WASMTIME_I32.value:
            ret.of.i32 = self._val
        elif ret.kind == WASMTIME_I64.value:
            ret.of.i64 = self._val
        elif ret.kind == WASMTIME_F32.value:
            ret.of.f32 = self._val
        elif ret.kind == WASMTIME_F64.value:
            ret.of.f64 = self._val
        elif ret.kind == WASMTIME_EXTERNREF.value:
            if self._val is not None:
                extern_id = _intern(self._val)
                if not wasmtime_externref_new(store._context(), extern_id, _externref_finalizer,
                                              byref(ret.of.externref)):
                    raise WasmtimeError("failed to create an externref value")
            else:

                ret.of.externref.store_id = 0
        elif ret.kind == WASMTIME_FUNCREF.value:
            if self._val is not None:
                ret.of.funcref = self._val._func
        else:
            raise RuntimeError("unknown kind")
        return ret

    @classmethod
    def _from_raw(cls, store: 'Storelike', raw: wasmtime_val_t, owned: bool = True) -> 'Val':
        val: typing.Any = None
        if raw.kind == WASMTIME_I32.value:
            val = raw.of.i32
        elif raw.kind == WASMTIME_I64.value:
            val = raw.of.i64
        elif raw.kind == WASMTIME_F32.value:
            val = raw.of.f32
        elif raw.kind == WASMTIME_F64.value:
            val = raw.of.f64
        elif raw.kind == WASMTIME_EXTERNREF.value:
            if raw.of.externref:
                extern_id = wasmtime_externref_data(store._context(), raw.of.externref)
                val = _unintern(extern_id)
        elif raw.kind == WASMTIME_FUNCREF.value:
            if raw.of.funcref.store_id != 0:
                val = wasmtime.Func._from_raw(raw.of.funcref)
        else:
            raise WasmtimeError("Unkown `wasmtime_valkind_t`: {}".format(raw.kind))

        if owned:
            wasmtime_val_unroot(store._context(), byref(raw))

        return Val(raw.kind, val)

    @property
    def value(self) -> typing.Any:
        """
        Get the the underlying value as a python value
        """
        return self._val

    def as_i32(self) -> typing.Optional[int]:
        """
        Get the 32-bit integer value of this value, or `None` if it's not an i32
        """
        if self._kind == WASMTIME_I32:
            assert(isinstance(self._val, int))
            return self._val
        else:
            return None

    def as_i64(self) -> typing.Optional[int]:
        """
        Get the 64-bit integer value of this value, or `None` if it's not an i64
        """
        if self._kind == WASMTIME_I64:
            assert(isinstance(self._val, int))
            return self._val
        else:
            return None

    def as_f32(self) -> typing.Optional[float]:
        """
        Get the 32-bit float value of this value, or `None` if it's not an f32
        """
        if self._kind == WASMTIME_F32:
            assert(isinstance(self._val, float))
            return self._val
        else:
            return None

    def as_f64(self) -> typing.Optional[float]:
        """
        Get the 64-bit float value of this value, or `None` if it's not an f64
        """
        if self._kind == WASMTIME_F64:
            assert(isinstance(self._val, float))
            return self._val
        else:
            return None

    def as_externref(self) -> typing.Optional[typing.Any]:
        """
        Get the extern data referenced by this `externref` value, or `None` if
        it's not an `externref`.
        """
        if self._kind == WASMTIME_EXTERNREF:
            return self._val
        else:
            return None

    def as_funcref(self) -> typing.Optional["wasmtime.Func"]:
        """
        Get the function that this `funcref` value is referencing, or `None` if
        this is not a `funcref` value, or is a null reference.
        """
        if self._kind == WASMTIME_FUNCREF:
            assert(isinstance(self._val, wasmtime.Func))
            return self._val
        else:
            return None

    @property
    def type(self) -> ValType:
        """
        Returns the `ValType` corresponding to this `Val`
        """
        kind = self._kind
        if kind == WASMTIME_I32:
            return ValType.i32()
        elif kind == WASMTIME_I64:
            return ValType.i64()
        elif kind == WASMTIME_F32:
            return ValType.f32()
        elif kind == WASMTIME_F64:
            return ValType.f64()
        elif kind == WASMTIME_V128:
            raise Exception("unimplemented v128 type")
        elif kind == WASMTIME_EXTERNREF:
            return ValType.externref()
        elif kind == WASMTIME_FUNCREF:
            return ValType.funcref()
        else:
            raise Exception("unknown kind %d" % kind.value)

Static methods

def externref(extern: Any | None) ‑> wasmtime._value.Val
def f32(val: float) ‑> wasmtime._value.Val

Create a new 32-bit float value

def f64(val: float) ‑> wasmtime._value.Val

Create a new 64-bit float value

def funcref(f: typing.Optional[Func]) ‑> wasmtime._value.Val
def i32(val: int) ‑> wasmtime._value.Val

Create a new 32-bit integer value

def i64(val: int) ‑> wasmtime._value.Val

Create a new 64-bit integer value

def ref_null(ty: wasmtime._types.ValType) ‑> wasmtime._value.Val

Create a null reference value of the given type.

Raise an exception if ty is not a reference type.

Instance variables

prop type : wasmtime._types.ValType
Expand source code
@property
def type(self) -> ValType:
    """
    Returns the `ValType` corresponding to this `Val`
    """
    kind = self._kind
    if kind == WASMTIME_I32:
        return ValType.i32()
    elif kind == WASMTIME_I64:
        return ValType.i64()
    elif kind == WASMTIME_F32:
        return ValType.f32()
    elif kind == WASMTIME_F64:
        return ValType.f64()
    elif kind == WASMTIME_V128:
        raise Exception("unimplemented v128 type")
    elif kind == WASMTIME_EXTERNREF:
        return ValType.externref()
    elif kind == WASMTIME_FUNCREF:
        return ValType.funcref()
    else:
        raise Exception("unknown kind %d" % kind.value)

Returns the ValType corresponding to this Val

prop value : Any
Expand source code
@property
def value(self) -> typing.Any:
    """
    Get the the underlying value as a python value
    """
    return self._val

Get the the underlying value as a python value

Methods

def as_externref(self) ‑> Any | None
Expand source code
def as_externref(self) -> typing.Optional[typing.Any]:
    """
    Get the extern data referenced by this `externref` value, or `None` if
    it's not an `externref`.
    """
    if self._kind == WASMTIME_EXTERNREF:
        return self._val
    else:
        return None

Get the extern data referenced by this externref value, or None if it's not an externref.

def as_f32(self) ‑> float | None
Expand source code
def as_f32(self) -> typing.Optional[float]:
    """
    Get the 32-bit float value of this value, or `None` if it's not an f32
    """
    if self._kind == WASMTIME_F32:
        assert(isinstance(self._val, float))
        return self._val
    else:
        return None

Get the 32-bit float value of this value, or None if it's not an f32

def as_f64(self) ‑> float | None
Expand source code
def as_f64(self) -> typing.Optional[float]:
    """
    Get the 64-bit float value of this value, or `None` if it's not an f64
    """
    if self._kind == WASMTIME_F64:
        assert(isinstance(self._val, float))
        return self._val
    else:
        return None

Get the 64-bit float value of this value, or None if it's not an f64

def as_funcref(self) ‑> wasmtime._func.Func | None
Expand source code
def as_funcref(self) -> typing.Optional["wasmtime.Func"]:
    """
    Get the function that this `funcref` value is referencing, or `None` if
    this is not a `funcref` value, or is a null reference.
    """
    if self._kind == WASMTIME_FUNCREF:
        assert(isinstance(self._val, wasmtime.Func))
        return self._val
    else:
        return None

Get the function that this funcref value is referencing, or None if this is not a funcref value, or is a null reference.

def as_i32(self) ‑> int | None
Expand source code
def as_i32(self) -> typing.Optional[int]:
    """
    Get the 32-bit integer value of this value, or `None` if it's not an i32
    """
    if self._kind == WASMTIME_I32:
        assert(isinstance(self._val, int))
        return self._val
    else:
        return None

Get the 32-bit integer value of this value, or None if it's not an i32

def as_i64(self) ‑> int | None
Expand source code
def as_i64(self) -> typing.Optional[int]:
    """
    Get the 64-bit integer value of this value, or `None` if it's not an i64
    """
    if self._kind == WASMTIME_I64:
        assert(isinstance(self._val, int))
        return self._val
    else:
        return None

Get the 64-bit integer value of this value, or None if it's not an i64

class ValType
Expand source code
class ValType(Managed["ctypes._Pointer[ffi.wasm_valtype_t]"]):
    _owner: Optional[Any]

    def _delete(self, ptr: "ctypes._Pointer[ffi.wasm_valtype_t]") -> None:
        # If this is owned by another object we don't free it since that object
        # is responsible for freeing the backing memory.
        if self._owner is None:
            ffi.wasm_valtype_delete(ptr)

    @classmethod
    def i32(cls) -> "ValType":
        ptr = ffi.wasm_valtype_new(ffi.WASM_I32)
        return ValType._from_ptr(ptr, None)

    @classmethod
    def i64(cls) -> "ValType":
        ptr = ffi.wasm_valtype_new(ffi.WASM_I64)
        return ValType._from_ptr(ptr, None)

    @classmethod
    def f32(cls) -> "ValType":
        ptr = ffi.wasm_valtype_new(ffi.WASM_F32)
        return ValType._from_ptr(ptr, None)

    @classmethod
    def f64(cls) -> "ValType":
        ptr = ffi.wasm_valtype_new(ffi.WASM_F64)
        return ValType._from_ptr(ptr, None)

    @classmethod
    def externref(cls) -> "ValType":
        ptr = ffi.wasm_valtype_new(ffi.WASM_ANYREF)
        return ValType._from_ptr(ptr, None)

    @classmethod
    def funcref(cls) -> "ValType":
        ptr = ffi.wasm_valtype_new(ffi.WASM_FUNCREF)
        return ValType._from_ptr(ptr, None)

    def __init__(self) -> None:
        raise WasmtimeError("cannot construct directly")

    @classmethod
    def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasm_valtype_t]", owner: Optional[Any]) -> "ValType":
        if not isinstance(ptr, POINTER(ffi.wasm_valtype_t)):
            raise TypeError("wrong pointer type")
        ty: "ValType" = cls.__new__(cls)
        ty._set_ptr(ptr)
        ty._owner = owner
        return ty

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, ValType):
            return False
        kind1 = ffi.wasm_valtype_kind(self.ptr())
        kind2 = ffi.wasm_valtype_kind(other.ptr())
        return kind1 == kind2

    def __ne__(self, other: object) -> bool:
        return not self.__eq__(other)

    def __repr__(self) -> str:
        return str(self)

    def __str__(self) -> str:
        kind = ffi.wasm_valtype_kind(self.ptr())
        if kind == ffi.WASM_I32.value:
            return 'i32'
        if kind == ffi.WASM_I64.value:
            return 'i64'
        if kind == ffi.WASM_F32.value:
            return 'f32'
        if kind == ffi.WASM_F64.value:
            return 'f64'
        if kind == ffi.WASM_ANYREF.value:
            return 'anyref'
        if kind == ffi.WASM_FUNCREF.value:
            return 'funcref'
        return 'ValType(%d)' % kind.value

    @classmethod
    def _from_list(cls, items: "ctypes._Pointer[ffi.wasm_valtype_vec_t]", owner: Optional[Any]) -> List["ValType"]:
        types = []
        for i in range(0, items.contents.size):
            types.append(ValType._from_ptr(items.contents.data[i], owner))
        return types

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Static methods

def externref() ‑> wasmtime._types.ValType
def f32() ‑> wasmtime._types.ValType
def f64() ‑> wasmtime._types.ValType
def funcref() ‑> wasmtime._types.ValType
def i32() ‑> wasmtime._types.ValType
def i64() ‑> wasmtime._types.ValType
class WasiConfig
Expand source code
class WasiConfig(Managed["ctypes._Pointer[ffi.wasi_config_t]"]):

    def __init__(self) -> None:
        self._set_ptr(ffi.wasi_config_new())

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

    @setter_property
    def argv(self, argv: List[str]) -> None:
        """
        Explicitly configure the `argv` for this WASI configuration
        """
        ptrs = to_char_array(argv)
        if not ffi.wasi_config_set_argv(self.ptr(), len(argv), ptrs):
            raise WasmtimeError("failed to configure argv")

    def inherit_argv(self) -> None:
        ffi.wasi_config_inherit_argv(self.ptr())

    @setter_property
    def env(self, pairs: Iterable[Iterable]) -> None:
        """
        Configure environment variables to be returned for this WASI
        configuration.

        The `pairs` provided must be an iterable list of key/value pairs of
        environment variables.
        """
        names = []
        values = []
        for name, value in pairs:
            names.append(name)
            values.append(value)
        name_ptrs = to_char_array(names)
        value_ptrs = to_char_array(values)
        if not ffi.wasi_config_set_env(self.ptr(), len(names), name_ptrs, value_ptrs):
            raise WasmtimeError("failed to configure environment")

    def inherit_env(self) -> None:
        """
        Configures the environment variables available within WASI to be those
        in this own process's environment. All environment variables are
        inherited.
        """
        ffi.wasi_config_inherit_env(self.ptr())

    @setter_property
    def stdin_file(self, path: Union[str, bytes, PathLike]) -> None:
        """
        Configures a file to be used as the stdin stream of this WASI
        configuration.

        Reads of the stdin stream will read the path specified.

        The file must already exist on the filesystem. If it cannot be
        opened then `WasmtimeError` is raised.
        """

        res = ffi.wasi_config_set_stdin_file(
            self.ptr(), c_char_p(_encode_path(path)))
        if not res:
            raise WasmtimeError("failed to set stdin file")

    def inherit_stdin(self) -> None:
        """
        Configures this own process's stdin to be used as the WASI program's
        stdin.

        Reads of the stdin stream will read this process's stdin.
        """
        ffi.wasi_config_inherit_stdin(self.ptr())

    @setter_property
    def stdout_file(self, path: str) -> None:
        """
        Configures a file to be used as the stdout stream of this WASI
        configuration.

        Writes to stdout will be written to the file specified.

        The file specified will be created if it doesn't exist, or truncated if
        it already exists. It must be available to open for writing. If it
        cannot be opened for writing then `WasmtimeError` is raised.
        """
        res = ffi.wasi_config_set_stdout_file(
            self.ptr(), c_char_p(_encode_path(path)))
        if not res:
            raise WasmtimeError("failed to set stdout file")

    def inherit_stdout(self) -> None:
        """
        Configures this own process's stdout to be used as the WASI program's
        stdout.

        Writes to stdout stream will write to this process's stdout.
        """
        ffi.wasi_config_inherit_stdout(self.ptr())

    @setter_property
    def stderr_file(self, path: str) -> None:
        """
        Configures a file to be used as the stderr stream of this WASI
        configuration.

        Writes to stderr will be written to the file specified.

        The file specified will be created if it doesn't exist, or truncated if
        it already exists. It must be available to open for writing. If it
        cannot be opened for writing then `WasmtimeError` is raised.
        """
        res = ffi.wasi_config_set_stderr_file(
            self.ptr(), c_char_p(_encode_path(path)))
        if not res:
            raise WasmtimeError("failed to set stderr file")

    def inherit_stderr(self) -> None:
        """
        Configures this own process's stderr to be used as the WASI program's
        stderr.

        Writes to stderr stream will write to this process's stderr.
        """
        ffi.wasi_config_inherit_stderr(self.ptr())

    def preopen_dir(self, path: str, guest_path: str, dir_perms: DirPerms = DirPerms.READ_WRITE, file_perms: FilePerms = FilePerms.READ_WRITE) -> None:
        """
        Allows the WASI program to access the directory at `path` using the
        path `guest_path` within the WASI program.

        `dir_perms` specifies the permissions that wasm will have to operate on
        `guest_path`. This can be used, for example, to provide readonly access to a
        directory.

        `file_perms` specifies the maximum set of permissions that can be used for
        any file in this directory.
        """
        path_ptr = c_char_p(path.encode('utf-8'))
        guest_path_ptr = c_char_p(guest_path.encode('utf-8'))
        if not ffi.wasi_config_preopen_dir(self.ptr(), path_ptr, guest_path_ptr, dir_perms.value, file_perms.value):
            raise WasmtimeError('failed to add preopen dir')

Abstract base class for types which contain an owned pointer in the C FFI layer.

Not exported directly from this package.

Ancestors

  • wasmtime._managed.Managed
  • typing.Generic

Instance variables

prop argv
Expand source code
@setter_property
def argv(self, argv: List[str]) -> None:
    """
    Explicitly configure the `argv` for this WASI configuration
    """
    ptrs = to_char_array(argv)
    if not ffi.wasi_config_set_argv(self.ptr(), len(argv), ptrs):
        raise WasmtimeError("failed to configure argv")

Explicitly configure the argv for this WASI configuration

Note that this field can only be set, it cannot be read

prop env
Expand source code
@setter_property
def env(self, pairs: Iterable[Iterable]) -> None:
    """
    Configure environment variables to be returned for this WASI
    configuration.

    The `pairs` provided must be an iterable list of key/value pairs of
    environment variables.
    """
    names = []
    values = []
    for name, value in pairs:
        names.append(name)
        values.append(value)
    name_ptrs = to_char_array(names)
    value_ptrs = to_char_array(values)
    if not ffi.wasi_config_set_env(self.ptr(), len(names), name_ptrs, value_ptrs):
        raise WasmtimeError("failed to configure environment")

Configure environment variables to be returned for this WASI configuration.

The pairs provided must be an iterable list of key/value pairs of environment variables.

Note that this field can only be set, it cannot be read

prop stderr_file
Expand source code
@setter_property
def stderr_file(self, path: str) -> None:
    """
    Configures a file to be used as the stderr stream of this WASI
    configuration.

    Writes to stderr will be written to the file specified.

    The file specified will be created if it doesn't exist, or truncated if
    it already exists. It must be available to open for writing. If it
    cannot be opened for writing then `WasmtimeError` is raised.
    """
    res = ffi.wasi_config_set_stderr_file(
        self.ptr(), c_char_p(_encode_path(path)))
    if not res:
        raise WasmtimeError("failed to set stderr file")

Configures a file to be used as the stderr stream of this WASI configuration.

Writes to stderr will be written to the file specified.

The file specified will be created if it doesn't exist, or truncated if it already exists. It must be available to open for writing. If it cannot be opened for writing then WasmtimeError is raised.

Note that this field can only be set, it cannot be read

prop stdin_file
Expand source code
@setter_property
def stdin_file(self, path: Union[str, bytes, PathLike]) -> None:
    """
    Configures a file to be used as the stdin stream of this WASI
    configuration.

    Reads of the stdin stream will read the path specified.

    The file must already exist on the filesystem. If it cannot be
    opened then `WasmtimeError` is raised.
    """

    res = ffi.wasi_config_set_stdin_file(
        self.ptr(), c_char_p(_encode_path(path)))
    if not res:
        raise WasmtimeError("failed to set stdin file")

Configures a file to be used as the stdin stream of this WASI configuration.

Reads of the stdin stream will read the path specified.

The file must already exist on the filesystem. If it cannot be opened then WasmtimeError is raised.

Note that this field can only be set, it cannot be read

prop stdout_file
Expand source code
@setter_property
def stdout_file(self, path: str) -> None:
    """
    Configures a file to be used as the stdout stream of this WASI
    configuration.

    Writes to stdout will be written to the file specified.

    The file specified will be created if it doesn't exist, or truncated if
    it already exists. It must be available to open for writing. If it
    cannot be opened for writing then `WasmtimeError` is raised.
    """
    res = ffi.wasi_config_set_stdout_file(
        self.ptr(), c_char_p(_encode_path(path)))
    if not res:
        raise WasmtimeError("failed to set stdout file")

Configures a file to be used as the stdout stream of this WASI configuration.

Writes to stdout will be written to the file specified.

The file specified will be created if it doesn't exist, or truncated if it already exists. It must be available to open for writing. If it cannot be opened for writing then WasmtimeError is raised.

Note that this field can only be set, it cannot be read

Methods

def inherit_argv(self) ‑> None
Expand source code
def inherit_argv(self) -> None:
    ffi.wasi_config_inherit_argv(self.ptr())
def inherit_env(self) ‑> None
Expand source code
def inherit_env(self) -> None:
    """
    Configures the environment variables available within WASI to be those
    in this own process's environment. All environment variables are
    inherited.
    """
    ffi.wasi_config_inherit_env(self.ptr())

Configures the environment variables available within WASI to be those in this own process's environment. All environment variables are inherited.

def inherit_stderr(self) ‑> None
Expand source code
def inherit_stderr(self) -> None:
    """
    Configures this own process's stderr to be used as the WASI program's
    stderr.

    Writes to stderr stream will write to this process's stderr.
    """
    ffi.wasi_config_inherit_stderr(self.ptr())

Configures this own process's stderr to be used as the WASI program's stderr.

Writes to stderr stream will write to this process's stderr.

def inherit_stdin(self) ‑> None
Expand source code
def inherit_stdin(self) -> None:
    """
    Configures this own process's stdin to be used as the WASI program's
    stdin.

    Reads of the stdin stream will read this process's stdin.
    """
    ffi.wasi_config_inherit_stdin(self.ptr())

Configures this own process's stdin to be used as the WASI program's stdin.

Reads of the stdin stream will read this process's stdin.

def inherit_stdout(self) ‑> None
Expand source code
def inherit_stdout(self) -> None:
    """
    Configures this own process's stdout to be used as the WASI program's
    stdout.

    Writes to stdout stream will write to this process's stdout.
    """
    ffi.wasi_config_inherit_stdout(self.ptr())

Configures this own process's stdout to be used as the WASI program's stdout.

Writes to stdout stream will write to this process's stdout.

def preopen_dir(self,
path: str,
guest_path: str,
dir_perms: wasmtime._wasi.DirPerms = DirPerms.READ_WRITE,
file_perms: wasmtime._wasi.FilePerms = FilePerms.READ_WRITE) ‑> None
Expand source code
def preopen_dir(self, path: str, guest_path: str, dir_perms: DirPerms = DirPerms.READ_WRITE, file_perms: FilePerms = FilePerms.READ_WRITE) -> None:
    """
    Allows the WASI program to access the directory at `path` using the
    path `guest_path` within the WASI program.

    `dir_perms` specifies the permissions that wasm will have to operate on
    `guest_path`. This can be used, for example, to provide readonly access to a
    directory.

    `file_perms` specifies the maximum set of permissions that can be used for
    any file in this directory.
    """
    path_ptr = c_char_p(path.encode('utf-8'))
    guest_path_ptr = c_char_p(guest_path.encode('utf-8'))
    if not ffi.wasi_config_preopen_dir(self.ptr(), path_ptr, guest_path_ptr, dir_perms.value, file_perms.value):
        raise WasmtimeError('failed to add preopen dir')

Allows the WASI program to access the directory at path using the path guest_path within the WASI program.

dir_perms specifies the permissions that wasm will have to operate on guest_path. This can be used, for example, to provide readonly access to a directory.

file_perms specifies the maximum set of permissions that can be used for any file in this directory.

class WasmtimeError (message: str)
Expand source code
class WasmtimeError(Exception, Managed["ctypes._Pointer[ffi.wasmtime_error_t]"]):
    __message: Optional[str]

    def __init__(self, message: str):
        self.__message = message

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

    @classmethod
    def _from_ptr(cls, ptr: "ctypes._Pointer") -> 'WasmtimeError':
        from . import _ffi as ffi
        if not isinstance(ptr, POINTER(ffi.wasmtime_error_t)):
            raise TypeError("wrong pointer type")

        exit_code = c_int(0)
        if ffi.wasmtime_error_exit_status(ptr, byref(exit_code)):
            exit_trap: ExitTrap = ExitTrap.__new__(ExitTrap)
            exit_trap._set_ptr(ptr)
            exit_trap.__message = None
            exit_trap.code = exit_code.value
            return exit_trap

        err: WasmtimeError = cls.__new__(cls)
        err._set_ptr(ptr)
        err.__message = None
        return err

    def __str__(self) -> str:
        if self.__message:
            return self.__message
        message_vec = ffi.wasm_byte_vec_t()
        ffi.wasmtime_error_message(self.ptr(), byref(message_vec))
        message = ffi.to_str(message_vec)
        ffi.wasm_byte_vec_delete(byref(message_vec))
        return message

Common base class for all non-exit exceptions.

Ancestors

  • builtins.Exception
  • builtins.BaseException
  • wasmtime._managed.Managed
  • typing.Generic

Subclasses

  • wasmtime._error.ExitTrap