libtmux/src/libtmux/common.py at v0.55.0 · tmux-python/libtmux

535 lines (418 loc) · 14.7 KB
"""Helper methods and mixins for libtmux.
libtmux.common
~~~~~~~~~~~~~~
from __future__ import annotations
import logging
import shlex
import shutil
import subprocess
import typing as t
from . import exc
from ._compat import LooseVersion
if t.TYPE_CHECKING:
    from collections.abc import Callable
logger = logging.getLogger(__name__)
#: Minimum version of tmux required to run libtmux
TMUX_MIN_VERSION = "3.2a"
#: Most recent version of tmux supported
TMUX_MAX_VERSION = "3.6"
SessionDict = dict[str, t.Any]
WindowDict = dict[str, t.Any]
WindowOptionDict = dict[str, t.Any]
PaneDict = dict[str, t.Any]
class CmdProtocol(t.Protocol):
    """Command protocol for tmux command."""
    def __call__(self, cmd: str, *args: t.Any, **kwargs: t.Any) -> tmux_cmd:
        """Wrap tmux_cmd."""
        ...
class CmdMixin:
    """Command mixin for tmux command."""
    cmd: CmdProtocol
class EnvironmentMixin:
    """Mixin for manager session and server level environment variables in tmux."""
    _add_option = None
    cmd: Callable[[t.Any, t.Any], tmux_cmd]
    def __init__(self, add_option: str | None = None) -> None:
        self._add_option = add_option
    def set_environment(self, name: str, value: str) -> None:
        """Set environment ``$ tmux set-environment <name> <value>``.
        Parameters
        ----------
        name : str
            The environment variable name, e.g. 'PATH'.
        value : str
            Environment value.
        Raises
        ------
        ValueError
            If tmux returns an error.
        """
        args = ["set-environment"]
        if self._add_option:
            args += [self._add_option]
        args += [name, value]
        cmd = self.cmd(*args)
        if cmd.stderr:
                cmd.stderr[0]
                if isinstance(cmd.stderr, list) and len(cmd.stderr) == 1
                else cmd.stderr
            msg = f"tmux set-environment stderr: {cmd.stderr}"
            raise ValueError(msg)
    def unset_environment(self, name: str) -> None:
        """Unset environment variable ``$ tmux set-environment -u <name>``.
        Parameters
        ----------
        name : str
            The environment variable name, e.g. 'PATH'.
        Raises
        ------
        ValueError
            If tmux returns an error.
        """
        args = ["set-environment"]
        if self._add_option:
            args += [self._add_option]
        args += ["-u", name]
        cmd = self.cmd(*args)
        if cmd.stderr:
                cmd.stderr[0]
                if isinstance(cmd.stderr, list) and len(cmd.stderr) == 1
                else cmd.stderr
            msg = f"tmux set-environment stderr: {cmd.stderr}"
            raise ValueError(msg)
    def remove_environment(self, name: str) -> None:
        """Remove environment variable ``$ tmux set-environment -r <name>``.
        Parameters
        ----------
        name : str
            The environment variable name, e.g. 'PATH'.
        Raises
        ------
        ValueError
            If tmux returns an error.
        """
        args = ["set-environment"]
        if self._add_option:
            args += [self._add_option]
        args += ["-r", name]
        cmd = self.cmd(*args)
        if cmd.stderr:
                cmd.stderr[0]
                if isinstance(cmd.stderr, list) and len(cmd.stderr) == 1
                else cmd.stderr
            msg = f"tmux set-environment stderr: {cmd.stderr}"
            raise ValueError(msg)
    def show_environment(self) -> dict[str, bool | str]:
        """Show environment ``$ tmux show-environment -t [session]``.
        Return dict of environment variables for the session.
        .. versionchanged:: 0.13
           Removed per-item lookups. Use :meth:`libtmux.common.EnvironmentMixin.getenv`.
        Returns
        -------
        dict
            environmental variables in dict, if no name, or str if name
            entered.
        """
        tmux_args = ["show-environment"]
        if self._add_option:
            tmux_args += [self._add_option]
        cmd = self.cmd(*tmux_args)
        output = cmd.stdout
        opts = [tuple(item.split("=", 1)) for item in output]
        opts_dict: dict[str, str | bool] = {}
        for _t in opts:
            if len(_t) == 2:
                opts_dict[_t[0]] = _t[1]
            elif len(_t) == 1:
                opts_dict[_t[0]] = True
            else:
                raise exc.VariableUnpackingError(variable=_t)
        return opts_dict
    def getenv(self, name: str) -> str | bool | None:
        """Show environment variable ``$ tmux show-environment -t [session] <name>``.
        Return the value of a specific variable if the name is specified.
        .. versionadded:: 0.13
        Parameters
        ----------
        name : str
            the environment variable name. such as 'PATH'.
        Returns
        -------
        str
            Value of environment variable
        """
        tmux_args: tuple[str | int, ...] = ()
        tmux_args += ("show-environment",)
        if self._add_option:
            tmux_args += (self._add_option,)
        tmux_args += (name,)
        cmd = self.cmd(*tmux_args)
        output = cmd.stdout
        opts = [tuple(item.split("=", 1)) for item in output]
        opts_dict: dict[str, str | bool] = {}
        for _t in opts:
            if len(_t) == 2:
                opts_dict[_t[0]] = _t[1]
            elif len(_t) == 1:
                opts_dict[_t[0]] = True
            else:
                raise exc.VariableUnpackingError(variable=_t)
        return opts_dict.get(name)
class tmux_cmd:
    """Run any :term:`tmux(1)` command through :py:mod:`subprocess`.
    Examples
    --------
    Create a new session, check for error:
    >>> proc = tmux_cmd(f'-L{server.socket_name}', 'new-session', '-d', '-P', '-F#S')
    >>> if proc.stderr:
    ...     raise exc.LibTmuxException(
    ...         'Command: %s returned error: %s' % (proc.cmd, proc.stderr)
    ...     )
    >>> print(f'tmux command returned {" ".join(proc.stdout)}')
    tmux command returned 2
    Equivalent to:
    .. code-block:: console
        $ tmux new-session -s my session
    .. versionchanged:: 0.8
        Renamed from ``tmux`` to ``tmux_cmd``.
    def __init__(self, *args: t.Any, tmux_bin: str | None = None) -> None:
        resolved = tmux_bin or shutil.which("tmux")
        if not resolved:
            raise exc.TmuxCommandNotFound
        cmd = [resolved]
        cmd += args  # add the command arguments to cmd
        cmd = [str(c) for c in cmd]
        self.cmd = cmd
        if logger.isEnabledFor(logging.DEBUG):
            cmd_str = shlex.join(cmd)
            logger.debug(
                "tmux command dispatched",
                extra={"tmux_cmd": cmd_str},
        try:
            self.process = subprocess.Popen(
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                errors="backslashreplace",
            stdout, stderr = self.process.communicate()
            returncode = self.process.returncode
        except FileNotFoundError:
            raise exc.TmuxCommandNotFound from None
        except Exception:
            logger.error(  # noqa: TRY400
                "tmux subprocess failed",
                extra={
                    "tmux_cmd": shlex.join(cmd),
            raise
        self.returncode = returncode
        stdout_split = stdout.split("\n")
        # remove trailing newlines from stdout
        while stdout_split and stdout_split[-1] == "":
            stdout_split.pop()
        stderr_split = stderr.split("\n")
        self.stderr = list(filter(None, stderr_split))  # filter empty values
        if "has-session" in cmd and len(self.stderr) and not stdout_split:
            self.stdout = [self.stderr[0]]
        else:
            self.stdout = stdout_split
        if logger.isEnabledFor(logging.DEBUG):
            logger.debug(
                "tmux command completed",
                extra={
                    "tmux_cmd": shlex.join(cmd),
                    "tmux_exit_code": self.returncode,
                    "tmux_stdout": self.stdout[:100],
                    "tmux_stderr": self.stderr[:100],
                    "tmux_stdout_len": len(self.stdout),
                    "tmux_stderr_len": len(self.stderr),
def get_version(tmux_bin: str | None = None) -> LooseVersion:
    """Return tmux version.
    If tmux is built from git master, the version returned will be the latest
    version appended with -master, e.g. ``2.4-master``.
    If using OpenBSD's base system tmux, the version will have ``-openbsd``
    appended to the latest version, e.g. ``2.4-openbsd``.
    Parameters
    ----------
    tmux_bin : str, optional
        Path to tmux binary. If *None*, uses the system tmux from
        :func:`shutil.which`.
    Returns
    -------
    :class:`distutils.version.LooseVersion`
        tmux version according to *tmux_bin* if provided, otherwise the
        system tmux from :func:`shutil.which`
    proc = tmux_cmd("-V", tmux_bin=tmux_bin)
    if proc.stderr:
        if proc.stderr[0] == "tmux: unknown option -- V":
            if sys.platform.startswith("openbsd"):  # openbsd has no tmux -V
                return LooseVersion(f"{TMUX_MAX_VERSION}-openbsd")
            msg = (
                f"libtmux supports tmux {TMUX_MIN_VERSION} and greater. This system"
                " does not meet the minimum tmux version requirement."
            raise exc.LibTmuxException(
        raise exc.VersionTooLow(proc.stderr)
    version = proc.stdout[0].split("tmux ")[1]
    # Allow latest tmux HEAD
    if version == "master":
        return LooseVersion(f"{TMUX_MAX_VERSION}-master")
    version = re.sub(r"[a-z-]", "", version)
    return LooseVersion(version)
def has_version(version: str, tmux_bin: str | None = None) -> bool:
    """Return True if tmux version installed.
    Parameters
    ----------
    version : str
        version number, e.g. '3.2a'
    tmux_bin : str, optional
        Path to tmux binary. If *None*, uses the system tmux.
    Returns
    -------
        True if version matches
    return get_version(tmux_bin=tmux_bin) == LooseVersion(version)
def has_gt_version(min_version: str, tmux_bin: str | None = None) -> bool:
    """Return True if tmux version greater than minimum.
    Parameters
    ----------
    min_version : str
        tmux version, e.g. '3.2a'
    tmux_bin : str, optional
        Path to tmux binary. If *None*, uses the system tmux.
    Returns
    -------
        True if version above min_version
    return get_version(tmux_bin=tmux_bin) > LooseVersion(min_version)
def has_gte_version(min_version: str, tmux_bin: str | None = None) -> bool:
    """Return True if tmux version greater or equal to minimum.
    Parameters
    ----------
    min_version : str
        tmux version, e.g. '3.2a'
    tmux_bin : str, optional
        Path to tmux binary. If *None*, uses the system tmux.
    Returns
    -------
        True if version above or equal to min_version
    return get_version(tmux_bin=tmux_bin) >= LooseVersion(min_version)
def has_lte_version(max_version: str, tmux_bin: str | None = None) -> bool:
    """Return True if tmux version less or equal to minimum.
    Parameters
    ----------
    max_version : str
        tmux version, e.g. '3.2a'
    tmux_bin : str, optional
        Path to tmux binary. If *None*, uses the system tmux.
    Returns
    -------
         True if version below or equal to max_version
    return get_version(tmux_bin=tmux_bin) <= LooseVersion(max_version)
def has_lt_version(max_version: str, tmux_bin: str | None = None) -> bool:
    """Return True if tmux version less than minimum.
    Parameters
    ----------
    max_version : str
        tmux version, e.g. '3.2a'
    tmux_bin : str, optional
        Path to tmux binary. If *None*, uses the system tmux.
    Returns
    -------
        True if version below max_version
    return get_version(tmux_bin=tmux_bin) < LooseVersion(max_version)
def has_minimum_version(raises: bool = True, tmux_bin: str | None = None) -> bool:
    """Return True if tmux meets version requirement. Version >= 3.2a.
    Parameters
    ----------
    raises : bool
        raise exception if below minimum version requirement
    tmux_bin : str, optional
        Path to tmux binary. If *None*, uses the system tmux.
    Returns
    -------
        True if tmux meets minimum required version.
    libtmux.exc.VersionTooLow
        tmux version below minimum required for libtmux
    .. versionchanged:: 0.49.0
        Minimum version bumped to 3.2a. For older tmux, use libtmux v0.48.x.
    .. versionchanged:: 0.7.0
        No longer returns version, returns True or False
    .. versionchanged:: 0.1.7
        Versions will now remove trailing letters per
        `Issue 55 <https://github.com/tmux-python/tmuxp/issues/55>`_.
    current_version = get_version(tmux_bin=tmux_bin)
    if current_version < LooseVersion(TMUX_MIN_VERSION):
        if raises:
            msg = (
                f"libtmux only supports tmux {TMUX_MIN_VERSION} and greater. This "
                f"system has {current_version} installed. Upgrade your "
                "tmux to use libtmux, or use libtmux v0.48.x for older tmux versions."
            raise exc.VersionTooLow(msg)
        return False
    return True
def session_check_name(session_name: str | None) -> None:
    """Raise exception session name invalid, modeled after tmux function.
    tmux(1) session names may not be empty, or include periods or colons.
    These delimiters are reserved for noting session, window and pane.
    Parameters
    ----------
    session_name : str
        Name of session.
    :exc:`exc.BadSessionName`
        Invalid session name.
    if session_name is None or len(session_name) == 0:
        raise exc.BadSessionName(reason="empty", session_name=session_name)
    if "." in session_name:
        raise exc.BadSessionName(reason="contains periods", session_name=session_name)
    if ":" in session_name:
        raise exc.BadSessionName(reason="contains colons", session_name=session_name)
def get_libtmux_version() -> LooseVersion:
    """Return libtmux version is a PEP386 compliant format.
    Returns
    -------
    distutils.version.LooseVersion
        libtmux version
    from libtmux.__about__ import __version__
    return LooseVersion(__version__)
Provide feedback

Saved searches

Use saved searches to filter your results more quickly

FilesExpand file tree

common.py

Latest commit

History

common.py

File metadata and controls
