asyncio: Asynchronous libtmux

pyproject.toml

@@ -61,6 +61,7 @@ dev = [
   "typing-extensions; python_version < '3.11'",
   "gp-libs",
   "pytest",
+  "pytest-asyncio",
   "pytest-rerunfailures",
   "pytest-mock",
   "pytest-watcher",
@@ -84,6 +85,7 @@ testing = [
   "typing-extensions; python_version < '3.11'",
   "gp-libs",
   "pytest",
+  "pytest-asyncio",
   "pytest-rerunfailures",
   "pytest-mock",
   "pytest-watcher",

src/libtmux/common.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import functools
 import logging
 import re
@@ -378,6 +379,143 @@ def __init__(self, *args: t.Any, tmux_bin: str | None = None) -> None:
             )
 
 
+class AsyncTmuxCmd:
+    """An asyncio-compatible class for running any tmux command via subprocess.
+
+    Attributes
+    ----------
+    cmd : list[str]
+        The full command (including the "tmux" binary path).
+    stdout : list[str]
+        Lines of stdout output from tmux.
+    stderr : list[str]
+        Lines of stderr output from tmux.
+    returncode : int
+        The process return code.
+
+    Examples
+    --------
+    >>> import asyncio
+    >>>
+    >>> async def main():
+    ...     proc = await AsyncTmuxCmd.run('-V')
+    ...     if proc.stderr:
+    ...         raise exc.LibTmuxException(
+    ...             f"Error invoking tmux: {proc.stderr}"
+    ...         )
+    ...     print("tmux version:", proc.stdout)
+    ...
+    >>> asyncio.run(main())
+        tmux version: [...]
+
+    This is equivalent to calling:
+
+    .. code-block:: console
+
+        $ tmux -V
+    """
+
+    def __init__(
+        self,
+        cmd: list[str],
+        stdout: list[str],
+        stderr: list[str],
+        returncode: int,
+    ) -> None:
+        """Store the results of a completed tmux subprocess run.
+
+        Parameters
+        ----------
+        cmd : list[str]
+            The command used to invoke tmux.
+        stdout : list[str]
+            Captured lines from tmux stdout.
+        stderr : list[str]
+            Captured lines from tmux stderr.
+        returncode : int
+            Subprocess exit code.
+        """
+        self.cmd: list[str] = cmd
+        self.stdout: list[str] = stdout
+        self.stderr: list[str] = stderr
+        self.returncode: int = returncode
+
+    @classmethod
+    async def run(cls, *args: t.Any) -> AsyncTmuxCmd:
+        """Execute a tmux command asynchronously and capture its output.
+
+        Parameters
+        ----------
+        *args : str
+            Arguments to be passed after the "tmux" binary name.
+
+        Returns
+        -------
+        AsyncTmuxCmd
+            An instance containing the cmd, stdout, stderr, and returncode.
+
+        Raises
+        ------
+        exc.TmuxCommandNotFound
+            If no "tmux" executable is found in the user's PATH.
+        exc.LibTmuxException
+            If there's any unexpected exception creating or communicating
+            with the tmux subprocess.
+        """
+        tmux_bin: str | None = shutil.which("tmux")
+        if not tmux_bin:
+            msg = "tmux executable not found in PATH"
+            raise exc.TmuxCommandNotFound(
+                msg,
+            )
+
+        cmd: list[str] = [tmux_bin] + [str(c) for c in args]
+
+        try:
+            process: asyncio.subprocess.Process = await asyncio.create_subprocess_exec(
+                *cmd,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            stdout_bytes, stderr_bytes = await process.communicate()
+            returncode: int = (
+                process.returncode if process.returncode is not None else -1
+            )
+
+        except Exception as e:
+            logger.exception("Exception for %s", " ".join(cmd))
+            msg = f"Exception while running tmux command: {e}"
+            raise exc.LibTmuxException(
+                msg,
+            ) from e
+
+        # Decode bytes to string with error handling
+        stdout = stdout_bytes.decode(errors="backslashreplace")
+        stderr = stderr_bytes.decode(errors="backslashreplace")
+
+        # Split on newlines and filter empty lines
+        stdout_split: list[str] = stdout.split("\n")
+        # remove trailing newlines from stdout
+        while stdout_split and stdout_split[-1] == "":
+            stdout_split.pop()
+
+        stderr_split = stderr.split("\n")
+        stderr_split = list(filter(None, stderr_split))
+
+        if "has-session" in cmd and stderr_split and not stdout_split:
+            stdout_split = [stderr_split[0]]
+
+        logger.debug("stdout for %s: %s", " ".join(cmd), stdout_split)
+        logger.debug("stderr for %s: %s", " ".join(cmd), stderr_split)
+
+        return cls(
+            cmd=cmd,
+            stdout=stdout_split,
+            stderr=stderr_split,
+            returncode=returncode,
+        )
+
+
 @functools.cache
 def get_version(tmux_bin: str | None = None) -> LooseVersion:
     """Return tmux version.

src/libtmux/pane.py

@@ -14,7 +14,7 @@
 import warnings
 
 from libtmux import exc
-from libtmux.common import has_gte_version, raise_if_stderr, tmux_cmd
+from libtmux.common import AsyncTmuxCmd, has_gte_version, raise_if_stderr, tmux_cmd
 from libtmux.constants import (
     PANE_DIRECTION_FLAG_MAP,
     RESIZE_ADJUSTMENT_DIRECTION_FLAG_MAP,
@@ -220,6 +220,53 @@ def cmd(
 
         return self.server.cmd(cmd, *args, target=target)
 
+    async def acmd(
+        self,
+        cmd: str,
+        *args: t.Any,
+        target: str | int | None = None,
+    ) -> AsyncTmuxCmd:
+        """Execute tmux subcommand within pane context.
+
+        Automatically binds target by adding  ``-t`` for object's pane ID to the
+        command. Pass ``target`` to keyword arguments to override.
+
+        Examples
+        --------
+        >>> import asyncio
+        >>> async def test_acmd():
+        ...     result = await pane.acmd('split-window', '-P')
+        ...     print(result.stdout[0])
+        >>> asyncio.run(test_acmd())
+        libtmux...:...
+
+        From raw output to an enriched `Pane` object:
+
+        >>> async def test_from_pane():
+        ...     pane_id_result = await pane.acmd(
+        ...         'split-window', '-P', '-F#{pane_id}'
+        ...     )
+        ...     return Pane.from_pane_id(
+        ...         pane_id=pane_id_result.stdout[0],
+        ...         server=session.server
+        ...     )
+        >>> asyncio.run(test_from_pane())
+        Pane(%... Window(@... ...:..., Session($1 libtmux_...)))
+
+        Parameters
+        ----------
+        target : str, optional
+            Optional custom target override. By default, the target is the pane ID.
+
+        Returns
+        -------
+        :meth:`server.cmd`
+        """
+        if target is None:
+            target = self.pane_id
+
+        return await self.server.acmd(cmd, *args, target=target)
+
     """
     Commands (tmux-like)
     """

src/libtmux/server.py

@@ -27,6 +27,7 @@
 from libtmux.window import Window
 
 from .common import (
+    AsyncTmuxCmd,
     EnvironmentMixin,
     PaneDict,
     SessionDict,
@@ -311,8 +312,12 @@ def cmd(
 
         Output of `tmux -L ... new-window -P -F#{window_id}` to a `Window` object:
 
-        >>> Window.from_window_id(window_id=session.cmd(
-        ... 'new-window', '-P', '-F#{window_id}').stdout[0], server=session.server)
+        >>> Window.from_window_id(
+        ...     window_id=session.cmd(
+        ...         'new-window', '-P', '-F#{window_id}'
+        ...     ).stdout[0],
+        ...     server=session.server,
+        ... )
         Window(@4 3:..., Session($1 libtmux_...))
 
         Create a pane from a window:
@@ -323,7 +328,9 @@ def cmd(
         Output of `tmux -L ... split-window -P -F#{pane_id}` to a `Pane` object:
 
         >>> Pane.from_pane_id(pane_id=window.cmd(
-        ... 'split-window', '-P', '-F#{pane_id}').stdout[0], server=window.server)
+        ...     'split-window', '-P', '-F#{pane_id}').stdout[0],
+        ...     server=window.server
+        ... )
         Pane(%... Window(@... ...:..., Session($1 libtmux_...)))
 
         Parameters
@@ -361,6 +368,90 @@ def cmd(
 
         return tmux_cmd(*svr_args, *cmd_args, tmux_bin=self.tmux_bin)
 
+    async def acmd(
+        self,
+        cmd: str,
+        *args: t.Any,
+        target: str | int | None = None,
+    ) -> AsyncTmuxCmd:
+        """Execute tmux command respective of socket name and file, return output.
+
+        Examples
+        --------
+        >>> import asyncio
+        >>> async def test_acmd():
+        ...     result = await server.acmd('display-message', 'hi')
+        ...     print(result.stdout)
+        >>> asyncio.run(test_acmd())
+        []
+
+        New session:
+
+        >>> async def test_new_session():
+        ...     result = await server.acmd(
+        ...         'new-session', '-d', '-P', '-F#{session_id}'
+        ...     )
+        ...     print(result.stdout[0])
+        >>> asyncio.run(test_new_session())
+        $...
+
+        Output of `tmux -L ... new-window -P -F#{window_id}` to a `Window` object:
+
+        >>> async def test_new_window():
+        ...     result = await session.acmd('new-window', '-P', '-F#{window_id}')
+        ...     window_id = result.stdout[0]
+        ...     window = Window.from_window_id(window_id=window_id, server=server)
+        ...     print(window)
+        >>> asyncio.run(test_new_window())
+        Window(@... ...:..., Session($... libtmux_...))
+
+        Create a pane from a window:
+
+        >>> async def test_split_window():
+        ...     result = await server.acmd('split-window', '-P', '-F#{pane_id}')
+        ...     print(result.stdout[0])
+        >>> asyncio.run(test_split_window())
+        %...
+
+        Output of `tmux -L ... split-window -P -F#{pane_id}` to a `Pane` object:
+
+        >>> async def test_pane():
+        ...     result = await window.acmd('split-window', '-P', '-F#{pane_id}')
+        ...     pane_id = result.stdout[0]
+        ...     pane = Pane.from_pane_id(pane_id=pane_id, server=server)
+        ...     print(pane)
+        >>> asyncio.run(test_pane())
+        Pane(%... Window(@... ...:..., Session($1 libtmux_...)))
+
+        Parameters
+        ----------
+        target : str, optional
+            Optional custom target.
+
+        Returns
+        -------
+        :class:`common.AsyncTmuxCmd`
+        """
+        svr_args: list[str | int] = [cmd]
+        cmd_args: list[str | int] = []
+        if self.socket_name:
+            svr_args.insert(0, f"-L{self.socket_name}")
+        if self.socket_path:
+            svr_args.insert(0, f"-S{self.socket_path}")
+        if self.config_file:
+            svr_args.insert(0, f"-f{self.config_file}")
+        if self.colors:
+            if self.colors == 256:
+                svr_args.insert(0, "-2")
+            elif self.colors == 88:
+                svr_args.insert(0, "-8")
+            else:
+                raise exc.UnknownColorOption
+
+        cmd_args = ["-t", str(target), *args] if target is not None else [*args]
+
+        return await AsyncTmuxCmd.run(*svr_args, *cmd_args)
+
     @property
     def attached_sessions(self) -> list[Session]:
         """Return active :class:`Session` instances.