diff --git a/pyproject.toml b/pyproject.toml index 5b3b682c2..a44de2c81 100644 --- a/pyproject.toml +++ b/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", diff --git a/src/libtmux/common.py b/src/libtmux/common.py index 06c92320f..ee3737ad1 100644 --- a/src/libtmux/common.py +++ b/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. diff --git a/src/libtmux/pane.py b/src/libtmux/pane.py index e5833111f..b1c8d77bf 100644 --- a/src/libtmux/pane.py +++ b/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) """ diff --git a/src/libtmux/server.py b/src/libtmux/server.py index 30ec0d699..916b6944e 100644 --- a/src/libtmux/server.py +++ b/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. diff --git a/src/libtmux/session.py b/src/libtmux/session.py index ff9a851e7..49e427e2a 100644 --- a/src/libtmux/session.py +++ b/src/libtmux/session.py @@ -24,6 +24,7 @@ from . import exc from .common import ( + AsyncTmuxCmd, EnvironmentMixin, WindowDict, session_check_name, @@ -361,6 +362,62 @@ def cmd( target = self.session_id 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 session context. + + Automatically binds target by adding ``-t`` for object's session ID to the + command. Pass ``target`` to keyword arguments to override. + + Examples + -------- + >>> import asyncio + >>> async def test_acmd(): + ... result = await session.acmd('new-window', '-P') + ... print(result.stdout[0]) + >>> asyncio.run(test_acmd()) + libtmux...:....0 + + From raw output to an enriched `Window` object: + + >>> async def test_from_window(): + ... window_id_result = await session.acmd( + ... 'new-window', '-P', '-F#{window_id}' + ... ) + ... return Window.from_window_id( + ... window_id=window_id_result.stdout[0], + ... server=session.server + ... ) + >>> asyncio.run(test_from_window()) + Window(@... ...:..., Session($1 libtmux_...)) + + Parameters + ---------- + target : str, optional + Optional custom target override. By default, the target is the session ID. + + Returns + ------- + :meth:`server.cmd` + + Notes + ----- + .. versionchanged:: 0.34 + + Passing target by ``-t`` is ignored. Use ``target`` keyword argument instead. + + .. versionchanged:: 0.8 + + Renamed from ``.tmux`` to ``.cmd``. + """ + if target is None: + target = self.session_id + return await self.server.acmd(cmd, *args, target=target) + """ Commands (tmux-like) """ diff --git a/src/libtmux/window.py b/src/libtmux/window.py index f0ce1142d..31743f1ea 100644 --- a/src/libtmux/window.py +++ b/src/libtmux/window.py @@ -28,7 +28,7 @@ from libtmux.pane import Pane from . import exc -from .common import PaneDict, WindowOptionDict +from .common import AsyncTmuxCmd, PaneDict, WindowOptionDict from .options import OptionsMixin if t.TYPE_CHECKING: @@ -304,6 +304,55 @@ 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 window context. + + Automatically binds target by adding ``-t`` for object's window ID to the + command. Pass ``target`` to keyword arguments to override. + + Examples + -------- + Create a pane from a window: + + >>> import asyncio + >>> async def test_acmd(): + ... result = await window.acmd('split-window', '-P', '-F#{pane_id}') + ... print(result.stdout[0]) + >>> asyncio.run(test_acmd()) + %... + + Magic, directly to a `Pane`: + + >>> async def test_from_pane(): + ... pane_id_result = await session.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 window ID. + + Returns + ------- + :meth:`server.cmd` + """ + if target is None: + target = self.window_id + + return await self.server.acmd(cmd, *args, target=target) + """ Commands (tmux-like) """ diff --git a/tests/test_async.py b/tests/test_async.py new file mode 100644 index 000000000..29a55fdf4 --- /dev/null +++ b/tests/test_async.py @@ -0,0 +1,27 @@ +"""Tests for libtmux with :mod`asyncio` support.""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +import pytest + +from libtmux.session import Session + +if TYPE_CHECKING: + from libtmux.server import Server + +logger = logging.getLogger(__name__) + + +@pytest.mark.asyncio +async def test_asyncio(server: Server) -> None: + """Test basic asyncio usage.""" + result = await server.acmd("new-session", "-d", "-P", "-F#{session_id}") + session_id = result.stdout[0] + session = Session.from_session_id( + session_id=session_id, + server=server, + ) + assert isinstance(session, Session) diff --git a/uv.lock b/uv.lock index ed2b5f74b..1d5de8cc2 100644 --- a/uv.lock +++ b/uv.lock @@ -114,6 +114,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/77/f5/21d2de20e8b8b0408f0681956ca2c69f1320a3848ac50e6e7f39c6159675/babel-2.18.0-py3-none-any.whl", hash = "sha256:e2b422b277c2b9a9630c1d7903c2a00d0830c409c59ac8cae9081c92f1aeba35", size = 10196845, upload-time = "2026-02-01T12:30:53.445Z" }, ] +[[package]] +name = "backports-asyncio-runner" +version = "1.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/8e/ff/70dca7d7cb1cbc0edb2c6cc0c38b65cba36cccc491eca64cabd5fe7f8670/backports_asyncio_runner-1.2.0.tar.gz", hash = "sha256:a5aa7b2b7d8f8bfcaa2b57313f70792df84e32a2a746f585213373f900b42162", size = 69893, upload-time = "2025-07-02T02:27:15.685Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a0/59/76ab57e3fe74484f48a53f8e337171b4a2349e506eabe136d7e01d059086/backports_asyncio_runner-1.2.0-py3-none-any.whl", hash = "sha256:0da0a936a8aeb554eccb426dc55af3ba63bcdc69fa1a600b5bb305413a4477b5", size = 12313, upload-time = "2025-07-02T02:27:14.263Z" }, +] + [[package]] name = "beautifulsoup4" version = "4.14.3" @@ -407,7 +416,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "typing-extensions", marker = "python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -634,6 +643,7 @@ dev = [ { name = "gp-sphinx" }, { name = "mypy" }, { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "pytest-cov" }, { name = "pytest-mock" }, { name = "pytest-rerunfailures" }, @@ -662,6 +672,7 @@ lint = [ testing = [ { name = "gp-libs" }, { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "pytest-mock" }, { name = "pytest-rerunfailures" }, { name = "pytest-watcher" }, @@ -683,6 +694,7 @@ dev = [ { name = "gp-sphinx", specifier = "==0.0.1a26" }, { name = "mypy" }, { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "pytest-cov" }, { name = "pytest-mock" }, { name = "pytest-rerunfailures" }, @@ -709,6 +721,7 @@ lint = [ testing = [ { name = "gp-libs" }, { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "pytest-mock" }, { name = "pytest-rerunfailures" }, { name = "pytest-watcher" }, @@ -1028,6 +1041,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d4/24/a372aaf5c9b7208e7112038812994107bc65a84cd00e0354a88c2c77a617/pytest-9.0.3-py3-none-any.whl", hash = "sha256:2c5efc453d45394fdd706ade797c0a81091eccd1d6e4bccfcd476e2b8e0ab5d9", size = 375249, upload-time = "2026-04-07T17:16:16.13Z" }, ] +[[package]] +name = "pytest-asyncio" +version = "1.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "backports-asyncio-runner", marker = "python_full_version < '3.11'" }, + { name = "pytest" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/43/7c/d36d04db312ecf4298932ef77e6e4a9e8ad017906e24e34f0b0c361a2473/pytest_asyncio-1.4.0.tar.gz", hash = "sha256:c6c0d2259945122819f171a32ecea2c349ead889ee28176caaf492143424be42", size = 58514, upload-time = "2026-05-26T09:56:04.083Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/03/e2/08a497ef684b88559c9cc5f4ad53a37e7b99e727094a86d6ea32536d5d3c/pytest_asyncio-1.4.0-py3-none-any.whl", hash = "sha256:933ca923a23075a87fb7070c0ec272a6848489824d887c85c812670932835aa1", size = 16930, upload-time = "2026-05-26T09:56:02.576Z" }, +] + [[package]] name = "pytest-cov" version = "7.1.0"