diff --git a/.tmuxp.yaml b/.tmuxp.yaml index 9baae31c7..c1a39a8a1 100644 --- a/.tmuxp.yaml +++ b/.tmuxp.yaml @@ -12,6 +12,7 @@ windows: panes: - focus: true - pane + - make watch_mypy - make watch_test - window_name: docs layout: main-horizontal diff --git a/CHANGES b/CHANGES index d240bd77c..e763cbb41 100644 --- a/CHANGES +++ b/CHANGES @@ -61,6 +61,14 @@ $ pip install --user --upgrade --pre libtmux window.show_window_option('DISPLAY') ``` +## What's new + +- **Improved typings** + + Now [`mypy --strict`] compliant ({issue}`383`) + + [`mypy --strict`]: https://mypy.readthedocs.io/en/stable/command_line.html#cmdoption-mypy-strict + ### Development - Fix incorrect function name `findWhere()` ({issue}`391`) diff --git a/Makefile b/Makefile index 9904fc1b8..c9609c7bc 100644 --- a/Makefile +++ b/Makefile @@ -56,3 +56,9 @@ watch_mypy: format_markdown: prettier --parser=markdown -w *.md docs/*.md docs/**/*.md CHANGES + +monkeytype_create: + poetry run monkeytype run `poetry run which py.test` + +monkeytype_apply: + poetry run monkeytype list-modules | xargs -n1 -I{} sh -c 'poetry run monkeytype apply {}' diff --git a/README.md b/README.md index 6d0a552d9..b30c501dc 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # libtmux -libtmux is a python scripting library for tmux. You can use it to command and control tmux servers, +libtmux is a [typed](https://docs.python.org/3/library/typing.html) python scripting library for tmux. You can use it to command and control tmux servers, sessions, windows, and panes. It is the tool powering [tmuxp], a tmux workspace manager. [![Python Package](https://img.shields.io/pypi/v/libtmux.svg)](https://pypi.org/project/libtmux/) diff --git a/docs/about.md b/docs/about.md index 5fbfe992f..d2aad5aaf 100644 --- a/docs/about.md +++ b/docs/about.md @@ -14,7 +14,7 @@ ``` -libtmux is an [abstraction layer] for tmux. +libtmux is a [typed](https://docs.python.org/3/library/typing.html) [abstraction layer] for tmux. It builds upon the concept of targets `-t`, to direct commands against individual session, windows and panes and `FORMATS`, template variables diff --git a/docs/conf.py b/docs/conf.py index 33d885f88..7264b730b 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -4,7 +4,7 @@ import sys from os.path import dirname, relpath from pathlib import Path -from typing import Dict, List +from typing import Dict, List, Union import libtmux # NOQA from libtmux import test # NOQA @@ -17,7 +17,7 @@ sys.path.insert(0, str(cwd / "_ext")) # package data -about: Dict = {} +about: Dict[str, str] = {} with open("../libtmux/__about__.py") as fp: exec(fp.read(), about) @@ -69,8 +69,8 @@ html_css_files = ["css/custom.css"] html_extra_path = ["manifest.json"] html_theme = "furo" -html_theme_path: List = [] -html_theme_options: Dict = { +html_theme_path: List[str] = [] +html_theme_options: Dict[str, Union[str, List[Dict[str, str]]]] = { "light_logo": "img/libtmux.svg", "dark_logo": "img/libtmux.svg", "footer_icons": [ @@ -162,7 +162,9 @@ intersphinx_mapping = {"http://docs.python.org/": None} -def linkcode_resolve(domain, info): # NOQA: C901 +def linkcode_resolve( + domain: str, info: Dict[str, str] +) -> Union[None, str]: # NOQA: C901 """ Determine the URL corresponding to Python object @@ -195,7 +197,8 @@ def linkcode_resolve(domain, info): # NOQA: C901 except AttributeError: pass else: - obj = unwrap(obj) + if callable(obj): + obj = unwrap(obj) try: fn = inspect.getsourcefile(obj) diff --git a/libtmux/__about__.py b/libtmux/__about__.py index e9d2e3f65..93e15a634 100644 --- a/libtmux/__about__.py +++ b/libtmux/__about__.py @@ -1,7 +1,7 @@ __title__ = "libtmux" __package_name__ = "libtmux" __version__ = "0.12.0" -__description__ = "Scripting library / ORM / API wrapper for tmux" +__description__ = "Typed scripting library / ORM / API wrapper for tmux" __email__ = "tony@git-pull.com" __author__ = "Tony Narlock" __github__ = "https://github.com/tmux-python/libtmux" diff --git a/libtmux/_compat.py b/libtmux/_compat.py index 82ec43567..d7b6bb274 100644 --- a/libtmux/_compat.py +++ b/libtmux/_compat.py @@ -1,5 +1,6 @@ # flake8: NOQA import sys +import types import typing as t console_encoding = sys.__stdout__.encoding @@ -14,7 +15,12 @@ def console_to_str(s: bytes) -> str: # TODO Consider removing, reraise does not seem to be called anywhere -def reraise(tp, value, tb=None): +def reraise( + tp: t.Type[BaseException], + value: BaseException, + tb: types.TracebackType, +) -> t.NoReturn: + if value.__traceback__ is not tb: raise (value.with_traceback(tb)) raise value diff --git a/libtmux/common.py b/libtmux/common.py index 940f6eb06..333edc748 100644 --- a/libtmux/common.py +++ b/libtmux/common.py @@ -13,10 +13,29 @@ import typing as t from collections.abc import MutableMapping from distutils.version import LooseVersion +from typing import ( + Any, + Dict, + Generic, + KeysView, + List, + Optional, + TypeVar, + Union, + overload, +) from . import exc from ._compat import console_to_str, str_from_console +if t.TYPE_CHECKING: + from typing_extensions import Literal + + from libtmux.pane import Pane + from libtmux.session import Session + from libtmux.window import Window + + logger = logging.getLogger(__name__) @@ -28,6 +47,7 @@ SessionDict = t.Dict[str, t.Any] WindowDict = t.Dict[str, t.Any] +WindowOptionDict = t.Dict[str, t.Any] PaneDict = t.Dict[str, t.Any] @@ -40,10 +60,12 @@ class EnvironmentMixin: _add_option = None - def __init__(self, add_option=None): + cmd: t.Callable[[t.Any, t.Any], "tmux_cmd"] + + def __init__(self, add_option: Optional[str] = None) -> None: self._add_option = add_option - def set_environment(self, name, value): + def set_environment(self, name: str, value: str) -> None: """ Set environment ``$ tmux set-environment ``. @@ -60,14 +82,17 @@ def set_environment(self, name, value): args += [name, value] - proc = self.cmd(*args) + cmd = self.cmd(*args) - if proc.stderr: - if isinstance(proc.stderr, list) and len(proc.stderr) == 1: - proc.stderr = proc.stderr[0] - raise ValueError("tmux set-environment stderr: %s" % proc.stderr) + if cmd.stderr: + stderr = ( + cmd.stderr[0] + if isinstance(cmd.stderr, list) and len(cmd.stderr) == 1 + else cmd.stderr + ) + raise ValueError("tmux set-environment stderr: %s" % cmd.stderr) - def unset_environment(self, name): + def unset_environment(self, name: str) -> None: """ Unset environment variable ``$ tmux set-environment -u ``. @@ -81,14 +106,17 @@ def unset_environment(self, name): args += [self._add_option] args += ["-u", name] - proc = self.cmd(*args) + cmd = self.cmd(*args) - if proc.stderr: - if isinstance(proc.stderr, list) and len(proc.stderr) == 1: - proc.stderr = proc.stderr[0] - raise ValueError("tmux set-environment stderr: %s" % proc.stderr) + if cmd.stderr: + stderr = ( + cmd.stderr[0] + if isinstance(cmd.stderr, list) and len(cmd.stderr) == 1 + else cmd.stderr + ) + raise ValueError("tmux set-environment stderr: %s" % cmd.stderr) - def remove_environment(self, name): + def remove_environment(self, name: str) -> None: """Remove environment variable ``$ tmux set-environment -r ``. Parameters @@ -101,14 +129,17 @@ def remove_environment(self, name): args += [self._add_option] args += ["-r", name] - proc = self.cmd(*args) + cmd = self.cmd(*args) - if proc.stderr: - if isinstance(proc.stderr, list) and len(proc.stderr) == 1: - proc.stderr = proc.stderr[0] - raise ValueError("tmux set-environment stderr: %s" % proc.stderr) + if cmd.stderr: + stderr = ( + cmd.stderr[0] + if isinstance(cmd.stderr, list) and len(cmd.stderr) == 1 + else cmd.stderr + ) + raise ValueError("tmux set-environment stderr: %s" % cmd.stderr) - def show_environment(self): + def show_environment(self) -> Dict[str, Union[bool, str]]: """Show environment ``$ tmux show-environment -t [session]``. Return dict of environment variables for the session. @@ -126,9 +157,10 @@ def show_environment(self): tmux_args = ["show-environment"] if self._add_option: tmux_args += [self._add_option] - vars = self.cmd(*tmux_args).stdout - vars = [tuple(item.split("=", 1)) for item in vars] - vars_dict = {} + cmd = self.cmd(*tmux_args) + output = cmd.stdout + vars = [tuple(item.split("=", 1)) for item in output] + vars_dict: t.Dict[str, t.Union[str, bool]] = {} for t in vars: if len(t) == 2: vars_dict[t[0]] = t[1] @@ -139,7 +171,7 @@ def show_environment(self): return vars_dict - def getenv(self, name): + def getenv(self, name: str) -> Optional[t.Union[str, bool]]: """Show environment variable ``$ tmux show-environment -t [session] ``. Return the value of a specific variable if the name is specified. @@ -156,13 +188,16 @@ def getenv(self, name): str Value of environment variable """ - tmux_args = ["show-environment"] + tmux_args: t.Tuple[t.Union[str, int], ...] = tuple() + + tmux_args += ("show-environment",) if self._add_option: - tmux_args += [self._add_option] - tmux_args += [name] - vars = self.cmd(*tmux_args).stdout - vars = [tuple(item.split("=", 1)) for item in vars] - vars_dict = {} + tmux_args += (self._add_option,) + tmux_args += (name,) + cmd = self.cmd(*tmux_args) + output = cmd.stdout + vars = [tuple(item.split("=", 1)) for item in output] + vars_dict: t.Dict[str, t.Union[str, bool]] = {} for t in vars: if len(t) == 2: vars_dict[t[0]] = t[1] @@ -214,7 +249,7 @@ class tmux_cmd: Renamed from ``tmux`` to ``tmux_cmd``. """ - def __init__(self, *args, **kwargs): + def __init__(self, *args: t.Any, **kwargs: t.Any) -> None: tmux_bin = which( "tmux", default_paths=kwargs.get( @@ -240,26 +275,28 @@ def __init__(self, *args, **kwargs): returncode = self.process.returncode except Exception as e: logger.error(f"Exception for {subprocess.list2cmdline(cmd)}: \n{e}") + raise self.returncode = returncode - self.stdout = console_to_str(stdout) - self.stdout = self.stdout.split("\n") - self.stdout = list(filter(None, self.stdout)) # filter empty values + stdout_str = console_to_str(stdout) + stdout_split = stdout_str.split("\n") + stdout_filtered = list(filter(None, stdout_split)) # filter empty values - self.stderr = console_to_str(stderr) - self.stderr = self.stderr.split("\n") - self.stderr = list(filter(None, self.stderr)) # filter empty values + stderr_str = console_to_str(stderr) + stderr_split = stderr_str.split("\n") + self.stderr = list(filter(None, stderr_split)) # filter empty values - if "has-session" in cmd and len(self.stderr): - if not self.stdout: - self.stdout = self.stderr[0] + if "has-session" in cmd and len(self.stderr) and not stdout_filtered: + self.stdout = [self.stderr[0]] + else: + self.stdout = stdout_filtered logger.debug("self.stdout for {}: \n{}".format(" ".join(cmd), self.stdout)) -class TmuxMappingObject(MutableMapping): - +# class TmuxMappingObject(t.Mapping[str, t.Union[str,int,bool]]): +class TmuxMappingObject(t.Mapping[t.Any, t.Any]): r"""Base: :py:class:`MutableMapping`. Convenience container. Base class for :class:`Pane`, :class:`Window`, @@ -278,37 +315,48 @@ class TmuxMappingObject(MutableMapping): :class:`Pane` :attr:`Pane.formatter_prefix` pane\_ ================ ================================== ============== """ + _info: t.Dict[t.Any, t.Any] + formatter_prefix: str - def __getitem__(self, key): - return self._info[key] + def __getitem__(self, key: str) -> str: + item = self._info[key] + assert item is not None + assert isinstance(item, str) + return item - def __setitem__(self, key, value): + def __setitem__(self, key: str, value: str) -> None: self._info[key] = value self.dirty = True - def __delitem__(self, key): + def __delitem__(self, key: str) -> None: del self._info[key] self.dirty = True - def keys(self): + def keys(self) -> KeysView[str]: """Return list of keys.""" return self._info.keys() - def __iter__(self): + def __iter__(self) -> t.Iterator[str]: return self._info.__iter__() - def __len__(self): + def __len__(self) -> int: return len(self._info.keys()) - def __getattr__(self, key): + def __getattr__(self, key: str) -> str: try: - return self._info[self.formatter_prefix + key] + val = self._info[self.formatter_prefix + key] + assert val is not None + assert isinstance(val, str) + return val except KeyError: raise AttributeError(f"{self.__class__} has no property {key}") -class TmuxRelationalObject: +O = TypeVar("O", "Pane", "Window", "Session") +D = TypeVar("D", "PaneDict", "WindowDict", "SessionDict") + +class TmuxRelationalObject(Generic[O, D]): """Base Class for managing tmux object child entities. .. # NOQA Manages collection of child objects (a :class:`Server` has a collection of @@ -337,7 +385,12 @@ class TmuxRelationalObject: ================ ================================== ============== """ - def find_where(self, attrs): + children: t.List[O] + child_id_attribute: str + + def find_where( + self, attrs: Dict[str, str] + ) -> Optional[Union["Pane", "Window", "Session"]]: """Return object on first match. .. versionchanged:: 0.4 @@ -349,7 +402,20 @@ def find_where(self, attrs): except IndexError: return None - def where(self, attrs, first=False): + @overload + def where(self, attrs: Dict[str, str], first: "Literal[True]") -> O: + ... + + @overload + def where(self, attrs: Dict[str, str], first: "Literal[False]") -> t.List[O]: + ... + + @overload + def where(self, attrs: Dict[str, str]) -> t.List[O]: + ... + + def where(self, attrs: Dict[str, str], first: bool = False) -> t.Union[List[O], O]: + # ) -> List[Union["Session", "Pane", "Window", t.Any]]: """ Return objects matching child objects properties. @@ -364,7 +430,7 @@ def where(self, attrs, first=False): """ # from https://github.com/serkanyersen/underscore.py - def by(val) -> bool: + def by(val: O) -> bool: for key in attrs.keys(): try: if attrs[key] != val[key]: @@ -373,13 +439,13 @@ def by(val) -> bool: return False return True - # TODO add type hint - target_children = list(filter(by, self.children)) + target_children: t.List[O] = [s for s in self.children if by(s)] + if first: return target_children[0] return target_children - def get_by_id(self, id): + def get_by_id(self, id: str) -> Optional[Union["Pane", "Window", "Session"]]: """ Return object based on ``child_id_attribute``. @@ -638,7 +704,7 @@ def has_minimum_version(raises: bool = True) -> bool: return True -def session_check_name(session_name: str): +def session_check_name(session_name: t.Optional[str]) -> None: """ Raises exception session name invalid, modeled after tmux function. @@ -655,7 +721,7 @@ def session_check_name(session_name: str): :exc:`exc.BadSessionName` Invalid session name. """ - if not session_name or len(session_name) == 0: + if session_name is None or len(session_name) == 0: raise exc.BadSessionName("tmux session names may not be empty.") elif "." in session_name: raise exc.BadSessionName( @@ -667,7 +733,7 @@ def session_check_name(session_name: str): ) -def handle_option_error(error: str): +def handle_option_error(error: str) -> t.Type[exc.OptionError]: """Raises exception if error in option command found. In tmux 3.0, show-option and show-window-otion return invalid option instead of diff --git a/libtmux/pane.py b/libtmux/pane.py index d944105fe..74b4ec4ab 100644 --- a/libtmux/pane.py +++ b/libtmux/pane.py @@ -7,11 +7,16 @@ """ import logging import typing as t +from typing import Dict, overload + +from libtmux.common import tmux_cmd from . import exc -from .common import TmuxMappingObject, TmuxRelationalObject +from .common import PaneDict, TmuxMappingObject, TmuxRelationalObject if t.TYPE_CHECKING: + from typing_extensions import Literal + from .server import Server from .session import Session from .window import Window @@ -20,7 +25,7 @@ logger = logging.getLogger(__name__) -class Pane(TmuxMappingObject, TmuxRelationalObject): +class Pane(TmuxMappingObject): """ A :term:`tmux(1)` :term:`Pane` [pane_manual]_. @@ -58,22 +63,27 @@ class Pane(TmuxMappingObject, TmuxRelationalObject): server: "Server" """:class:`libtmux.Server` pane is linked to""" - def __init__(self, window: "Window", **kwargs): + def __init__( + self, + window: "Window", + # pane_id: t.Optional[t.Union[str, int]] = None, + pane_id: t.Union[str, int], + **kwargs: t.Any, + ) -> None: self.window = window self.session = self.window.session self.server = self.session.server - self._pane_id = kwargs["pane_id"] + self._pane_id = pane_id self.server._update_panes() @property - def _info(self): - + def _info(self) -> PaneDict: # type: ignore # mypy#1362 attrs = {"pane_id": self._pane_id} # from https://github.com/serkanyersen/underscore.py - def by(val) -> bool: + def by(val: PaneDict) -> bool: for key in attrs.keys(): try: if attrs[key] != val[key]: @@ -82,11 +92,11 @@ def by(val) -> bool: return False return True - # TODO add type hint - target_panes = list(filter(by, self.server._panes)) + target_panes = [s for s in self.server._panes if by(s)] + return target_panes[0] - def cmd(self, cmd, *args, **kwargs): + def cmd(self, cmd: str, *args: t.Any, **kwargs: t.Any) -> tmux_cmd: """Return :meth:`Server.cmd` defaulting to ``target_pane`` as target. Send command to tmux with :attr:`pane_id` as ``target-pane``. @@ -103,7 +113,13 @@ def cmd(self, cmd, *args, **kwargs): return self.server.cmd(cmd, *args, **kwargs) - def send_keys(self, cmd, enter=True, suppress_history=True, literal=False): + def send_keys( + self, + cmd: str, + enter: t.Optional[bool] = True, + suppress_history: t.Optional[bool] = True, + literal: t.Optional[bool] = False, + ) -> None: """ ``$ tmux send-keys`` to the pane. @@ -131,7 +147,19 @@ def send_keys(self, cmd, enter=True, suppress_history=True, literal=False): if enter: self.enter() - def display_message(self, cmd, get_text=False): + @overload + def display_message( + self, cmd: str, get_text: "Literal[True]" + ) -> t.Union[str, t.List[str]]: + ... + + @overload + def display_message(self, cmd: str, get_text: "Literal[False]") -> None: + ... + + def display_message( + self, cmd: str, get_text: bool = False + ) -> t.Optional[t.Union[str, t.List[str]]]: """ ``$ tmux display-message`` to the pane. @@ -152,20 +180,25 @@ def display_message(self, cmd, get_text=False): """ if get_text: return self.cmd("display-message", "-p", cmd).stdout - else: - self.cmd("display-message", cmd) - def clear(self): + self.cmd("display-message", cmd) + return None + + def clear(self) -> None: """Clear pane.""" self.send_keys("reset") - def reset(self): + def reset(self) -> None: """Reset and clear pane history.""" self.cmd("send-keys", r"-R \; clear-history") def split_window( - self, attach=False, vertical=True, start_directory=None, percent=None + self, + attach: bool = False, + vertical: bool = True, + start_directory: t.Optional[str] = None, + percent: t.Optional[int] = None, ) -> "Pane": """ Split window at pane and return newly created :class:`Pane`. @@ -193,7 +226,7 @@ def split_window( percent=percent, ) - def set_width(self, width): + def set_width(self, width: int) -> None: """ Set width of pane. @@ -204,7 +237,7 @@ def set_width(self, width): """ self.resize_pane(width=width) - def set_height(self, height): + def set_height(self, height: int) -> None: """ Set height of pane. @@ -215,7 +248,7 @@ def set_height(self, height): """ self.resize_pane(height=height) - def resize_pane(self, *args, **kwargs): + def resize_pane(self, *args: t.Any, **kwargs: t.Any) -> "Pane": """ ``$ tmux resize-pane`` of pane and return ``self``. @@ -239,7 +272,6 @@ def resize_pane(self, *args, **kwargs): ------ exc.LibTmuxException """ - if "height" in kwargs: proc = self.cmd("resize-pane", "-y%s" % int(kwargs["height"])) elif "width" in kwargs: @@ -253,7 +285,7 @@ def resize_pane(self, *args, **kwargs): self.server._update_panes() return self - def enter(self): + def enter(self) -> None: """ Send carriage return to pane. @@ -261,7 +293,7 @@ def enter(self): """ self.cmd("send-keys", "Enter") - def capture_pane(self): + def capture_pane(self) -> t.Union[str, t.List[str]]: """ Capture text from pane. diff --git a/libtmux/py.typed b/libtmux/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/libtmux/server.py b/libtmux/server.py index d31f39296..b31a3d061 100644 --- a/libtmux/server.py +++ b/libtmux/server.py @@ -8,6 +8,9 @@ import os import typing as t +from libtmux.common import tmux_cmd +from libtmux.session import Session + from . import exc, formats from .common import ( EnvironmentMixin, @@ -17,14 +20,12 @@ WindowDict, has_gte_version, session_check_name, - tmux_cmd, ) -from .session import Session logger = logging.getLogger(__name__) -class Server(TmuxRelationalObject, EnvironmentMixin): +class Server(TmuxRelationalObject["Session", "SessionDict"], EnvironmentMixin): """ The :term:`tmux(1)` :term:`Server` [server_manual]_. @@ -75,15 +76,15 @@ class Server(TmuxRelationalObject, EnvironmentMixin): def __init__( self, - socket_name=None, - socket_path=None, - config_file=None, - colors=None, - **kwargs, - ): + socket_name: t.Optional[str] = None, + socket_path: t.Optional[str] = None, + config_file: t.Optional[str] = None, + colors: t.Optional[int] = None, + **kwargs: t.Any, + ) -> None: EnvironmentMixin.__init__(self, "-g") - self._windows = [] - self._panes = [] + self._windows: t.List[WindowDict] = [] + self._panes: t.List[PaneDict] = [] if socket_name: self.socket_name = socket_name @@ -97,7 +98,7 @@ def __init__( if colors: self.colors = colors - def cmd(self, *args, **kwargs): + def cmd(self, *args: t.Any, **kwargs: t.Any) -> tmux_cmd: """ Execute tmux command and return output. @@ -112,22 +113,22 @@ def cmd(self, *args, **kwargs): Renamed from ``.tmux`` to ``.cmd``. """ - args = list(args) + cmd_args: t.List[t.Union[str, int]] = list(args) if self.socket_name: - args.insert(0, f"-L{self.socket_name}") + cmd_args.insert(0, f"-L{self.socket_name}") if self.socket_path: - args.insert(0, f"-S{self.socket_path}") + cmd_args.insert(0, f"-S{self.socket_path}") if self.config_file: - args.insert(0, f"-f{self.config_file}") + cmd_args.insert(0, f"-f{self.config_file}") if self.colors: if self.colors == 256: - args.insert(0, "-2") + cmd_args.insert(0, "-2") elif self.colors == 88: - args.insert(0, "-8") + cmd_args.insert(0, "-8") else: raise ValueError("Server.colors must equal 88 or 256") - return tmux_cmd(*args, **kwargs) + return tmux_cmd(*cmd_args, **kwargs) def _list_sessions(self) -> t.List[SessionDict]: """ @@ -148,25 +149,27 @@ def _list_sessions(self) -> t.List[SessionDict]: tmux_args = ("-F%s" % formats.FORMAT_SEPARATOR.join(tmux_formats),) # output - proc = self.cmd("list-sessions", *tmux_args) + list_sessions_cmd = self.cmd("list-sessions", *tmux_args) - if proc.stderr: - raise exc.LibTmuxException(proc.stderr) + if list_sessions_cmd.stderr: + raise exc.LibTmuxException(list_sessions_cmd.stderr) sformats = formats.SESSION_FORMATS tmux_formats = ["#{%s}" % format for format in sformats] - sessions = proc.stdout + sessions_output = list_sessions_cmd.stdout # combine format keys with values returned from ``tmux list-sessions`` - sessions = [ + sessions_formatters = [ dict(zip(sformats, session.split(formats.FORMAT_SEPARATOR))) - for session in sessions + for session in sessions_output ] # clear up empty dict - sessions = [{k: v for k, v in session.items() if v} for session in sessions] + sessions_formatters_filtered = [ + {k: v for k, v in session.items() if v} for session in sessions_formatters + ] - return sessions + return sessions_formatters_filtered @property def _sessions(self) -> t.List[SessionDict]: @@ -190,7 +193,7 @@ def sessions(self) -> t.List[Session]: return self.list_sessions() #: Alias :attr:`sessions` for :class:`~libtmux.common.TmuxRelationalObject` - children = sessions + children = sessions # type: ignore def _list_windows(self) -> t.List[WindowDict]: """ @@ -218,28 +221,30 @@ def _list_windows(self) -> t.List[WindowDict]: if proc.stderr: raise exc.LibTmuxException(proc.stderr) - windows = proc.stdout + window_output = proc.stdout wformats = ["session_name", "session_id"] + formats.WINDOW_FORMATS # combine format keys with values returned from ``tmux list-windows`` - windows = [ + window_formatters = [ dict(zip(wformats, window.split(formats.FORMAT_SEPARATOR))) - for window in windows + for window in window_output ] # clear up empty dict - windows = [{k: v for k, v in window.items() if v} for window in windows] + window_formatters_filtered = [ + {k: v for k, v in window.items() if v} for window in window_formatters + ] # tmux < 1.8 doesn't have window_id, use window_name - for w in windows: + for w in window_formatters_filtered: if "window_id" not in w: w["window_id"] = w["window_name"] if self._windows: self._windows[:] = [] - self._windows.extend(windows) + self._windows.extend(window_formatters_filtered) return self._windows @@ -282,7 +287,7 @@ def _list_panes(self) -> t.List[PaneDict]: if proc.stderr: raise exc.LibTmuxException(proc.stderr) - panes = proc.stdout + pane_output = proc.stdout pformats = [ "session_name", @@ -293,24 +298,24 @@ def _list_panes(self) -> t.List[PaneDict]: ] + formats.PANE_FORMATS # combine format keys with values returned from ``tmux list-panes`` - panes = [ - dict(zip(pformats, window.split(formats.FORMAT_SEPARATOR))) - for window in panes + pane_formatters = [ + dict(zip(pformats, formatter.split(formats.FORMAT_SEPARATOR))) + for formatter in pane_output ] - # clear up empty dict - panes = [ + # Filter empty values + pane_formatters_filtered = [ { - k: v for k, v in window.items() if v or k == "pane_current_path" + k: v for k, v in formatter.items() if v or k == "pane_current_path" } # preserve pane_current_path, in case it entered a new process # where we may not get a cwd from. - for window in panes + for formatter in pane_formatters ] if self._panes: self._panes[:] = [] - self._panes.extend(panes) + self._panes.extend(pane_formatters_filtered) return self._panes @@ -382,11 +387,11 @@ def has_session(self, target_session: str, exact: bool = True) -> bool: return False - def kill_server(self): + def kill_server(self) -> None: """``$ tmux kill-server``.""" self.cmd("kill-server") - def kill_session(self, target_session=None) -> "Server": + def kill_session(self, target_session: t.Union[str, int]) -> "Server": """ Kill the tmux session with ``$ tmux kill-session``, return ``self``. @@ -404,8 +409,6 @@ def kill_session(self, target_session=None) -> "Server": ------ :exc:`exc.BadSessionName` """ - session_check_name(target_session) - proc = self.cmd("kill-session", "-t%s" % target_session) if proc.stderr: @@ -413,7 +416,7 @@ def kill_session(self, target_session=None) -> "Server": return self - def switch_client(self, target_session): + def switch_client(self, target_session: str) -> None: """ ``$ tmux switch-client``. @@ -433,7 +436,7 @@ def switch_client(self, target_session): if proc.stderr: raise exc.LibTmuxException(proc.stderr) - def attach_session(self, target_session=None): + def attach_session(self, target_session: t.Optional[str] = None) -> None: """``$ tmux attach-session`` aka alias: ``$ tmux attach``. Parameters @@ -447,7 +450,7 @@ def attach_session(self, target_session=None): """ session_check_name(target_session) - tmux_args = tuple() + tmux_args: t.Tuple[str, ...] = tuple() if target_session: tmux_args += ("-t%s" % target_session,) @@ -458,14 +461,14 @@ def attach_session(self, target_session=None): def new_session( self, - session_name=None, - kill_session=False, - attach=False, - start_directory=None, - window_name=None, - window_command=None, - *args, - **kwargs, + session_name: t.Optional[str] = None, + kill_session: bool = False, + attach: bool = False, + start_directory: t.Optional[str] = None, + window_name: t.Optional[str] = None, + window_command: t.Optional[str] = None, + *args: t.Any, + **kwargs: t.Any, ) -> Session: """ Return :class:`Session` from ``$ tmux new-session``. @@ -516,6 +519,7 @@ def new_session( :exc:`exc.BadSessionName` """ session_check_name(session_name) + assert session_name is not None if self.has_session(session_name): if kill_session: @@ -534,7 +538,7 @@ def new_session( if env: del os.environ["TMUX"] - tmux_args: t.Tuple = ( + tmux_args: t.Tuple[t.Union[str, int], ...] = ( "-s%s" % session_name, "-P", "-F%s" % formats.FORMAT_SEPARATOR.join(tmux_formats), # output @@ -562,17 +566,19 @@ def new_session( if proc.stderr: raise exc.LibTmuxException(proc.stderr) - session = proc.stdout[0] + session_stdout = proc.stdout[0] if env: os.environ["TMUX"] = env - # combine format keys with values returned from ``tmux list-windows`` - session = dict(zip(sformats, session.split(formats.FORMAT_SEPARATOR))) + # Combine format keys with values returned from ``tmux list-windows`` + session_params = dict( + zip(sformats, session_stdout.split(formats.FORMAT_SEPARATOR)) + ) - # clear up empty dict - session = {k: v for k, v in session.items() if v} + # Filter empty values + session_params = {k: v for k, v in session_params.items() if v} - session = Session(server=self, **session) + session = Session(server=self, **session_params) return session diff --git a/libtmux/session.py b/libtmux/session.py index 78d2b25e3..1133c9934 100644 --- a/libtmux/session.py +++ b/libtmux/session.py @@ -8,9 +8,13 @@ import os import typing as t +from libtmux.common import tmux_cmd +from libtmux.window import Window + from . import exc, formats from .common import ( EnvironmentMixin, + SessionDict, TmuxMappingObject, TmuxRelationalObject, WindowDict, @@ -18,7 +22,6 @@ has_version, session_check_name, ) -from .window import Window if t.TYPE_CHECKING: from .pane import Pane @@ -28,7 +31,9 @@ logger = logging.getLogger(__name__) -class Session(TmuxMappingObject, TmuxRelationalObject, EnvironmentMixin): +class Session( + TmuxMappingObject, TmuxRelationalObject["Window", "WindowDict"], EnvironmentMixin +): """ A :term:`tmux(1)` :term:`Session` [session_manual]_. @@ -58,21 +63,18 @@ class Session(TmuxMappingObject, TmuxRelationalObject, EnvironmentMixin): server: "Server" """:class:`libtmux.server.Server` session is linked to""" - def __init__(self, server: "Server", **kwargs): + def __init__(self, server: "Server", session_id: str, **kwargs: t.Any) -> None: EnvironmentMixin.__init__(self) self.server = server - if "session_id" not in kwargs: - raise ValueError("Session requires a `session_id`") - self._session_id = kwargs["session_id"] + self._session_id = session_id self.server._update_windows() @property - def _info(self): - + def _info(self) -> t.Optional[SessionDict]: # type: ignore # mypy#1362 attrs = {"session_id": str(self._session_id)} - def by(val) -> bool: + def by(val: SessionDict) -> bool: for key in attrs.keys(): try: if attrs[key] != val[key]: @@ -81,14 +83,14 @@ def by(val) -> bool: return False return True - # TODO add type hint - target_sessions = list(filter(by, self.server._sessions)) + target_sessions = [s for s in self.server._sessions if by(s)] try: return target_sessions[0] except IndexError as e: logger.error(e) + return None - def cmd(self, *args, **kwargs): + def cmd(self, *args: t.Any, **kwargs: t.Any) -> tmux_cmd: """ Return :meth:`server.cmd`. @@ -105,20 +107,24 @@ def cmd(self, *args, **kwargs): # if -t is not set in any arg yet if not any("-t" in str(x) for x in args): # insert -t immediately after 1st arg, as per tmux format - new_args = [args[0]] - new_args += ["-t", self.id] - new_args += args[1:] + new_args: t.Tuple[str, ...] = tuple() + new_args += (args[0],) + new_args += ( + "-t", + self.id, + ) + new_args += tuple(args[1:]) args = new_args return self.server.cmd(*args, **kwargs) - def attach_session(self): + def attach_session(self) -> None: """Return ``$ tmux attach-session`` aka alias: ``$ tmux attach``.""" proc = self.cmd("attach-session", "-t%s" % self.id) if proc.stderr: raise exc.LibTmuxException(proc.stderr) - def kill_session(self): + def kill_session(self) -> None: """``$ tmux kill-session``.""" proc = self.cmd("kill-session", "-t%s" % self.id) @@ -126,7 +132,7 @@ def kill_session(self): if proc.stderr: raise exc.LibTmuxException(proc.stderr) - def switch_client(self): + def switch_client(self) -> None: """ Switch client to this session. @@ -177,11 +183,11 @@ def rename_session(self, new_name: str) -> "Session": def new_window( self, - window_name=None, - start_directory=None, - attach=True, - window_index="", - window_shell=None, + window_name: t.Optional[str] = None, + start_directory: None = None, + attach: bool = True, + window_index: str = "", + window_shell: None = None, ) -> Window: """ Return :class:`Window` from ``$ tmux new-window``. @@ -215,7 +221,7 @@ def new_window( wformats = ["session_name", "session_id"] + formats.WINDOW_FORMATS tmux_formats = ["#{%s}" % f for f in wformats] - window_args: t.Tuple = tuple() + window_args: t.Tuple[str, ...] = tuple() if not attach: window_args += ("-d",) @@ -242,24 +248,26 @@ def new_window( if window_shell: window_args += (window_shell,) - proc = self.cmd("new-window", *window_args) + cmd = self.cmd("new-window", *window_args) - if proc.stderr: - raise exc.LibTmuxException(proc.stderr) + if cmd.stderr: + raise exc.LibTmuxException(cmd.stderr) - window = proc.stdout[0] + window_output = cmd.stdout[0] - window = dict(zip(wformats, window.split(formats.FORMAT_SEPARATOR))) + window_formatters = dict( + zip(wformats, window_output.split(formats.FORMAT_SEPARATOR)) + ) # clear up empty dict - window = {k: v for k, v in window.items() if v} - window = Window(session=self, **window) + window_formatters_filtered = {k: v for k, v in window_formatters.items() if v} + window = Window(session=self, **window_formatters_filtered) self.server._update_windows() return window - def kill_window(self, target_window=None): + def kill_window(self, target_window: t.Optional[str] = None) -> None: """Close a tmux window, and all panes inside it, ``$ tmux kill-window`` Kill the current window or the window at ``target-window``. removing it @@ -312,7 +320,7 @@ def windows(self) -> t.List[Window]: return self.list_windows() #: Alias :attr:`windows` for :class:`~libtmux.common.TmuxRelationalObject` - children = windows + children = windows # type: ignore # mypy#1362 @property def attached_window(self) -> Window: @@ -339,7 +347,7 @@ def attached_window(self) -> Window: if len(self._windows) == 0: raise exc.LibTmuxException("No Windows") - def select_window(self, target_window: str) -> Window: + def select_window(self, target_window: t.Union[str, int]) -> Window: """ Return :class:`Window` selected via ``$ tmux select-window``. @@ -378,7 +386,9 @@ def attached_pane(self) -> t.Optional["Pane"]: return self.attached_window.attached_pane - def set_option(self, option, value, _global=False): + def set_option( + self, option: str, value: t.Union[str, int], _global: bool = False + ) -> None: """ Set option ``$ tmux set-option