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: Union[str, bytes]) ‑> bytearray

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): 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

Global configuration, used to create an Engine.

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

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)

Ancestors

wasmtime._managed.Managed
typing.Generic

Instance variables

prop cache

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

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)

prop consume_fuel

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

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)

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

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

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)

prop epoch_interruption

Configures whether wasm execution can be interrupted via epoch increments.

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

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)

prop parallel_compilation

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

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)

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

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

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))

prop wasm_bulk_memory

Configures whether the wasm bulk memory proposal is enabled.

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

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)

prop wasm_memory64

Configures whether the wasm memory64 proposal is enabled.

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

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)

prop wasm_multi_memory

Configures whether the wasm multi memory proposal is enabled.

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

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)

prop wasm_multi_value

Configures whether the wasm multi value proposal is enabled.

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

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)

prop wasm_reference_types

Configures whether the wasm reference types proposal is enabled.

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

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)

prop wasm_relaxed_simd

Configures whether the wasm relaxed simd proposal is enabled.

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

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)

prop wasm_relaxed_simd_deterministic

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

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)

prop wasm_simd

Configures whether the wasm SIMD proposal is enabled.

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

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)

prop wasm_tail_call

Configures whether the wasm tail call proposal is enabled.

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

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)

prop wasm_threads

Configures whether the wasm threads proposal is enabled.

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

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)

class Engine (config: Optional[wasmtime._config.Config] = None)

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

Not exported directly from this package.

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())

Ancestors

wasmtime._managed.Managed
typing.Generic

Methods

def increment_epoch(self) ‑> None

class ExitTrap (message: str)

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.

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

Ancestors

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

Class variables

var code : int

class ExportType

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

Not exported directly from this package.

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)

Ancestors

wasmtime._managed.Managed
typing.Generic

Instance variables

prop name : str

Returns the name in the modulethis export type refers to

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)

prop type : AsExternType

Returns the type that this export refers to

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)

class Frame

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

Not exported directly from this package.

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())

Ancestors

wasmtime._managed.Managed
typing.Generic

Instance variables

prop func_index : int

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

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())

prop func_name : Optional[str]

Returns the name of the function this frame corresponds to

May return None if no name can be inferred

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

prop func_offset : int

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

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())

prop module_name : Optional[str]

Returns the name of the module this frame corresponds to

May return None if no name can be inferred

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

prop module_offset : int

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

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())

class Func (store: Union[wasmtime._store.Store, ForwardRef('Caller')], ty: wasmtime._types.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.

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)

Methods

def type(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')]) ‑> wasmtime._types.FuncType: Gets the type of this func as a FuncType

class FuncType (params: List[wasmtime._types.ValType], results: List[wasmtime._types.ValType])

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

Not exported directly from this package.

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())

Ancestors

wasmtime._managed.Managed
typing.Generic

Instance variables

prop params : List[ValType]

Returns the list of parameter types for this function type

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)

prop results : List[ValType]

Returns the list of result types for this function type

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)

class Global (store: Union[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: Union[wasmtime._store.Store, ForwardRef('Caller')], val: Any) ‑> None: Sets the value of this global to a new value
def type(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')]) ‑> wasmtime._types.GlobalType: Gets the type of this global as a GlobalType
def value(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')]) ‑> Any: Gets the current value of this global

Returns a native python type

class GlobalType (valtype: wasmtime._types.ValType, mutable: bool)

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

Not exported directly from this package.

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())

Ancestors

wasmtime._managed.Managed
typing.Generic

Instance variables

prop content : wasmtime._types.ValType

Returns the type this global contains

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)

prop mutable : bool

Returns whether this global is mutable or not

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

class ImportType

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

Not exported directly from this package.

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)

Ancestors

wasmtime._managed.Managed
typing.Generic

Instance variables

prop module : str

Returns the module this import type refers to

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)

prop name : 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.

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

prop type : AsExternType

Returns the type that this import refers to

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)

class Instance (store: Union[wasmtime._store.Store, ForwardRef('Caller')], module: wasmtime._module.Module, imports: Sequence[Union[ForwardRef('Func'), ForwardRef('Table'), ForwardRef('Memory'), ForwardRef('Global'), ForwardRef('Instance'), ForwardRef('Module')]])

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.

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)

Methods

def exports(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')]) ‑> wasmtime._instance.InstanceExports: 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: Optional[int])

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)

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.

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")

Ancestors

wasmtime._managed.Managed
typing.Generic

Class variables

var engine : wasmtime._engine.Engine

Instance variables

prop allow_shadowing

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

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

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)

Methods

def define(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')], module: str, name: str, item: Union[ForwardRef('Func'), ForwardRef('Table'), ForwardRef('Memory'), ForwardRef('Global'), ForwardRef('Instance'), ForwardRef('Module')]): 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: 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: Union[wasmtime._store.Store, ForwardRef('Caller')], name: str, instance: wasmtime._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.
def define_module(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')], name: str, module: wasmtime._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.
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.
def get(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')], module: str, name: str): 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: Union[wasmtime._store.Store, ForwardRef('Caller')], name: str) ‑> wasmtime._func.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: Union[wasmtime._store.Store, ForwardRef('Caller')], module: wasmtime._module.Module) ‑> wasmtime._instance.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: Union[wasmtime._store.Store, ForwardRef('Caller')], ty: wasmtime._types.MemoryType)

Creates a new memory in store with the given ty

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)

Methods

def data_len(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')]) ‑> int: Returns the raw byte length of this memory.
def data_ptr(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')]): 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: Union[wasmtime._store.Store, ForwardRef('Caller')], size: 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
def grow(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')], delta: int) ‑> int: Grows this memory by the given number of pages
def read(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')], start: Optional[int] = 0, stop: 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.
def size(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')]) ‑> int: Returns the size, in WebAssembly pages, of this memory.
def type(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')]) ‑> wasmtime._types.MemoryType: Gets the type of this memory as a MemoryType
def write(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')], value: Union[bytearray, bytes], start: 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

class MemoryType (limits: wasmtime._types.Limits, is_64: bool = False)

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

Not exported directly from this package.

Expand source code

class MemoryType(Managed["ctypes._Pointer[ffi.wasm_memorytype_t]"]):
    def __init__(self, limits: Limits, is_64: 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)
        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())

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

Ancestors

wasmtime._managed.Managed
typing.Generic

Instance variables

prop is_64 : bool

Returns whether or not this is a 64-bit memory

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())

prop limits : wasmtime._types.Limits

Returns the limits on the size of this table

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)

class Module (engine: wasmtime._engine.Engine, wasm: Union[str, bytes])

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

Not exported directly from this package.

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)

Ancestors

wasmtime._managed.Managed
typing.Generic

Static methods

def deserialize(engine: wasmtime._engine.Engine, encoded: Union[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: Union[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: Union[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]

Returns the types of the exports that this module has

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

prop imports : List[wasmtime._types.ImportType]

Returns the types of imports that this module has

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

Methods

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.

class Store (engine: Optional[wasmtime._engine.Engine] = None, data: Optional[Any] = None)

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

Not exported directly from this package.

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)

Ancestors

wasmtime._managed.Managed
typing.Generic

Methods

def data(self) ‑> Optional[Any]

TODO

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.

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.

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.

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.

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.

def set_wasi(self, wasi: WasiConfig)

TODO

class Table (store: wasmtime._store.Store, ty: wasmtime._types.TableType, init: Any)

Creates a new table within store with the specified ty.

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_uint32(0)
        error = ffi.wasmtime_table_grow(store._context(), byref(self._table), c_uint32(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)

Methods

def get(self, store: wasmtime._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.
def grow(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')], 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.
def set(self, store: wasmtime._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.
def size(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')]) ‑> int: Gets the size, in elements, of this table
def type(self, store: Union[wasmtime._store.Store, ForwardRef('Caller')]) ‑> wasmtime._types.TableType: Gets the type of this table as a TableType

class TableType (valtype: wasmtime._types.ValType, limits: wasmtime._types.Limits)

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

Not exported directly from this package.

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())

Ancestors

wasmtime._managed.Managed
typing.Generic

Instance variables

prop element : wasmtime._types.ValType

Returns the type of this table's elements

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)

prop limits : wasmtime._types.Limits

Returns the limits on the size of thi stable

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)

class Trap (message: str)

Common base class for all non-exit exceptions.

Creates a new trap with the given message

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

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

Returns the message for this trap

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

prop trap_code : Optional[wasmtime._trap.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.

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

class TrapCode (*args, **kwds)

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.

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

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: Optional[Any]) ‑> 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

Returns the ValType corresponding to this Val

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)

prop value : Any

Get the the underlying value as a python value

Expand source code

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

Methods

def as_externref(self) ‑> Optional[Any]: Get the extern data referenced by this externref value, or None if it's not an externref.
def as_f32(self) ‑> Optional[float]: Get the 32-bit float value of this value, or None if it's not an f32
def as_f64(self) ‑> Optional[float]: Get the 64-bit float value of this value, or None if it's not an f64
def as_funcref(self) ‑> Optional[wasmtime._func.Func]: 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) ‑> Optional[int]: Get the 32-bit integer value of this value, or None if it's not an i32
def as_i64(self) ‑> Optional[int]: Get the 64-bit integer value of this value, or None if it's not an i64

class ValType

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

Not exported directly from this package.

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

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

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

Not exported directly from this package.

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) -> None:
        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):
            raise WasmtimeError('failed to add preopen dir')

Ancestors

wasmtime._managed.Managed
typing.Generic

Instance variables

prop argv

Explicitly configure the argv for this WASI configuration

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

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")

prop env

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

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")

prop 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

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")

prop 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

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")

prop 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

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")

Methods

def inherit_argv(self) ‑> None
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.
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.
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.
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.
def preopen_dir(self, path: str, guest_path: str) ‑> None

class WasmtimeError (message: str)

Common base class for all non-exit exceptions.

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

Ancestors

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

Subclasses

wasmtime._error.ExitTrap