feat: make tmux_cmd_async awaitable for typing

Author: tony

conftest.py

@@ -10,12 +10,15 @@
 
 from __future__ import annotations
 
+import asyncio
 import shutil
 import typing as t
 
 import pytest
+import pytest_asyncio
 from _pytest.doctest import DoctestItem
 
+from libtmux.common_async import get_version, tmux_cmd_async
 from libtmux.pane import Pane
 from libtmux.pytest_plugin import USING_ZSH
 from libtmux.server import Server
@@ -50,7 +53,6 @@ def add_doctest_fixtures(
 
         # Add async support for async doctests
         doctest_namespace["asyncio"] = asyncio
-        from libtmux.common_async import tmux_cmd_async, get_version
         doctest_namespace["tmux_cmd_async"] = tmux_cmd_async
         doctest_namespace["get_version"] = get_version
 
@@ -83,9 +85,6 @@ def setup_session(
 
 # Async test fixtures
 # These require pytest-asyncio to be installed
-import asyncio
-
-import pytest_asyncio
 
 
 @pytest_asyncio.fixture

examples/async_demo.py

@@ -7,16 +7,18 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import sys
+import time
 from pathlib import Path
 
 # Try importing from installed package, fallback to development mode
 try:
-    from libtmux.common_async import tmux_cmd_async, get_version
+    from libtmux.common_async import get_version, tmux_cmd_async
 except ImportError:
     # Development mode: add parent to path
     sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
-    from libtmux.common_async import tmux_cmd_async, get_version
+    from libtmux.common_async import get_version, tmux_cmd_async
 
 
 async def demo_basic_command() -> None:
@@ -60,7 +62,7 @@ async def demo_concurrent_commands() -> None:
     )
 
     commands = ["list-sessions", "list-windows", "list-panes", "show-options -g"]
-    for cmd, result in zip(commands, results):
+    for cmd, result in zip(commands, results, strict=True):
         if isinstance(result, Exception):
             print(f"\n[{cmd}] Error: {result}")
         else:
@@ -75,7 +77,6 @@ async def demo_comparison_with_sync() -> None:
     print("Demo 3: Performance Comparison")
     print("=" * 60)
 
-    import time
     from libtmux.common import tmux_cmd
 
     # Commands to run
@@ -95,10 +96,8 @@ async def demo_comparison_with_sync() -> None:
     print("\nSync execution (sequential)...")
     start = time.time()
     for cmd in commands:
-        try:
+        with contextlib.suppress(Exception):
             tmux_cmd(*cmd.split())
-        except Exception:
-            pass
     sync_time = time.time() - start
     print(f"  Time: {sync_time:.4f} seconds")
 
@@ -155,6 +154,7 @@ async def main() -> None:
     except Exception as e:
         print(f"\nDemo failed with error: {e}")
         import traceback
+
         traceback.print_exc()
 
 

examples/hybrid_async_demo.py

@@ -12,18 +12,17 @@
 
 import asyncio
 import sys
+import time
 from pathlib import Path
 
 # Try importing from installed package, fallback to development mode
 try:
-    from libtmux.common import AsyncTmuxCmd
-    from libtmux.common_async import tmux_cmd_async, get_version
+    from libtmux.common_async import get_version, tmux_cmd_async
     from libtmux.server import Server
 except ImportError:
     # Development mode: add parent to path
     sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
-    from libtmux.common import AsyncTmuxCmd
-    from libtmux.common_async import tmux_cmd_async, get_version
+    from libtmux.common_async import get_version, tmux_cmd_async
     from libtmux.server import Server
 
 
@@ -56,13 +55,24 @@ async def demo_pattern_a_acmd_methods() -> None:
 
     # Get session details
     print("\n2. Getting session details...")
-    result = await server.acmd("display-message", "-p", "-t", session_id, "-F#{session_name}")
+    result = await server.acmd(
+        "display-message",
+        "-p",
+        "-t",
+        session_id,
+        "-F#{session_name}",
+    )
     session_name = result.stdout[0] if result.stdout else "unknown"
     print(f"   Session name: {session_name}")
 
     # List windows
     print("\n3. Listing windows in session...")
-    result = await server.acmd("list-windows", "-t", session_id, "-F#{window_index}:#{window_name}")
+    result = await server.acmd(
+        "list-windows",
+        "-t",
+        session_id,
+        "-F#{window_index}:#{window_name}",
+    )
     print(f"   Found {len(result.stdout)} windows")
     for window in result.stdout:
         print(f"   - {window}")
@@ -182,12 +192,10 @@ async def demo_performance_comparison() -> None:
     print("=" * 70)
     print()
 
-    import time
-
     # Create test sessions
     print("Setting up test sessions...")
     sessions = []
-    for i in range(4):
+    for _ in range(4):
         cmd = await tmux_cmd_async("new-session", "-d", "-P", "-F#{session_id}")
         sessions.append(cmd.stdout[0])
     print(f"Created {len(sessions)} test sessions")
@@ -203,10 +211,9 @@ async def demo_performance_comparison() -> None:
     # Parallel execution
     print("\n2. Parallel execution (all at once)...")
     start = time.time()
-    await asyncio.gather(*[
-        tmux_cmd_async("list-windows", "-t", session_id)
-        for session_id in sessions
-    ])
+    await asyncio.gather(
+        *[tmux_cmd_async("list-windows", "-t", session_id) for session_id in sessions]
+    )
     parallel_time = time.time() - start
     print(f"   Time: {parallel_time:.4f} seconds")
 
@@ -216,10 +223,9 @@ async def demo_performance_comparison() -> None:
 
     # Cleanup
     print("\nCleaning up test sessions...")
-    await asyncio.gather(*[
-        tmux_cmd_async("kill-session", "-t", session_id)
-        for session_id in sessions
-    ])
+    await asyncio.gather(
+        *[tmux_cmd_async("kill-session", "-t", session_id) for session_id in sessions]
+    )
 
 
 async def main() -> None:
@@ -265,6 +271,7 @@ async def main() -> None:
     except Exception as e:
         print(f"\n❌ Demo failed with error: {e}")
         import traceback
+
         traceback.print_exc()
 
 

examples/test_examples.py

@@ -65,9 +65,9 @@ def test_examples_directory_structure() -> None:
     """Verify examples directory has expected structure."""
     assert EXAMPLES_DIR.exists(), "Examples directory not found"
     assert (EXAMPLES_DIR / "async_demo.py").exists(), "async_demo.py not found"
-    assert (
-        EXAMPLES_DIR / "hybrid_async_demo.py"
-    ).exists(), "hybrid_async_demo.py not found"
+    assert (EXAMPLES_DIR / "hybrid_async_demo.py").exists(), (
+        "hybrid_async_demo.py not found"
+    )
 
 
 def test_example_has_docstring() -> None:
@@ -80,9 +80,7 @@ def test_example_has_docstring() -> None:
         assert '"""' in content, f"{script} missing docstring"
 
         # Check for shebang (makes it executable)
-        assert content.startswith("#!/usr/bin/env python"), (
-            f"{script} missing shebang"
-        )
+        assert content.startswith("#!/usr/bin/env python"), f"{script} missing shebang"
 
 
 def test_example_is_self_contained() -> None:

src/libtmux/common_async.py

@@ -65,12 +65,13 @@
 import shutil
 import sys
 import typing as t
+from collections.abc import Awaitable, Generator
 
 from . import exc
 from ._compat import LooseVersion
 
 if t.TYPE_CHECKING:
-    from collections.abc import Callable, Coroutine
+    from collections.abc import Callable
 
 logger = logging.getLogger(__name__)
 
@@ -88,11 +89,11 @@
 
 
 class AsyncEnvironmentMixin:
-    """Async mixin for manager session and server level environment variables in tmux."""
+    """Async mixin for managing session and server-level environment variables."""
 
     _add_option = None
 
-    acmd: Callable[[t.Any, t.Any], Coroutine[t.Any, t.Any, tmux_cmd_async]]
+    acmd: Callable[[t.Any, t.Any], Awaitable[tmux_cmd_async]]
 
     def __init__(self, add_option: str | None = None) -> None:
         self._add_option = add_option
@@ -179,7 +180,8 @@ async def show_environment(self) -> dict[str, bool | str]:
 
         .. versionchanged:: 0.13
 
-           Removed per-item lookups. Use :meth:`libtmux.common_async.AsyncEnvironmentMixin.getenv`.
+           Removed per-item lookups.
+           Use :meth:`libtmux.common_async.AsyncEnvironmentMixin.getenv`.
 
         Returns
         -------
@@ -242,7 +244,7 @@ async def getenv(self, name: str) -> str | bool | None:
         return opts_dict.get(name)
 
 
-class tmux_cmd_async:
+class tmux_cmd_async(Awaitable["tmux_cmd_async"]):
     """Run any :term:`tmux(1)` command through :py:mod:`asyncio.subprocess`.
 
     This is the async-first implementation. The tmux_cmd class is auto-generated
@@ -254,7 +256,9 @@ class tmux_cmd_async:
 
     >>> async def basic_example():
     ...     # Execute command with isolated socket
-    ...     proc = await tmux_cmd_async('-L', server.socket_name, 'new-session', '-d', '-P', '-F#S')
+    ...     proc = await tmux_cmd_async(
+    ...         '-L', server.socket_name, 'new-session', '-d', '-P', '-F#S'
+    ...     )
     ...     # Verify command executed successfully
     ...     return len(proc.stdout) > 0 and not proc.stderr
     >>> asyncio.run(basic_example())
@@ -279,7 +283,9 @@ class tmux_cmd_async:
     >>> async def check_session():
     ...     # Non-existent session returns non-zero returncode
     ...     sock = server.socket_name
-    ...     result = await tmux_cmd_async('-L', sock, 'has-session', '-t', 'nonexistent_12345')
+    ...     result = await tmux_cmd_async(
+    ...         '-L', sock, 'has-session', '-t', 'nonexistent_12345'
+    ...     )
     ...     return result.returncode != 0
     >>> asyncio.run(check_session())
     True
@@ -341,10 +347,10 @@ def __init__(
         self.returncode = returncode
         self._executed = False
 
-    async def execute(self) -> None:
+    async def execute(self) -> tmux_cmd_async:
         """Execute the tmux command asynchronously."""
         if self._executed:
-            return
+            return self
 
         try:
             process = await asyncio.create_subprocess_exec(
@@ -361,6 +367,15 @@ async def execute(self) -> None:
             raise
 
         self._executed = True
+        return self
+
+    async def _run(self) -> tmux_cmd_async:
+        await self.execute()
+        return self
+
+    def __await__(self) -> Generator[t.Any, None, tmux_cmd_async]:
+        """Allow ``await tmux_cmd_async(...)`` to execute the command."""
+        return self._run().__await__()
 
     @property
     def stdout(self) -> list[str]:
@@ -387,12 +402,9 @@ def stderr(self) -> list[str]:
         stderr_split = self._stderr.split("\n")
         return list(filter(None, stderr_split))  # filter empty values
 
-    async def __new__(cls, *args: t.Any, **kwargs: t.Any) -> tmux_cmd_async:
-        """Create and execute tmux command asynchronously."""
-        instance = object.__new__(cls)
-        instance.__init__(*args, **kwargs)
-        await instance.execute()
-        return instance
+    def __new__(cls, *args: t.Any, **kwargs: t.Any) -> tmux_cmd_async:
+        """Create tmux command instance (execution happens when awaited)."""
+        return super().__new__(cls)
 
 
 async def get_version() -> LooseVersion:

tests/asyncio/test_acmd.py

@@ -142,8 +142,7 @@ async def test_concurrent_acmd_operations(async_server: Server) -> None:
 async def test_acmd_error_handling(async_server: Server) -> None:
     """Test .acmd() properly handles errors."""
     # Create a session first to ensure server socket exists
-    result = await async_server.acmd("new-session", "-d", "-P", "-F#{session_id}")
-    session_id = result.stdout[0]
+    await async_server.acmd("new-session", "-d", "-P", "-F#{session_id}")
 
     # Invalid command (server socket now exists)
     result = await async_server.acmd("invalid-command-12345")