pytauri.ipc

tauri::ipc

Classes:

Name	Description
ArgumentsType	The bound arguments of a command.
Invoke	tauri::ipc::Invoke
InvokeResolver	tauri::ipc::InvokeResolver
ParametersType	The parameters of a command.
ResolvedArgumentsType	The resolved arguments of a invoked command.
State	A marker for state in ArgumentsType.
InvokeException	Indicates that an exception occurred in a command. Similar to Rust's Result::Err.
Commands	This class provides features similar to tauri::generate_handler.
JavaScriptChannelId	This class is a wrapper around pytauri.ffi.ipc.JavaScriptChannelId.
Channel	This class is a wrapper around pytauri.ffi.ipc.Channel.

Attributes:

Name	Type	Description
Headers		http::header::HeaderMap::iter

Headers module-attribute

Headers = TypeAliasType('Headers', list[tuple[bytes, bytes]])

http::header::HeaderMap::iter

(key, value) pairs of headers.

Each key will be yielded once per associated value. So, if a key has 3 associated values, it will be yielded 3 times.

[(b"key0", b"value00"), (b"key0", b"value01"), (b"key1", b"value1")]

Tip

You can use libraries like multidict or httpx.Headers to convert it into dict for more efficient retrieval of a specific header.

__all__ module-attribute

__all__ = ['ArgumentsType', 'Channel', 'Commands', 'Headers', 'Invoke', 'InvokeException', 'InvokeResolver', 'JavaScriptChannelId', 'ParametersType', 'ResolvedArgumentsType', 'State']

ArgumentsType

Bases: _BaseArgumentsType

The bound arguments of a command.

Each key is optional, depending on the keys of the bound ResolvedArgumentsType.

You can use it like **kwargs, for example command(**arguments).

Attributes:

Name	Type	Description
body	bytes	The body of this ipc message.
app_handle	AppHandle	The handle of the app.
webview_window	WebviewWindow	The WebviewWindow of this Invoke.
headers	Headers	The headers of this ipc message.

body instance-attribute

body: bytes

The body of this ipc message.

app_handle instance-attribute

app_handle: AppHandle

The handle of the app.

webview_window instance-attribute

webview_window: WebviewWindow

The WebviewWindow of this Invoke.

headers instance-attribute

headers: Headers

The headers of this ipc message.

Invoke

tauri::ipc::Invoke

Methods:

Name	Description
bind_to	Consumes this Invoke and binds parameters.
resolve	Consumes this Invoke and resolves the command with the given value.
reject	Consumes this Invoke and rejects the command with the given value.

Attributes:

Name	Type	Description
command	str	The name of the current command.

command property

command: str

The name of the current command.

bind_to

bind_to(parameters: ParametersType) -> Optional[InvokeResolver[_ArgumentsTypeVar]]

Consumes this Invoke and binds parameters.

If the frontend illegally calls the IPC, this method will automatically reject this Invoke and return None.

The return value InvokeResolver.arguments is not the same object as the input parameters.

Source code in python/pytauri/src/pytauri/ffi/ipc.py

def bind_to(
    self, parameters: ParametersType
) -> Optional["InvokeResolver[_ArgumentsTypeVar]"]:
    """Consumes this `Invoke` and binds parameters.

    If the frontend illegally calls the IPC,
    this method will automatically reject this `Invoke` and return `None`.

    The return value [InvokeResolver.arguments][pytauri.ffi.ipc.InvokeResolver.arguments]
    is not the same object as the input `parameters`.
    """

resolve

resolve(value: _InvokeResponseBody) -> None

Consumes this Invoke and resolves the command with the given value.

Parameters:

Name	Type	Description	Default
value	_InvokeResponseBody	The value to resolve the command with. If str, it will be serialized as JSON on the frontend. If bytes, it will be sent as ArrayBuffer to the frontend.	required

Source code in python/pytauri/src/pytauri/ffi/ipc.py

def resolve(self, value: _InvokeResponseBody) -> None:
    """Consumes this `Invoke` and resolves the command with the given value.

    Args:
        value: The value to resolve the command with.

            - If `str`, it will be serialized as JSON on the frontend.
            - If `bytes`, it will be sent as `ArrayBuffer` to the frontend.
    """
    ...

reject

reject(value: str) -> None

Consumes this Invoke and rejects the command with the given value.

Source code in python/pytauri/src/pytauri/ffi/ipc.py

def reject(self, value: str) -> None:
    """Consumes this `Invoke` and rejects the command with the given value."""
    ...

InvokeResolver

Bases: Generic[_ArgumentsTypeVar]

tauri::ipc::InvokeResolver

Methods:

Name	Description
resolve	Consumes this InvokeResolver and resolves the command with the given value.
reject	Consumes this InvokeResolver and rejects the command with the given value.

Attributes:

Name	Type	Description
arguments	_ArgumentsTypeVar	The bound arguments of the current command.

arguments property

arguments: _ArgumentsTypeVar

The bound arguments of the current command.

resolve

resolve(value: _InvokeResponseBody) -> None

Consumes this InvokeResolver and resolves the command with the given value.

Parameters:

Name	Type	Description	Default
value	_InvokeResponseBody	The value to resolve the command with. If str, it will be serialized as JSON on the frontend. If bytes, it will be sent as ArrayBuffer to the frontend.	required

Source code in python/pytauri/src/pytauri/ffi/ipc.py

def resolve(self, value: _InvokeResponseBody) -> None:
    """Consumes this `InvokeResolver` and resolves the command with the given value.

    Args:
        value: The value to resolve the command with.

            - If `str`, it will be serialized as JSON on the frontend.
            - If `bytes`, it will be sent as `ArrayBuffer` to the frontend.
    """

reject

reject(value: str) -> None

Consumes this InvokeResolver and rejects the command with the given value.

Source code in python/pytauri/src/pytauri/ffi/ipc.py

def reject(self, value: str) -> None:
    """Consumes this `InvokeResolver` and rejects the command with the given value."""
    ...

ParametersType

Bases: TypedDict

The parameters of a command.

You can use this with Invoke.bind_to. All keys are optional. If a key exists, it will be assigned a value corresponding to ResolvedArgumentsType.

Attributes:

Name	Type	Description
body	Any	Whatever. We just use the key, not the value.
app_handle	Any	Whatever. We just use the key, not the value.
webview_window	Any	Whatever. We just use the key, not the value.
headers	Any	Whatever. We just use the key, not the value.
states	dict[str, type[Any]]	A dictionary of state classes.

body instance-attribute

body: Any

Whatever. We just use the key, not the value.

app_handle instance-attribute

app_handle: Any

Whatever. We just use the key, not the value.

webview_window instance-attribute

webview_window: Any

Whatever. We just use the key, not the value.

headers instance-attribute

headers: Any

Whatever. We just use the key, not the value.

states instance-attribute

states: dict[str, type[Any]]

A dictionary of state classes.

ResolvedArgumentsType

Bases: _BaseArgumentsType

The resolved arguments of a invoked command.

It contains the resolved values of the parameters in ParametersType.

Attributes:

Name	Type	Description
body	bytes	The body of this ipc message.
app_handle	AppHandle	The handle of the app.
webview_window	WebviewWindow	The WebviewWindow of this Invoke.
headers	Headers	The headers of this ipc message.
states	dict[str, Any]	The value is an instance of the state class in ParametersType.states.

body instance-attribute

body: bytes

The body of this ipc message.

app_handle instance-attribute

app_handle: AppHandle

The handle of the app.

webview_window instance-attribute

webview_window: WebviewWindow

The WebviewWindow of this Invoke.

headers instance-attribute

headers: Headers

The headers of this ipc message.

states instance-attribute

states: dict[str, Any]

The value is an instance of the state class in ParametersType.states.

State

A marker for state in ArgumentsType.

Examples

from typing import Annotated

from pytauri import Commands, State

commands = Commands()


class MyState:
    foo: int


@commands.command()
async def command(state: Annotated[MyState, State()]) -> int:
    return state.foo

InvokeException

InvokeException(value: str)

Bases: Exception

Indicates that an exception occurred in a command. Similar to Rust's Result::Err.

When this exception is raised in a command, pytauri will return it to the frontend through Invoke.reject(value) and will not log the exception on the python side.

Attributes:

Name	Type	Description
value	str	The error message that will be returned to the frontend.

Source code in python/pytauri/src/pytauri/ipc.py

def __init__(self, value: str) -> None:  # noqa: D107
    self.value = value

value instance-attribute

value: str = value

The error message that will be returned to the frontend.

Commands

Commands(*, experimental_gen_ts: bool = False)

Bases: UserDict[str, _PyInvokHandleData]

This class provides features similar to tauri::generate_handler.

Typically, you would use Commands.command to register a command handler function. Then, use Commands.generate_handler to get an invoke_handler for use with BuilderArgs.

Methods:

Name	Description
generate_handler	This method is similar to tauri::generate_handler.
wrap_pyfunc	Wrap a Callable to conform to the definition of PyHandlerType.
parse_parameters	Check the signature of a Callable and return the parameters.
set_command	Set a command handler.
command	A decorator to register a command handler.
experimental_gen_ts	Generate TypeScript types and API client from the registered commands.
experimental_gen_ts_background	Generate TypeScript types and API client from the registered commands in the background.

Source code in python/pytauri/src/pytauri/ipc.py

def __init__(self, *, experimental_gen_ts: bool = False) -> None:  # noqa: D107
    super().__init__()

    self._experimental_gen_ts = {} if experimental_gen_ts else None

    data = self.data

    async def _async_invoke_handler(invoke: Invoke) -> None:
        # NOTE:
        # - the implementer of this function must not raise exceptions
        # - and must ensure to fulfill `invoke/resolver`
        resolver = None
        try:
            command = invoke.command
            handler_data = data.get(command)
            if handler_data is None:
                invoke.reject(f"no python handler `{command}` found")
                return

            parameters = handler_data.parameters
            handler = handler_data.handler

            resolver: Optional[InvokeResolver[ResolvedArgumentsType]] = (
                invoke.bind_to(parameters)
            )
            if resolver is None:
                # `invoke` has already been rejected
                return

            try:
                arguments = resolver.arguments
                states = arguments.pop("states", {})
                # PERF: `**ArgumentsType(**arguments, **states)`
                resp = await handler(**arguments, **states)
                # TODO, PERF: idk if this will block?
            except InvokeException as e:
                resolver.reject(e.value)
            except Exception as e:
                # # TODO: Should we return the traceback to the frontend?
                # # It might leak information.
                # from traceback import format_exc
                # resolver.reject(format_exc())
                _logger.exception(
                    f"invoke_handler {handler}: `{handler.__name__}` raised an exception",
                    exc_info=e,
                )
                resolver.reject(repr(e))
            else:
                resolver.resolve(resp)

        except Exception as e:
            msg = f"{_async_invoke_handler} implementation raised an exception, please report this as a pytauri bug"

            _logger.critical(msg, exc_info=e)
            if resolver is not None:
                resolver.reject(msg)
            else:
                invoke.reject(msg)
            raise

    self._async_invoke_handler = _async_invoke_handler

generate_handler

generate_handler(portal: BlockingPortal) -> _InvokeHandlerProto

This method is similar to tauri::generate_handler.

You can use this method to get invoke_handler for use with BuilderArgs.

Examples:

from anyio.from_thread import start_blocking_portal

commands = Commands()

with start_blocking_portal(backend) as portal:
    invoke_handler = commands.generate_handler(portal)
    ...

Warning

The portal must remain valid while the returned invoke_handler is being used.

Source code in python/pytauri/src/pytauri/ipc.py

def generate_handler(self, portal: BlockingPortal, /) -> _InvokeHandlerProto:
    """This method is similar to [tauri::generate_handler](https://docs.rs/tauri/latest/tauri/macro.generate_handler.html).

    You can use this method to get `invoke_handler` for use with [BuilderArgs][pytauri.BuilderArgs].

    Examples:
        ```py
        from anyio.from_thread import start_blocking_portal

        commands = Commands()

        with start_blocking_portal(backend) as portal:
            invoke_handler = commands.generate_handler(portal)
            ...
        ```

    !!! warning
        The `portal` must remain valid while the returned `invoke_handler` is being used.
    """
    async_invoke_handler = self._async_invoke_handler

    def invoke_handler(invoke: Invoke) -> None:
        # NOTE:
        # - `invoke_handler` must not raise exception
        # - must not block

        # this func will be call in extern thread, so it's ok to use `start_task_soon`
        portal.start_task_soon(async_invoke_handler, invoke)

    return invoke_handler

wrap_pyfunc

wrap_pyfunc(pyfunc: _WrappablePyHandlerType, *, _gen_ts_cmd: Optional[str] = None) -> _PyHandlerType

Wrap a Callable to conform to the definition of PyHandlerType.

Specifically:

• If pyfunc has a KEYWORD_ONLY parameter named body:
  • If body is bytes: do nothing.
  • If issubclass(body, BaseModel): wrap this callable as a new function with a body: bytes parameter.
  • Otherwise: try to convert it to a BaseModel/TypeAdapter, and proceed as in the BaseModel branch.
• Handle the return type:
  • If the return type is bytes: do nothing.
  • If issubclass(return_type, BaseModel): wrap this callable as a new function with return: str value.
  • Otherwise: try to convert it to a BaseModel/TypeAdapter, and proceed as in the BaseModel branch.
• If no wrapping is needed, the original pyfunc will be returned.

The pyfunc will be decorated using functools.wraps, and its __signature__ will also be updated.

Source code in python/pytauri/src/pytauri/ipc.py

def wrap_pyfunc(  # noqa: C901, PLR0912, PLR0915  # TODO: simplify this method
    self, pyfunc: _WrappablePyHandlerType, *, _gen_ts_cmd: Optional[str] = None
) -> _PyHandlerType:
    """Wrap a `Callable` to conform to the definition of PyHandlerType.

    Specifically:

    - If `pyfunc` has a `KEYWORD_ONLY` parameter named `body`:
        - If `body` is `bytes`:
            do nothing.
        - If `issubclass(body, BaseModel)`:
            wrap this callable as a new function with a `body: bytes` parameter.
        - Otherwise:
            try to convert it to a `BaseModel`/`TypeAdapter`, and proceed as in the `BaseModel` branch.
    - Handle the return type:
        - If the return type is `bytes`:
            do nothing.
        - If `issubclass(return_type, BaseModel)`:
            wrap this callable as a new function with `return: str` value.
        - Otherwise:
            try to convert it to a `BaseModel`/`TypeAdapter`, and proceed as in the `BaseModel` branch.
    - If no wrapping is needed, the original `pyfunc` will be returned.

    The `pyfunc` will be decorated using [functools.wraps][], and its `__signature__` will also be updated.
    """
    serializer: Optional[_Serializer[Union[BaseModel, Any]]] = None
    deserializer: Optional[_Deserializer[Union[BaseModel, Any]]] = None
    input_type = NoneType
    output_type = NoneType

    body_key = "body"

    # TODO: drop py39 support: <https://github.com/pytauri/pytauri/issues/249#issuecomment-3123432528>
    if sys.version_info >= (3, 10):
        sig = signature(pyfunc, eval_str=True)
    else:
        sig = signature(pyfunc)
    parameters = sig.parameters
    return_annotation = sig.return_annotation

    body_param = parameters.get(body_key)
    if body_param is not None:
        if body_param.kind not in {
            Parameter.KEYWORD_ONLY,
            Parameter.POSITIONAL_OR_KEYWORD,
        }:
            raise ValueError(
                f"Expected `{body_key}` to be KEYWORD parameter, but got `{body_param.kind}` parameter"
            )

        body_type = body_param.annotation
        if body_type is Parameter.empty:
            raise ValueError(
                f"Expected `{body_key}` to have type annotation, but got empty"
            )
        elif body_type is bytes:
            serializer = None
            input_type = bytes
        # `Annotated`, `Union`, `None`, etc are not `type`
        elif isinstance(body_type, type) and issubclass(body_type, BaseModel):
            serializer = body_type.model_validate_json
            input_type = body_type
        else:
            # PERF, FIXME: `cast` make pyright happy, it mistakenly thinks this is `Any | type[Unknown]`
            body_type = cast(Any, body_type)
            try:
                model_serde = _type_to_model(body_type)
            except Exception as e:
                raise ValueError(
                    f"Failed to convert `{body_type}` type to pydantic Model, "
                    f"please explicitly use {BaseModel} or {bytes} as `{body_key}` type annotation instead."
                ) from e
            serializer = model_serde.serializer
            input_type = model_serde.model

    if return_annotation is Signature.empty:
        raise ValueError(
            "Expected the return value to have type annotation, but got empty. "
            "Please explicitly use `def foo() -> None:` instead."
        )
    elif return_annotation is bytes:
        deserializer = None
        output_type = bytes
    # `Annotated`, `Union`, `None`, etc are not `type`
    elif isinstance(return_annotation, type) and issubclass(
        return_annotation, BaseModel
    ):
        # PERF: maybe we should cache this closure?
        def _deserializer(data: BaseModel) -> str:
            return data.model_dump_json()

        deserializer = _deserializer
        output_type = return_annotation
    else:
        # PERF, FIXME: `cast` make pyright happy, it mistakenly thinks this is `Any | type[Unknown]`
        return_annotation = cast(Any, return_annotation)
        try:
            model_serde = _type_to_model(return_annotation)
        except Exception as e:
            raise ValueError(
                f"Failed to convert `{return_annotation}` type to pydantic Model, "
                f"please explicitly use {BaseModel} or {bytes} as return type annotation instead."
            ) from e
        deserializer = model_serde.deserializer
        output_type = model_serde.model

    if not serializer and not deserializer:
        if _gen_ts_cmd is not None:
            assert self._experimental_gen_ts is not None
            self._experimental_gen_ts[_gen_ts_cmd] = InputOutput(
                cmd_handler=pyfunc, input_type=input_type, output_type=output_type
            )
        return cast(_PyHandlerType, pyfunc)  # `cast` make typing happy

    # NOTE: Use `wraps` to ensure the docstring is preserved correctly
    @wraps(pyfunc)
    async def wrapper(*args: Any, **kwargs: Any) -> _InvokeResponseBody:
        nonlocal serializer, deserializer

        if serializer is not None:
            body_bytes = kwargs[body_key]
            assert isinstance(body_bytes, bytes)  # PERF
            try:
                body_arg = serializer(body_bytes)
            except ValidationError as e:
                raise InvokeException(repr(e)) from e
            kwargs[body_key] = body_arg

        resp = await pyfunc(*args, **kwargs)

        if deserializer is not None:
            # - subclass of `BaseModel`
            # - other types that are not `bytes`
            assert not isinstance(resp, bytes)  # PERF
            return deserializer(resp)
        else:
            assert isinstance(resp, bytes)  # PERF
            return resp

    new_parameters = parameters.copy()
    new_return_annotation = return_annotation
    if serializer is not None:
        new_parameters[body_key] = parameters[body_key].replace(annotation=bytes)
    if deserializer is not None:
        new_return_annotation = str

    # see: <https://docs.python.org/3.13/library/inspect.html#inspect.signature>
    wrapper.__signature__ = sig.replace(  # pyright: ignore[reportAttributeAccessIssue]
        parameters=tuple(new_parameters.values()),
        return_annotation=new_return_annotation,
    )

    if _gen_ts_cmd is not None:
        assert self._experimental_gen_ts is not None
        self._experimental_gen_ts[_gen_ts_cmd] = InputOutput(
            cmd_handler=wrapper, input_type=input_type, output_type=output_type
        )
    return wrapper

parse_parameters staticmethod

parse_parameters(pyfunc: _PyHandlerType) -> ParametersType

Check the signature of a Callable and return the parameters.

Check if the Signature of pyfunc conforms to ArgumentsType, and if the return value is bytes or str.

Parameters:

Name	Type	Description	Default
pyfunc	_PyHandlerType	The Callable to check.	required

Returns:

Type	Description
ParametersType	The parameters of the pyfunc. You can use it with Invoke.bind_to.

Raises:

Type	Description
ValueError	If the signature does not conform to the expected pattern.

Source code in python/pytauri/src/pytauri/ipc.py

@staticmethod
def parse_parameters(pyfunc: _PyHandlerType, /) -> ParametersType:
    """Check the signature of a `Callable` and return the parameters.

    Check if the [Signature][inspect.Signature] of `pyfunc` conforms to [ArgumentsType][pytauri.ipc.ArgumentsType],
    and if the return value is [bytes][] or [str][].

    Args:
        pyfunc: The `Callable` to check.

    Returns:
        The parameters of the `pyfunc`. You can use it with [Invoke.bind_to][pytauri.ipc.Invoke.bind_to].

    Raises:
        ValueError: If the signature does not conform to the expected pattern.
    """
    # TODO: drop py39 support: <https://github.com/pytauri/pytauri/issues/249#issuecomment-3123432528>
    if sys.version_info >= (3, 10):
        sig = signature(pyfunc, eval_str=True)
    else:
        sig = signature(pyfunc)
    parameters = sig.parameters
    return_annotation = sig.return_annotation
    handled_parameters: ParametersType = {}

    known_arguments_type = {
        "body": bytes,
        "app_handle": AppHandle,
        "webview_window": WebviewWindow,
        "headers": Headers,
    }

    for name, param in parameters.items():
        # check if the `parameters` type hint conforms to [pytauri.ipc.ArgumentsType][]

        kind = param.kind
        annotation = param.annotation

        correct_annotation = known_arguments_type.get(name)
        if correct_annotation is not None:
            if kind not in {
                Parameter.KEYWORD_ONLY,
                Parameter.POSITIONAL_OR_KEYWORD,
            }:
                raise ValueError(
                    f"Expected `{name}` to be KEYWORD parameter, but got `{kind}` parameter"
                )
            if annotation is not correct_annotation:
                raise ValueError(
                    f"Expected `{name}` to be `{correct_annotation}`, but got `{annotation}`"
                )
            handled_parameters[name] = annotation
        elif get_origin(annotation) is Annotated:
            [origin, hint, *_] = get_args(annotation)
            if not isinstance(origin, type) or not isinstance(hint, State):
                raise ValueError(
                    f"Expected `{name}` to be `{Annotated[type[Any], State()]}`, but got `{annotation}`"
                )
            handled_parameters.setdefault("states", {})[name] = origin
        else:
            raise ValueError(
                f"Unexpected parameter `{name}`, expected one of {tuple(known_arguments_type.keys())} or `{Annotated}`"
            )

    if return_annotation not in {bytes, str}:
        raise ValueError(
            f"Expected `return_annotation` to be {bytes} or {str}, got `{return_annotation}`"
        )

    return handled_parameters

set_command

set_command(command: str, handler: _WrappablePyHandlerType) -> None

Set a command handler.

This method internally calls parse_parameters and wrap_pyfunc, parse_parameters(wrap_pyfunc(handler)).

Source code in python/pytauri/src/pytauri/ipc.py

def set_command(
    self,
    command: str,
    handler: _WrappablePyHandlerType,
    /,
) -> None:
    """Set a command handler.

    This method internally calls [parse_parameters][pytauri.Commands.parse_parameters]
    and [wrap_pyfunc][pytauri.Commands.wrap_pyfunc], `parse_parameters(wrap_pyfunc(handler))`.
    """
    new_handler = self.wrap_pyfunc(
        handler,
        _gen_ts_cmd=command if self._experimental_gen_ts is not None else None,
    )
    parameters = self.parse_parameters(new_handler)
    self.data[command] = _PyInvokHandleData(parameters, new_handler)

command

command(command: Optional[str] = None) -> _RegisterType[_WrappablePyHandlerTypeVar]

A decorator to register a command handler.

Examples:

commands = Commands()


@commands.command()
async def my_command(body: FooModel, app_handle: AppHandle) -> BarModel: ...


@commands.command("foo_command")
async def my_command2(
    body: FooModel, app_handle: AppHandle
) -> BarModel: ...

This method internally calls set_command, which means the function signature must conform to ArgumentsType.

Parameters:

Name	Type	Description	Default
command	Optional[str]	The name of the command. If not provided, the __name__ of callable will be used.	None

Raises:

Type	Description
ValueError	If a command with the same name already exists. If it's expected, use set_command instead.

Source code in python/pytauri/src/pytauri/ipc.py

def command(
    self, command: Optional[str] = None, /
) -> _RegisterType[_WrappablePyHandlerTypeVar]:
    """A [decorator](https://docs.python.org/3/glossary.html#term-decorator) to register a command handler.

    Examples:
        ```py
        commands = Commands()


        @commands.command()
        async def my_command(body: FooModel, app_handle: AppHandle) -> BarModel: ...


        @commands.command("foo_command")
        async def my_command2(
            body: FooModel, app_handle: AppHandle
        ) -> BarModel: ...
        ```

    This method internally calls [set_command][pytauri.Commands.set_command],
    which means the function signature must conform to [ArgumentsType][pytauri.ipc.ArgumentsType].

    Args:
        command: The name of the command. If not provided, the `__name__` of `callable` will be used.

    Raises:
        ValueError: If a command with the same name already exists.
            If it's expected, use [set_command][pytauri.Commands.set_command] instead.
    """
    if command is None:
        return self._register
    else:
        return partial(self._register, command=command)

experimental_gen_ts async

experimental_gen_ts(output_dir: Union[str, PathLike[str]], json2ts_cmd: str, *, cmd_alias: Optional[Callable[[str], str]] = to_camel) -> None

Generate TypeScript types and API client from the registered commands.

This method is only available if experimental_gen_ts is set to True when creating the Commands instance.

Parameters:

Name	Type	Description	Default
output_dir	Union[str, PathLike[str]]	The directory to output the generated TypeScript files.	required
json2ts_cmd	str	The command to run json-schema-to-typescript to generate TypeScript types.	required
cmd_alias	Optional[Callable[[str], str]]	An optional function to convert command names to TypeScript style. By default, it uses to_camel.	to_camel

Raises:

Type	Description
RuntimeError	If experimental_gen_ts is not enabled when creating the Commands instance.

Source code in python/pytauri/src/pytauri/ipc.py

async def experimental_gen_ts(
    self,
    output_dir: Union[str, PathLike[str]],
    json2ts_cmd: str,
    *,
    cmd_alias: Optional[Callable[[str], str]] = to_camel,
) -> None:
    """Generate TypeScript types and API client from the registered commands.

    This method is only available if `experimental_gen_ts` is set to `True`
    when creating the `Commands` instance.

    Args:
        output_dir: The directory to output the generated TypeScript files.
        json2ts_cmd: The command to run [json-schema-to-typescript] to generate TypeScript types.
            [json-schema-to-typescript]: https://github.com/bcherny/json-schema-to-typescript/
        cmd_alias: An optional function to convert command names to TypeScript style.
            By default, it uses [to_camel][pydantic.alias_generators.to_camel].

    Raises:
        RuntimeError: If `experimental_gen_ts` is not enabled when creating the `Commands`
            instance.
    """
    if self._experimental_gen_ts is None:
        raise RuntimeError(
            "Experimental TypeScript generation is not enabled. "
            "Please set `experimental_gen_ts=True` when creating `Commands`."
        )
    cmd_in_out = self._experimental_gen_ts
    del self._experimental_gen_ts  # release memory

    await gen_ts(cmd_in_out, output_dir, json2ts_cmd, cmd_alias=cmd_alias)

experimental_gen_ts_background async

experimental_gen_ts_background(output_dir: Union[str, PathLike[str]], json2ts_cmd: str, *, cmd_alias: Optional[Callable[[str], str]] = to_camel) -> None

Generate TypeScript types and API client from the registered commands in the background.

See experimental_gen_ts for more details. This method is equivalent to:

try:
    await self.experimental_gen_ts(
        output_dir,
        json2ts_cmd,
        cmd_alias=cmd_alias,
    )
except Exception:
    logging.exception(...)
    return None

Source code in python/pytauri/src/pytauri/ipc.py

async def experimental_gen_ts_background(
    self,
    output_dir: Union[str, PathLike[str]],
    json2ts_cmd: str,
    *,
    cmd_alias: Optional[Callable[[str], str]] = to_camel,
) -> None:
    """Generate TypeScript types and API client from the registered commands in the background.

    See [experimental_gen_ts][pytauri.ipc.Commands.experimental_gen_ts] for more details.
    This method is equivalent to:

    ```python
    try:
        await self.experimental_gen_ts(
            output_dir,
            json2ts_cmd,
            cmd_alias=cmd_alias,
        )
    except Exception:
        logging.exception(...)
        return None
    ```
    """
    start_time = current_time()
    _logger.info("Generating TypeScript types and API client in the background.")
    try:
        await self.experimental_gen_ts(
            output_dir,
            json2ts_cmd,
            cmd_alias=cmd_alias,
        )
    except Exception:
        _logger.exception("Failed to generate TypeScript types and API client.")
        return None

    elapsed = current_time() - start_time
    _logger.info(
        f"Finished generating TypeScript types and API client in {elapsed:.2f} seconds."
    )

JavaScriptChannelId

Bases: RootModel[_FFIJavaScriptChannelIdAnno], Generic[_ModelTypeVar]

This class is a wrapper around pytauri.ffi.ipc.JavaScriptChannelId.

You can use this class as model field in pydantic model directly, or use it as model directly.

pytauri.ffi.ipc.JavaScriptChannelId can't be used directly in pydantic model.

Examples

from asyncio import Task, create_task, sleep
from typing import Any

from pydantic import BaseModel, RootModel
from pydantic.networks import HttpUrl
from pytauri import Commands
from pytauri.ipc import JavaScriptChannelId, WebviewWindow

commands = Commands()

Progress = RootModel[int]


class Download(BaseModel):
    url: HttpUrl
    channel: JavaScriptChannelId[Progress]


background_tasks: set[Task[Any]] = set()


@commands.command()
async def download(body: Download, webview_window: WebviewWindow) -> None:
    channel = body.channel.channel_on(webview_window.as_ref_webview())

    async def task():
        progress = Progress(0)
        while progress.root <= 100:
            channel.send_model(progress)
            await sleep(0.1)
            progress.root += 1

    t = create_task(task())
    background_tasks.add(t)
    t.add_done_callback(background_tasks.discard)


# Or you can use it as `body` model directly
@commands.command()
async def my_command(body: JavaScriptChannelId) -> bytes: ...

Methods:

Name	Description
from_str	See pytauri.ffi.ipc.JavaScriptChannelId.from_str.
channel_on	See pytauri.ffi.ipc.JavaScriptChannelId.channel_on.

from_str classmethod

from_str(value: str) -> Self

See pytauri.ffi.ipc.JavaScriptChannelId.from_str.

Source code in python/pytauri/src/pytauri/ipc.py

@classmethod
def from_str(cls, value: str, /) -> Self:
    """See [pytauri.ffi.ipc.JavaScriptChannelId.from_str][]."""
    ffi_js_channel_id = _FFIJavaScriptChannelId.from_str(value)
    return cls(ffi_js_channel_id)

channel_on

channel_on(webview: Webview) -> Channel[_ModelTypeVar]

See pytauri.ffi.ipc.JavaScriptChannelId.channel_on.

Source code in python/pytauri/src/pytauri/ipc.py

def channel_on(self, webview: Webview, /) -> "Channel[_ModelTypeVar]":
    """See [pytauri.ffi.ipc.JavaScriptChannelId.channel_on][]."""
    ffi_channel = self.root.channel_on(webview)
    return Channel(ffi_channel)

Channel

Channel(ffi_channel: Channel)

Bases: Generic[_ModelTypeVar]

This class is a wrapper around pytauri.ffi.ipc.Channel.

It adds the following methods:

• send_model

Examples

See JavaScriptChannelId

Methods:

Name	Description
id	See pytauri.ffi.ipc.Channel.id.
send	See pytauri.ffi.ipc.Channel.send.
send_model	Equivalent to self.send(model.model_dump_json()).

Source code in python/pytauri/src/pytauri/ipc.py

def __init__(self, ffi_channel: _FFIChannel, /):  # noqa: D107
    self._ffi_channel = ffi_channel

id

id() -> int

See pytauri.ffi.ipc.Channel.id.

Source code in python/pytauri/src/pytauri/ipc.py

def id(self, /) -> int:
    """See [pytauri.ffi.ipc.Channel.id][]."""
    return self._ffi_channel.id()

send

send(data: _InvokeResponseBody) -> None

See pytauri.ffi.ipc.Channel.send.

Source code in python/pytauri/src/pytauri/ipc.py

def send(self, data: _InvokeResponseBody, /) -> None:
    """See [pytauri.ffi.ipc.Channel.send][]."""
    self._ffi_channel.send(data)

send_model

send_model(model: _ModelTypeVar) -> None

Equivalent to self.send(model.model_dump_json()).

Source code in python/pytauri/src/pytauri/ipc.py

def send_model(self, model: _ModelTypeVar, /) -> None:
    """Equivalent to `self.send(model.model_dump_json())`."""
    self.send(model.model_dump_json())