diff --git a/CHANGES b/CHANGES index 64b36ec79..a21ae251a 100644 --- a/CHANGES +++ b/CHANGES @@ -14,6 +14,37 @@ $ pip install --user --upgrade --pre libtmux +### Breaking changes + +- Finding objects / relations + + - 0.16 and below: `session._windows()`, `session.list_windows()`, etc. + + 0.17 and after: {attr}`session.windows ` + + - 0.16 and below: `session.find_where({'window_name': my_window})` + + 0.17 and after: {meth}`session.windows.get(window_name=my_window, default=None) ` + + - If not found and not `default`, raises {exc}`~libtmux._internal.query_list.ObjectDoesNotExist` + - If multiple objects found, raises {exc}`~libtmux._internal.query_list.MultipleObjectsReturned` + + - 0.16 and below: `session.where({'window_name': my_window})` + + 0.17 and after: {meth}`session.windows.filter(window_name=my_window) ` + +- Accessing attributes + + - 0.16 and below: `window['id']` + + 0.17 and after: `window.id` + - 0.16 and below: `window.get('id')` + + 0.17 and after: `window.id` + - 0.16 and below: `window.get('id', None)` + + 0.17 and after: `getattr(window, 'id', None)` + ### New features #### Detect if server active (#448) diff --git a/MIGRATION b/MIGRATION index 9d0d28b5f..987892cba 100644 --- a/MIGRATION +++ b/MIGRATION @@ -19,6 +19,39 @@ well. [tracker]: https://github.com/tmux-python/libtmux/discussions ``` +## 0.17.x: Simplified attributes + +### Finding objects / relations + +- 0.16 and below: `session._windows()`, `session.list_windows()`, etc. + + 0.17 and after: {attr}`session.windows ` + +- 0.16 and below: `session.find_where({'window_name': my_window})` + + 0.17 and after: {meth}`session.windows.get(window_name=my_window, default=None) ` + + - If not found and not `default`, raises {exc}`~libtmux._internal.query_list.ObjectDoesNotExist` + - If multiple objects found, raises {exc}`~libtmux._internal.query_list.MultipleObjectsReturned` + + + +- 0.16 and below: `session.where({'window_name': my_window})` + + 0.17 and after: {meth}`session.windows.filter(window_name=my_window) ` + +### Accessing attributes + +- 0.16 and below: `window['id']` + + 0.17 and after: `window.id` +- 0.16 and below: `window.get('id')` + + 0.17 and after: `window.id` +- 0.16 and below: `window.get('id', None)` + + 0.17 and after: `getattr(window, 'id', None)` + ## Next release _Migration instructions for the upcoming release will be added here_ diff --git a/README.md b/README.md index e71025bbd..dabad67fa 100644 --- a/README.md +++ b/README.md @@ -69,14 +69,14 @@ current tmux server / session / window pane. List sessions: ```python ->>> server.list_sessions() +>>> server.sessions [Session($1 ...), Session($0 ...)] ``` Find session: ```python ->>> server.get_by_id('$1') +>>> server.sessions.filter(session_id='$1')[0] Session($1 ...) ``` @@ -85,7 +85,7 @@ Find session by dict lookup: ```python >>> server.sessions[0].rename_session('foo') Session($1 foo) ->>> server.find_where({ "session_name": "foo" }) +>>> server.sessions.filter(session_name="foo")[0] Session($1 foo) ``` @@ -150,12 +150,14 @@ Type inside the pane (send key strokes): >>> pane.send_keys('echo hey', enter=False) >>> pane.enter() +Pane(%1 ...) ``` Grab the output of pane: ```python >>> pane.clear() # clear the pane +Pane(%1 ...) >>> pane.send_keys("cowsay 'hello'", enter=True) >>> print('\n'.join(pane.cmd('capture-pane', '-p').stdout)) # doctest: +SKIP $ cowsay 'hello' diff --git a/docs/about.md b/docs/about.md index d2aad5aaf..37197e55d 100644 --- a/docs/about.md +++ b/docs/about.md @@ -85,8 +85,8 @@ How is libtmux able to keep references to panes, windows and sessions? > Window index refers to the # of the window in the session. > > To assert pane, window and session data, libtmux will use -> {meth}`Server.list_sessions()`, {meth}`Session.list_windows()`, -> {meth}`Window.list_panes()` to update objects. +> {meth}`Server.sessions()`, {meth}`Session.windows()`, +> {meth}`Window.panes()` to update objects. ## Naming conventions diff --git a/docs/internals/dataclasses.md b/docs/internals/dataclasses.md new file mode 100644 index 000000000..96249dc28 --- /dev/null +++ b/docs/internals/dataclasses.md @@ -0,0 +1,8 @@ +# Dataclass helpers - `libtmux._internal.dataclasses` + +```{eval-rst} +.. automodule:: libtmux._internal.dataclasses + :members: + :special-members: + +``` diff --git a/docs/internals/index.md b/docs/internals/index.md index 348a1df32..cc23abcc4 100644 --- a/docs/internals/index.md +++ b/docs/internals/index.md @@ -3,9 +3,9 @@ # Internals :::{warning} +Be careful with these! Internal APIs are **not** covered by version policies. They can break or be removed between minor versions! -These APIs are internal and not covered by versioning policy. - +If you need an internal API stabilized please [file an issue](https://github.com/tmux-python/libtmux/issues). ::: ## Environmental variables @@ -23,3 +23,10 @@ to split `tmux(1)`'s formatting information. If you find any compatibility problems with the default, or better yet find a string copacetic many environments and tmux releases, note it at . + +## Changes + +```{toctree} +dataclasses +query_list +`` diff --git a/docs/internals/query_list.md b/docs/internals/query_list.md new file mode 100644 index 000000000..29f1f9c33 --- /dev/null +++ b/docs/internals/query_list.md @@ -0,0 +1,6 @@ +# List querying - `libtmux._internal.query_list` + +```{eval-rst} +.. automodule:: libtmux._internal.query_list + :members: +``` diff --git a/docs/quickstart.md b/docs/quickstart.md index 749c4cb95..ed6386a89 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -125,10 +125,10 @@ Windows and Panes. If you have multiple tmux sessions open, you can see that all of the methods in {class}`Server` are available. -We can list sessions with {meth}`Server.list_sessions`: +We can list sessions with {meth}`Server.sessions`: ```python ->>> server.list_sessions() +>>> server.sessions [Session($1 ...), Session($0 ...)] ``` @@ -136,22 +136,22 @@ This returns a list of {class}`Session` objects you can grab. We can find our current session with: ```python ->>> server.list_sessions()[0] +>>> server.sessions[0] Session($1 ...) ``` However, this isn't guaranteed, libtmux works against current tmux information, the session's name could be changed, or another tmux session may be created, -so {meth}`Server.get_by_id` and {meth}`Server.find_where` exists as a lookup. +so {meth}`Server.sessions` and {meth}`Server.windows` exists as a lookup. ## Get session by ID tmux sessions use the `$[0-9]` convention as a way to identify sessions. -`$1` is whatever the ID `list_sessions()` returned above. +`$1` is whatever the ID `sessions()` returned above. ```python ->>> server.get_by_id('$1') +>>> server.sessions.filter(session_id='$1')[0] Session($1 ...) ``` @@ -163,13 +163,16 @@ You may `session = server.get_by_id('$')` to use the session object. >>> server.sessions[0].rename_session('foo') Session($1 foo) ->>> server.find_where({ "session_name": "foo" }) +>>> server.sessions.filter(session_name="foo")[0] +Session($1 foo) + +>>> server.sessions.get(session_name="foo") Session($1 foo) ``` -With `find_where`, pass in a dict and return the first object found. In +With `filter`, pass in attributes and return a list of matches. In this case, a {class}`Server` holds a collection of child {class}`Session`. -{class}`Session` and {class}`Window` both utilize `find_where` to sift +{class}`Session` and {class}`Window` both utilize `filter` to sift through Windows and Panes, respectively. So you may now use: @@ -178,7 +181,7 @@ So you may now use: >>> server.sessions[0].rename_session('foo') Session($1 foo) ->>> session = server.find_where({ "session_name": "foo" }) +>>> session = server.sessions.get(session_name="foo") >>> session Session($1 foo) ``` @@ -213,7 +216,7 @@ Let's delete that window ({meth}`Session.kill_window`). Method 1: Use passthrough to tmux's `target` system. ```python ->>> session.kill_window(window.id) +>>> session.kill_window(window.window_id) ``` The window in the bg dissappeared. This was the equivalent of @@ -260,7 +263,7 @@ And kill: >>> window.kill_window() ``` -Use {meth}`Session.list_windows()` and {meth}`Session.find_where()` to list and sort +Use {meth}`Session.windows` and {meth}`Session.windows.filter()` to list and sort through active {class}`Window`'s. ## Manipulating windows @@ -346,6 +349,7 @@ using {meth}`Pane.enter()`: ```python >>> pane.enter() +Pane(%1 ...) ``` ### Avoid cluttering shell history diff --git a/docs/reference/properties.md b/docs/reference/properties.md index 085361f3f..2f5e63906 100644 --- a/docs/reference/properties.md +++ b/docs/reference/properties.md @@ -49,25 +49,24 @@ Session($1 libtmux_...) Quick access to basic attributes: ```python ->>> session.name +>>> session.session_name 'libtmux_...' ->>> session.id +>>> session.session_id '$1' ``` To see all attributes for a session: ```python ->>> sorted(list(session._info.keys())) +from libtmux.neo import Obj + +>>> sorted(list(Obj.__dataclass_fields__.keys())) ['session_attached', 'session_created', ...] ``` -Some may conflict with python API, to access them, you can use `.get()`, to get the count -of sessions in a window: - ```python ->>> session.get('session_windows') +>>> session.session_windows '...' ``` @@ -85,30 +84,23 @@ Window(@1 ...:..., Session($1 ...)) Basics: ```python ->>> window.name +>>> window.window_name '...' ->>> window.id +>>> window.window_id '@1' ->>> window.height +>>> window.window_height '...' ->>> window.width +>>> window.window_width '...' ``` -Everything available: - -```python ->>> sorted(list(window.keys())) -['session_id', 'session_name', 'window_active', ..., 'window_width'] -``` - -Use `get()` for details not accessible via properties: +Use attribute access for details not accessible via properties: ```python ->>> window.get('window_panes') +>>> window.window_panes '1' ``` @@ -126,33 +118,20 @@ Pane(%1 Window(@1 ...:..., Session($1 libtmux_...))) Basics: ```python ->>> pane.current_command +>>> pane.pane_current_command '...' ->>> type(pane.current_command) +>>> type(pane.pane_current_command) ->>> pane.height +>>> pane.pane_height '...' ->>> pane.width +>>> pane.pane_width '...' ->>> pane.index +>>> pane.pane_index '0' ``` -Everything: - -````python ->>> sorted(list(pane._info.keys())) -['alternate_on', 'alternate_saved_x', ..., 'wrap_flag'] - -Use `get()` for details keys: - -```python ->>> pane.get('pane_width') -'...' -```` - [formats]: http://man.openbsd.org/OpenBSD-5.9/man1/tmux.1#FORMATS diff --git a/src/libtmux/_internal/__init__.py b/src/libtmux/_internal/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/libtmux/_internal/dataclasses.py b/src/libtmux/_internal/dataclasses.py new file mode 100644 index 000000000..5cabfb19d --- /dev/null +++ b/src/libtmux/_internal/dataclasses.py @@ -0,0 +1,84 @@ +""":mod:`dataclasses` utilities. + +Note +---- +This is an internal API not covered by versioning policy. +""" +import dataclasses +from operator import attrgetter + + +class SkipDefaultFieldsReprMixin: + r"""Skip default fields in :func:`~dataclasses.dataclass` + :func:`object representation `. + + Notes + ----- + + Credit: Pietro Oldrati, 2022-05-08, Unilicense + + https://stackoverflow.com/a/72161437/1396928 + + Examples + -------- + + >>> @dataclasses.dataclass() + ... class Item: + ... name: str + ... unit_price: float = 1.00 + ... quantity_on_hand: int = 0 + ... + + >>> @dataclasses.dataclass(repr=False) + ... class ItemWithMixin(SkipDefaultFieldsReprMixin): + ... name: str + ... unit_price: float = 1.00 + ... quantity_on_hand: int = 0 + ... + + >>> Item('Test') + Item(name='Test', unit_price=1.0, quantity_on_hand=0) + + >>> ItemWithMixin('Test') + ItemWithMixin(name=Test) + + >>> Item('Test', quantity_on_hand=2) + Item(name='Test', unit_price=1.0, quantity_on_hand=2) + + >>> ItemWithMixin('Test', quantity_on_hand=2) + ItemWithMixin(name=Test, quantity_on_hand=2) + + If you want to copy/paste the :meth:`~.__repr__()` + directly, you can omit the ``repr=False``: + + >>> @dataclasses.dataclass + ... class ItemWithMixin(SkipDefaultFieldsReprMixin): + ... name: str + ... unit_price: float = 1.00 + ... quantity_on_hand: int = 0 + ... __repr__ = SkipDefaultFieldsReprMixin.__repr__ + ... + + >>> ItemWithMixin('Test') + ItemWithMixin(name=Test) + + >>> ItemWithMixin('Test', unit_price=2.00) + ItemWithMixin(name=Test, unit_price=2.0) + + >>> item = ItemWithMixin('Test') + >>> item.unit_price = 2.05 + + >>> item + ItemWithMixin(name=Test, unit_price=2.05) + """ + + def __repr__(self) -> str: + """Omit default fields in object representation.""" + nodef_f_vals = ( + (f.name, attrgetter(f.name)(self)) + for f in dataclasses.fields(self) + if attrgetter(f.name)(self) != f.default + ) + + nodef_f_repr = ", ".join(f"{name}={value}" for name, value in nodef_f_vals) + return f"{self.__class__.__name__}({nodef_f_repr})" diff --git a/src/libtmux/_internal/query_list.py b/src/libtmux/_internal/query_list.py new file mode 100644 index 000000000..f4ea936ce --- /dev/null +++ b/src/libtmux/_internal/query_list.py @@ -0,0 +1,384 @@ +"""Utilities for filtering or searching :class:`list` of objects / list data. + +Note +---- +This is an internal API not covered by versioning policy. +""" +import re +import traceback +from collections.abc import Mapping, Sequence +from re import Pattern +from typing import TYPE_CHECKING, Any, Callable, List, Optional, TypeVar, Union + +if TYPE_CHECKING: + from typing_extensions import Protocol + + class LookupProtocol(Protocol): + """Protocol for :class:`QueryList` filtering operators.""" + + def __call__( + self, + data: Union[str, List[str], "Mapping[str, str]"], + rhs: Union[str, List[str], "Mapping[str, str]", "Pattern[str]"], + ) -> bool: + """Callback for :class:`QueryList` filtering operators.""" + ... + + +T = TypeVar("T", Any, Any) + +no_arg = object() + + +class MultipleObjectsReturned(Exception): + """The requested object does not exist""" + + +class ObjectDoesNotExist(Exception): + """The query returned multiple objects when only one was expected.""" + + +def keygetter( + obj: "Mapping[str, Any]", + path: str, +) -> Union[None, Any, str, List[str], "Mapping[str, str]"]: + """obj, "foods__breakfast", obj['foods']['breakfast'] + + >>> keygetter({ "foods": { "breakfast": "cereal" } }, "foods__breakfast") + 'cereal' + >>> keygetter({ "foods": { "breakfast": "cereal" } }, "foods") + {'breakfast': 'cereal'} + + """ + try: + sub_fields = path.split("__") + dct = obj + for sub_field in sub_fields: + if isinstance(dct, dict): + dct = dct[sub_field] + elif hasattr(dct, sub_field): + dct = getattr(dct, sub_field) + + return dct + except Exception as e: + traceback.print_stack() + print(f"Above error was {e}") + return None + + +def parse_lookup(obj: "Mapping[str, Any]", path: str, lookup: str) -> Optional[Any]: + """Check if field lookup key, e.g. "my__path__contains" has comparator, return val. + + If comparator not used or value not found, return None. + + mykey__endswith("mykey") -> "mykey" else None + + >>> parse_lookup({ "food": "red apple" }, "food__istartswith", "__istartswith") + 'red apple' + """ + try: + if isinstance(path, str) and isinstance(lookup, str) and path.endswith(lookup): + field_name = path.rsplit(lookup)[0] + if field_name is not None: + return keygetter(obj, field_name) + except Exception: + traceback.print_stack() + return None + + +def lookup_exact( + data: Union[str, List[str], "Mapping[str, str]"], + rhs: Union[str, List[str], "Mapping[str, str]", "Pattern[str]"], +) -> bool: + return rhs == data + + +def lookup_iexact( + data: Union[str, List[str], "Mapping[str, str]"], + rhs: Union[str, List[str], "Mapping[str, str]", "Pattern[str]"], +) -> bool: + if not isinstance(rhs, str) or not isinstance(data, str): + return False + + return rhs.lower() == data.lower() + + +def lookup_contains( + data: Union[str, List[str], "Mapping[str, str]"], + rhs: Union[str, List[str], "Mapping[str, str]", "Pattern[str]"], +) -> bool: + if not isinstance(rhs, str) or not isinstance(data, (str, Mapping, list)): + return False + + return rhs in data + + +def lookup_icontains( + data: Union[str, List[str], "Mapping[str, str]"], + rhs: Union[str, List[str], "Mapping[str, str]", "Pattern[str]"], +) -> bool: + if not isinstance(rhs, str) or not isinstance(data, (str, Mapping, list)): + return False + + if isinstance(data, str): + return rhs.lower() in data.lower() + if isinstance(data, Mapping): + return rhs.lower() in [k.lower() for k in data.keys()] + + return False + + +def lookup_startswith( + data: Union[str, List[str], "Mapping[str, str]"], + rhs: Union[str, List[str], "Mapping[str, str]", "Pattern[str]"], +) -> bool: + if not isinstance(rhs, str) or not isinstance(data, str): + return False + + return data.startswith(rhs) + + +def lookup_istartswith( + data: Union[str, List[str], "Mapping[str, str]"], + rhs: Union[str, List[str], "Mapping[str, str]", "Pattern[str]"], +) -> bool: + if not isinstance(rhs, str) or not isinstance(data, str): + return False + + return data.lower().startswith(rhs.lower()) + + +def lookup_endswith( + data: Union[str, List[str], "Mapping[str, str]"], + rhs: Union[str, List[str], "Mapping[str, str]", "Pattern[str]"], +) -> bool: + if not isinstance(rhs, str) or not isinstance(data, str): + return False + + return data.endswith(rhs) + + +def lookup_iendswith( + data: Union[str, List[str], "Mapping[str, str]"], + rhs: Union[str, List[str], "Mapping[str, str]", "Pattern[str]"], +) -> bool: + if not isinstance(rhs, str) or not isinstance(data, str): + return False + return data.lower().endswith(rhs.lower()) + + +def lookup_in( + data: Union[str, List[str], "Mapping[str, str]"], + rhs: Union[str, List[str], "Mapping[str, str]", "Pattern[str]"], +) -> bool: + if isinstance(rhs, list): + return data in rhs + + try: + if isinstance(rhs, str) and isinstance(data, Mapping): + return rhs in data + if isinstance(rhs, str) and isinstance(data, (str, list)): + return rhs in data + if isinstance(rhs, str) and isinstance(data, Mapping): + return rhs in data + # TODO: Add a deep Mappingionary matcher + # if isinstance(rhs, Mapping) and isinstance(data, Mapping): + # return rhs.items() not in data.items() + except Exception: + return False + return False + + +def lookup_nin( + data: Union[str, List[str], "Mapping[str, str]"], + rhs: Union[str, List[str], "Mapping[str, str]", "Pattern[str]"], +) -> bool: + if isinstance(rhs, list): + return data not in rhs + + try: + if isinstance(rhs, str) and isinstance(data, Mapping): + return rhs not in data + if isinstance(rhs, str) and isinstance(data, (str, list)): + return rhs not in data + if isinstance(rhs, str) and isinstance(data, Mapping): + return rhs not in data + # TODO: Add a deep Mappingionary matcher + # if isinstance(rhs, Mapping) and isinstance(data, Mapping): + # return rhs.items() not in data.items() + except Exception: + return False + return False + + +def lookup_regex( + data: Union[str, List[str], "Mapping[str, str]"], + rhs: Union[str, List[str], "Mapping[str, str]", "Pattern[str]"], +) -> bool: + if isinstance(data, (str, bytes, re.Pattern)) and isinstance(rhs, (str, bytes)): + return bool(re.search(rhs, data)) + return False + + +def lookup_iregex( + data: Union[str, List[str], "Mapping[str, str]"], + rhs: Union[str, List[str], "Mapping[str, str]", "Pattern[str]"], +) -> bool: + if isinstance(data, (str, bytes, re.Pattern)) and isinstance(rhs, (str, bytes)): + return bool(re.search(rhs, data, re.IGNORECASE)) + return False + + +LOOKUP_NAME_MAP: 'Mapping[str, "LookupProtocol"]' = { + "eq": lookup_exact, + "exact": lookup_exact, + "iexact": lookup_iexact, + "contains": lookup_contains, + "icontains": lookup_icontains, + "startswith": lookup_startswith, + "istartswith": lookup_istartswith, + "endswith": lookup_endswith, + "iendswith": lookup_iendswith, + "in": lookup_in, + "nin": lookup_nin, + "regex": lookup_regex, + "iregex": lookup_iregex, +} + + +class QueryList(List[T]): + """Filter list of object/dictionaries. For small, local datasets. + + *Experimental, unstable*. + + >>> query = QueryList( + ... [ + ... { + ... "place": "Largo", + ... "city": "Tampa", + ... "state": "Florida", + ... "foods": {"fruit": ["banana", "orange"], "breakfast": "cereal"}, + ... }, + ... { + ... "place": "Chicago suburbs", + ... "city": "Elmhurst", + ... "state": "Illinois", + ... "foods": {"fruit": ["apple", "cantelope"], "breakfast": "waffles"}, + ... }, + ... ] + ... ) + >>> query.filter(place="Chicago suburbs")[0]['city'] + 'Elmhurst' + >>> query.filter(place__icontains="chicago")[0]['city'] + 'Elmhurst' + >>> query.filter(foods__breakfast="waffles")[0]['city'] + 'Elmhurst' + >>> query.filter(foods__fruit__in="cantelope")[0]['city'] + 'Elmhurst' + >>> query.filter(foods__fruit__in="orange")[0]['city'] + 'Tampa' + >>> query.get(foods__fruit__in="orange")['city'] + 'Tampa' + """ + + data: "Sequence[T]" + pk_key: Optional[str] + + def items(self) -> List[T]: + data: "Sequence[T]" + + if self.pk_key is None: + raise Exception("items() require a pk_key exists") + return [(getattr(item, self.pk_key), item) for item in self] + + def __eq__( + self, + other: object, + # other: Union[ + # "QueryList[T]", + # List[Mapping[str, str]], + # List[Mapping[str, int]], + # List[Mapping[str, Union[str, Mapping[str, Union[List[str], str]]]]], + # ], + ) -> bool: + data = other + + if not isinstance(self, list) or not isinstance(data, list): + return False + + if len(self) == len(data): + for (a, b) in zip(self, data): + if isinstance(a, Mapping): + a_keys = a.keys() + if a.keys == b.keys(): + for key in a_keys: + if abs(a[key] - b[key]) > 1: + return False + else: + if a != b: + return False + + return True + return False + + def filter( + self, matcher: Optional[Union[Callable[[T], bool], T]] = None, **kwargs: Any + ) -> "QueryList[T]": + """Filter list of objects.""" + + def filter_lookup(obj: Any) -> bool: + for path, v in kwargs.items(): + try: + lhs, op = path.rsplit("__", 1) + + if op not in LOOKUP_NAME_MAP: + raise ValueError(f"{op} not in LOOKUP_NAME_MAP") + except ValueError: + lhs = path + op = "exact" + + assert op in LOOKUP_NAME_MAP + path = lhs + data = keygetter(obj, path) + + if data is None or not LOOKUP_NAME_MAP[op](data, v): + return False + + return True + + if callable(matcher): + _filter = matcher + elif matcher is not None: + + def val_match(obj: Union[str, List[Any]]) -> bool: + if isinstance(matcher, list): + return obj in matcher + else: + return obj == matcher + + _filter = val_match + else: + _filter = filter_lookup + + return self.__class__(k for k in self if _filter(k)) + + def get( + self, + matcher: Optional[Union[Callable[[T], bool], T]] = None, + default: Optional[Any] = no_arg, + **kwargs: Any, + ) -> Optional[T]: + """Retrieve one object + + Raises exception if multiple objects found. + + Raises exception if no object found, unless ``default`` + """ + objs = self.filter(matcher=matcher, **kwargs) + if len(objs) > 1: + raise MultipleObjectsReturned() + elif len(objs) == 0: + if default == no_arg: + raise ObjectDoesNotExist() + return default + return objs[0] diff --git a/src/libtmux/common.py b/src/libtmux/common.py index 88029b30c..281abc7ec 100644 --- a/src/libtmux/common.py +++ b/src/libtmux/common.py @@ -274,174 +274,6 @@ def __init__(self, *args: t.Any, **kwargs: t.Any) -> None: ) -# 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`, - :class:`Session` and :class:`Server`. - - Instance attributes for useful information :term:`tmux(1)` uses for - Session, Window, Pane, stored :attr:`self._info`. For example, a - :class:`Window` will have a ``window_id`` and ``window_name``. - - ================ ================================== ============== - Object formatter_prefix value - ================ ================================== ============== - :class:`Server` n/a n/a - :class:`Session` :attr:`Session.formatter_prefix` session\_ - :class:`Window` :attr:`Window.formatter_prefix` window\_ - :class:`Pane` :attr:`Pane.formatter_prefix` pane\_ - ================ ================================== ============== - """ - _info: t.Dict[t.Any, t.Any] - formatter_prefix: str - - 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: str, value: str) -> None: - self._info[key] = value - self.dirty = True - - def __delitem__(self, key: str) -> None: - del self._info[key] - self.dirty = True - - def keys(self) -> KeysView[str]: - """Return list of keys.""" - return self._info.keys() - - def __iter__(self) -> t.Iterator[str]: - return self._info.__iter__() - - def __len__(self) -> int: - return len(self._info.keys()) - - def __getattr__(self, key: str) -> str: - try: - 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}") - - -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 - :class:`Session` objects, a :class:`Session` has collection of - :class:`Window`) - - Children of :class:`TmuxRelationalObject` are going to have a - ``self.children``, ``self.child_id_attribute``. - - ================ ========================= ================================= - Object .children method - ================ ========================= ================================= - :class:`Server` :attr:`Server._sessions` :meth:`Server.list_sessions` - :class:`Session` :attr:`Session._windows` :meth:`Session.list_windows` - :class:`Window` :attr:`Window._panes` :meth:`Window.list_panes` - :class:`Pane` n/a n/a - ================ ========================= ================================= - - ================ ================================== ============== - Object child_id_attribute value - ================ ================================== ============== - :class:`Server` :attr:`Server.child_id_attribute` session_id - :class:`Session` :attr:`Session.child_id_attribute` window_id - :class:`Window` :attr:`Window.child_id_attribute` pane_id - :class:`Pane` n/a n/a - ================ ================================== ============== - """ - - children: t.List[O] - child_id_attribute: str - - def find_where(self, attrs: D) -> Optional[Union["Pane", "Window", "Session"]]: - """Return object on first match. - - .. versionchanged:: 0.4 - Renamed from ``.findWhere`` to ``.find_where``. - - """ - try: - return self.where(attrs)[0] - except IndexError: - return None - - @overload - def where(self, attrs: D, first: "Literal[True]") -> O: - ... - - @overload - def where(self, attrs: D, first: "Literal[False]") -> t.List[O]: - ... - - @overload - def where(self, attrs: D) -> t.List[O]: - ... - - def where(self, attrs: D, first: bool = False) -> t.Union[List[O], O]: - """ - Return objects matching child objects properties. - - Parameters - ---------- - attrs : dict - tmux properties to match values of - - Returns - ------- - list of objects, or one object if ``first=True`` - """ - - # from https://github.com/serkanyersen/underscore.py - def by(val: O) -> bool: - for key in attrs.keys(): - try: - if attrs[key] != val[key]: - return False - except KeyError: - return False - return True - - 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: str) -> Optional[O]: - """ - Return object based on ``child_id_attribute``. - - Parameters - ---------- - val : str - - Returns - ------- - object - """ - for child in self.children: - if child[self.child_id_attribute] == id: - return child - else: - continue - - return None - - def get_version() -> LooseVersion: """ Return tmux version. diff --git a/src/libtmux/formats.py b/src/libtmux/formats.py index ce3745318..b74bd989f 100644 --- a/src/libtmux/formats.py +++ b/src/libtmux/formats.py @@ -19,7 +19,7 @@ "session_created", "session_created_string", "session_attached", - "session_grouped", + # "session_grouped", Apparently unused, mistake found while adding dataclasses "session_group", ] @@ -64,7 +64,7 @@ "pane_index", "pane_width", "pane_height", - "pane_title", + # "pane_title", # removed in 3.1+ "pane_id", "pane_active", "pane_dead", diff --git a/src/libtmux/neo.py b/src/libtmux/neo.py new file mode 100644 index 000000000..be1621214 --- /dev/null +++ b/src/libtmux/neo.py @@ -0,0 +1,244 @@ +import dataclasses +import logging +import typing as t + +from libtmux import exc +from libtmux.common import tmux_cmd +from libtmux.formats import FORMAT_SEPARATOR + +if t.TYPE_CHECKING: + from typing_extensions import Literal + + ListCmd = Literal["list-sessions", "list-windows", "list-panes"] + ListExtraArgs = t.Optional[t.Iterable[str]] + + from libtmux.server import Server + +logger = logging.getLogger(__name__) + + +OutputRaw = t.Dict[str, t.Any] +OutputsRaw = t.List[OutputRaw] + + +""" +Quirks: + +QUIRK_TMUX_3_1_X_0001: + +- tmux 3.1 and 3.1a: +- server crash with list-panes w/ buffer_created, client_activity, client_created +""" + + +@dataclasses.dataclass() +class Obj: + server: "Server" + + active_window_index: t.Union[str, None] = None + alternate_saved_x: t.Union[str, None] = None + alternate_saved_y: t.Union[str, None] = None + # See QUIRK_TMUX_3_1_X_0001 + # buffer_created: t.Union[str, None] = None + buffer_name: t.Union[str, None] = None + buffer_sample: t.Union[str, None] = None + buffer_size: t.Union[str, None] = None + # See QUIRK_TMUX_3_1_X_0001 + # client_activity: t.Union[str, None] = None + client_cell_height: t.Union[str, None] = None + client_cell_width: t.Union[str, None] = None + # See QUIRK_TMUX_3_1_X_0001 + # client_created: t.Union[str, None] = None + client_discarded: t.Union[str, None] = None + client_flags: t.Union[str, None] = None + client_height: t.Union[str, None] = None + client_key_table: t.Union[str, None] = None + client_name: t.Union[str, None] = None + client_pid: t.Union[str, None] = None + client_termname: t.Union[str, None] = None + client_tty: t.Union[str, None] = None + client_uid: t.Union[str, None] = None + client_user: t.Union[str, None] = None + client_width: t.Union[str, None] = None + client_written: t.Union[str, None] = None + command_list_alias: t.Union[str, None] = None + command_list_name: t.Union[str, None] = None + command_list_usage: t.Union[str, None] = None + config_files: t.Union[str, None] = None + copy_cursor_line: t.Union[str, None] = None + copy_cursor_word: t.Union[str, None] = None + copy_cursor_x: t.Union[str, None] = None + copy_cursor_y: t.Union[str, None] = None + current_file: t.Union[str, None] = None + cursor_character: t.Union[str, None] = None + cursor_flag: t.Union[str, None] = None + cursor_x: t.Union[str, None] = None + cursor_y: t.Union[str, None] = None + history_bytes: t.Union[str, None] = None + history_limit: t.Union[str, None] = None + history_size: t.Union[str, None] = None + insert_flag: t.Union[str, None] = None + keypad_cursor_flag: t.Union[str, None] = None + keypad_flag: t.Union[str, None] = None + last_window_index: t.Union[str, None] = None + line: t.Union[str, None] = None + mouse_all_flag: t.Union[str, None] = None + mouse_any_flag: t.Union[str, None] = None + mouse_button_flag: t.Union[str, None] = None + mouse_sgr_flag: t.Union[str, None] = None + mouse_standard_flag: t.Union[str, None] = None + next_session_id: t.Union[str, None] = None + origin_flag: t.Union[str, None] = None + pane_active: t.Union[str, None] = None # Not detected by script + pane_bg: t.Union[str, None] = None + pane_bottom: t.Union[str, None] = None + pane_current_command: t.Union[str, None] = None + pane_current_path: t.Union[str, None] = None + pane_dead_signal: t.Union[str, None] = None + pane_dead_status: t.Union[str, None] = None + pane_dead_time: t.Union[str, None] = None + pane_fg: t.Union[str, None] = None + pane_height: t.Union[str, None] = None + pane_id: t.Union[str, None] = None + pane_index: t.Union[str, None] = None + pane_left: t.Union[str, None] = None + pane_pid: t.Union[str, None] = None + pane_right: t.Union[str, None] = None + pane_search_string: t.Union[str, None] = None + pane_start_command: t.Union[str, None] = None + pane_start_path: t.Union[str, None] = None + pane_tabs: t.Union[str, None] = None + pane_top: t.Union[str, None] = None + pane_tty: t.Union[str, None] = None + pane_width: t.Union[str, None] = None + + pid: t.Union[str, None] = None + scroll_position: t.Union[str, None] = None + scroll_region_lower: t.Union[str, None] = None + scroll_region_upper: t.Union[str, None] = None + search_match: t.Union[str, None] = None + selection_end_x: t.Union[str, None] = None + selection_end_y: t.Union[str, None] = None + selection_start_x: t.Union[str, None] = None + selection_start_y: t.Union[str, None] = None + session_activity: t.Union[str, None] = None + session_alerts: t.Union[str, None] = None + session_attached: t.Union[str, None] = None + session_attached_list: t.Union[str, None] = None + session_created: t.Union[str, None] = None + session_group: t.Union[str, None] = None + session_group_attached: t.Union[str, None] = None + session_group_list: t.Union[str, None] = None + session_group_size: t.Union[str, None] = None + session_id: t.Union[str, None] = None + session_last_attached: t.Union[str, None] = None + session_name: t.Union[str, None] = None + session_path: t.Union[str, None] = None + session_stack: t.Union[str, None] = None + session_windows: t.Union[str, None] = None + socket_path: t.Union[str, None] = None + start_time: t.Union[str, None] = None + uid: t.Union[str, None] = None + user: t.Union[str, None] = None + version: t.Union[str, None] = None + window_active: t.Union[str, None] = None # Not detected by script + window_active_clients: t.Union[str, None] = None + window_active_sessions: t.Union[str, None] = None + window_activity: t.Union[str, None] = None + window_cell_height: t.Union[str, None] = None + window_cell_width: t.Union[str, None] = None + window_height: t.Union[str, None] = None + window_id: t.Union[str, None] = None + window_index: t.Union[str, None] = None + window_layout: t.Union[str, None] = None + window_linked: t.Union[str, None] = None + window_linked_sessions: t.Union[str, None] = None + window_linked_sessions_list: t.Union[str, None] = None + window_marked_flag: t.Union[str, None] = None + window_name: t.Union[str, None] = None + window_offset_x: t.Union[str, None] = None + window_offset_y: t.Union[str, None] = None + window_panes: t.Union[str, None] = None + window_raw_flags: t.Union[str, None] = None + window_stack_index: t.Union[str, None] = None + window_width: t.Union[str, None] = None + wrap_flag: t.Union[str, None] = None + + def _refresh( + self, + obj_key: str, + obj_id: str, + list_cmd: "ListCmd" = "list-panes", + ) -> None: + assert isinstance(obj_id, str) + obj = fetch_obj( + obj_key=obj_key, obj_id=obj_id, list_cmd=list_cmd, server=self.server + ) + assert obj is not None + if obj is not None: + for k, v in obj.items(): + setattr(self, k, v) + + +def fetch_objs( + server: "Server", list_cmd: "ListCmd", list_extra_args: "ListExtraArgs" = None +) -> OutputsRaw: + formats = list(Obj.__dataclass_fields__.keys()) + + cmd_args: t.List[t.Union[str, int]] = list() + + if server.socket_name: + cmd_args.insert(0, f"-L{server.socket_name}") + if server.socket_path: + cmd_args.insert(0, f"-S{server.socket_path}") + tmux_formats = [f"#{{{f}}}{FORMAT_SEPARATOR}" for f in formats] + + tmux_cmds = [ + *cmd_args, + list_cmd, + ] + + if list_extra_args is not None and isinstance(list_extra_args, t.Iterable): + tmux_cmds.extend(list(list_extra_args)) + + tmux_cmds.append("-F%s" % "".join(tmux_formats)) + + proc = tmux_cmd(*tmux_cmds) # output + + if proc.stderr: + raise exc.LibTmuxException(proc.stderr) + + obj_output = proc.stdout + + obj_formatters = [ + dict(zip(formats, formatter.split(FORMAT_SEPARATOR))) + for formatter in obj_output + ] + + # Filter empty values + obj_formatters_filtered = [ + {k: v for k, v in formatter.items() if v} for formatter in obj_formatters + ] + + return obj_formatters_filtered + + +def fetch_obj( + server: "Server", + obj_key: str, + obj_id: str, + list_cmd: "ListCmd" = "list-panes", + list_extra_args: "ListExtraArgs" = None, +) -> OutputRaw: + obj_formatters_filtered = fetch_objs( + server=server, list_cmd=list_cmd, list_extra_args=list_extra_args + ) + + obj = None + for _obj in obj_formatters_filtered: + if _obj.get(obj_key) == obj_id: + obj = _obj + + assert obj is not None + + return obj diff --git a/src/libtmux/pane.py b/src/libtmux/pane.py index 469a0d9b3..bdc161b0d 100644 --- a/src/libtmux/pane.py +++ b/src/libtmux/pane.py @@ -5,14 +5,17 @@ ~~~~~~~~~~~~ """ +import dataclasses import logging import typing as t +import warnings from typing import overload from libtmux.common import tmux_cmd +from libtmux.neo import Obj, fetch_obj -from . import exc -from .common import PaneDict, TmuxMappingObject, TmuxRelationalObject +from . import exc, formats +from .common import PaneDict if t.TYPE_CHECKING: from typing_extensions import Literal @@ -25,7 +28,8 @@ logger = logging.getLogger(__name__) -class Pane(TmuxMappingObject): +@dataclasses.dataclass() +class Pane(Obj): """ A :term:`tmux(1)` :term:`Pane` [pane_manual]_. @@ -68,47 +72,41 @@ class Pane(TmuxMappingObject): Accessed April 1st, 2018. """ - formatter_prefix = "pane_" - """Namespace used for :class:`~libtmux.common.TmuxMappingObject`""" - window: "Window" - """:class:`libtmux.Window` pane is linked to""" - session: "Session" - """:class:`libtmux.Session` pane is linked to""" server: "Server" - """:class:`libtmux.Server` pane is linked to""" - def __init__( - self, - window: "Window", - 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 = pane_id + def refresh(self) -> None: + assert isinstance(self.pane_id, str) + return super()._refresh(obj_key="pane_id", obj_id=self.pane_id) + + @classmethod + def from_pane_id(cls, server: "Server", pane_id: str) -> "Pane": + pane = fetch_obj( + obj_key="pane_id", + obj_id=pane_id, + server=server, + list_cmd="list-panes", + list_extra_args=("-a",), + ) + return cls(server=server, **pane) - self.server._update_panes() + # + # Relations + # @property - def _info(self) -> PaneDict: # type: ignore # mypy#1362 - attrs = {"pane_id": self._pane_id} - - # from https://github.com/serkanyersen/underscore.py - def by(val: PaneDict) -> bool: - for key in attrs.keys(): - try: - if attrs[key] != val[key]: - return False - except KeyError: - return False - return True + def window(self) -> "Window": + assert isinstance(self.window_id, str) + from libtmux.window import Window - target_panes = [s for s in self.server._panes if by(s)] + return Window.from_window_id(server=self.server, window_id=self.window_id) - return target_panes[0] + @property + def session(self) -> "Session": + return self.window.session + # + # Command (pane-scoped) + # def cmd(self, cmd: str, *args: t.Any, **kwargs: t.Any) -> tmux_cmd: """Return :meth:`Server.cmd` defaulting to ``target_pane`` as target. @@ -116,16 +114,56 @@ def cmd(self, cmd: str, *args: t.Any, **kwargs: t.Any) -> tmux_cmd: Specifying ``('-t', 'custom-target')`` or ``('-tcustom_target')`` in ``args`` will override using the object's ``pane_id`` as target. - - Returns - ------- - :class:`Server.cmd` """ if not any(arg.startswith("-t") for arg in args): - args = ("-t", self.get("pane_id")) + args + args = ("-t", self.pane_id) + args return self.server.cmd(cmd, *args, **kwargs) + # + # Commands (tmux-like) + # + def resize_pane(self, *args: t.Any, **kwargs: t.Any) -> "Pane": + """ + ``$ tmux resize-pane`` of pane and return ``self``. + + Parameters + ---------- + target_pane : str + ``target_pane``, or ``-U``,``-D``, ``-L``, ``-R``. + + Other Parameters + ---------------- + height : int + ``resize-pane -y`` dimensions + width : int + ``resize-pane -x`` dimensions + + Raises + ------ + exc.LibTmuxException + """ + if "height" in kwargs: + proc = self.cmd("resize-pane", "-y%s" % int(kwargs["height"])) + elif "width" in kwargs: + proc = self.cmd("resize-pane", "-x%s" % int(kwargs["width"])) + else: + proc = self.cmd("resize-pane", args[0]) + + if proc.stderr: + raise exc.LibTmuxException(proc.stderr) + + self.refresh() + return self + + def capture_pane(self) -> t.Union[str, t.List[str]]: + """ + Capture text from pane. + + ``$ tmux capture-pane`` to pane. + """ + return self.cmd("capture-pane", "-p").stdout + def send_keys( self, cmd: str, @@ -204,11 +242,6 @@ def display_message( get_text : bool, optional Returns only text without displaying a message in target-client status line. - - Returns - ------- - :class:`list` - :class:`None` """ if get_text: return self.cmd("display-message", "-p", cmd).stdout @@ -216,14 +249,25 @@ def display_message( self.cmd("display-message", cmd) return None - def clear(self) -> None: - """Clear pane.""" - self.send_keys("reset") - - def reset(self) -> None: - """Reset and clear pane history.""" + # + # Commands ("climber"-helpers) + # + # These are commands that climb to the parent scope's methods with + # additional scoped window info. + # + def select_pane(self) -> "Pane": + """ + Select pane. Return ``self``. - self.cmd("send-keys", r"-R \; clear-history") + To select a window object asynchrously. If a ``pane`` object exists + and is no longer longer the current window, ``w.select_pane()`` + will make ``p`` the current pane. + """ + assert isinstance(self.pane_id, str) + pane = self.window.select_pane(self.pane_id) + if pane is None: + raise exc.LibTmuxException(f"Pane not found: {self}") + return pane def split_window( self, @@ -231,7 +275,7 @@ def split_window( vertical: bool = True, start_directory: t.Optional[str] = None, percent: t.Optional[int] = None, - ) -> "Pane": + ) -> "Pane": # New Pane, not self """ Split window at pane and return newly created :class:`Pane`. @@ -245,20 +289,19 @@ def split_window( specifies the working directory in which the new pane is created. percent: int, optional percentage to occupy with respect to current pane - - Returns - ------- - :class:`Pane` """ return self.window.split_window( - target=self.get("pane_id"), + target=self.pane_id, start_directory=start_directory, attach=attach, vertical=vertical, percent=percent, ) - def set_width(self, width: int) -> None: + # + # Commands (helpers) + # + def set_width(self, width: int) -> "Pane": """ Set width of pane. @@ -268,8 +311,9 @@ def set_width(self, width: int) -> None: pane width, in cells """ self.resize_pane(width=width) + return self - def set_height(self, height: int) -> None: + def set_height(self, height: int) -> "Pane": """ Set height of pane. @@ -279,82 +323,102 @@ def set_height(self, height: int) -> None: height of pain, in cells """ self.resize_pane(height=height) + return self - def resize_pane(self, *args: t.Any, **kwargs: t.Any) -> "Pane": + def enter(self) -> "Pane": """ - ``$ tmux resize-pane`` of pane and return ``self``. + Send carriage return to pane. - Parameters - ---------- - target_pane : str - ``target_pane``, or ``-U``,``-D``, ``-L``, ``-R``. + ``$ tmux send-keys`` send Enter to the pane. + """ + self.cmd("send-keys", "Enter") + return self - Other Parameters - ---------------- - height : int - ``resize-pane -y`` dimensions - width : int - ``resize-pane -x`` dimensions + def clear(self) -> "Pane": + """Clear pane.""" + self.send_keys("reset") + return self - Returns - ------- - :class:`Pane` + def reset(self) -> "Pane": + """Reset and clear pane history.""" - Raises - ------ - exc.LibTmuxException - """ - if "height" in kwargs: - proc = self.cmd("resize-pane", "-y%s" % int(kwargs["height"])) - elif "width" in kwargs: - proc = self.cmd("resize-pane", "-x%s" % int(kwargs["width"])) - else: - proc = self.cmd("resize-pane", args[0]) + self.cmd("send-keys", r"-R \; clear-history") + return self - if proc.stderr: - raise exc.LibTmuxException(proc.stderr) + # + # Dunder + # + def __eq__(self, other: object) -> bool: + assert isinstance(other, Pane) + return self.pane_id == other.pane_id - self.server._update_panes() - return self + def __repr__(self) -> str: + return "{}({} {})".format(self.__class__.__name__, self.pane_id, self.window) - def enter(self) -> None: - """ - Send carriage return to pane. + # + # Aliases + # + @property + def id(self) -> t.Optional[str]: + """Alias of :attr:`Pane.pane_id` - ``$ tmux send-keys`` send Enter to the pane. - """ - self.cmd("send-keys", "Enter") + >>> pane.id + '%1' - def capture_pane(self) -> t.Union[str, t.List[str]]: + >>> pane.id == pane.pane_id + True """ - Capture text from pane. + return self.pane_id - ``$ tmux capture-pane`` to pane. + @property + def index(self) -> t.Optional[str]: + """Alias of :attr:`Pane.pane_index` - Returns - ------- - :class:`list` + >>> pane.index + '0' + + >>> pane.index == pane.pane_index + True """ - return self.cmd("capture-pane", "-p").stdout + return self.pane_index - def select_pane(self) -> "Pane": + @property + def height(self) -> t.Optional[str]: + """Alias of :attr:`Pane.pane_height` + + >>> pane.height.isdigit() + True + + >>> pane.height == pane.pane_height + True """ - Select pane. Return ``self``. + return self.pane_height - To select a window object asynchrously. If a ``pane`` object exists - and is no longer longer the current window, ``w.select_pane()`` - will make ``p`` the current pane. + @property + def width(self) -> t.Optional[str]: + """Alias of :attr:`Pane.pane_width` + + >>> pane.width.isdigit() + True - Returns - ------- - :class:`pane` + >>> pane.width == pane.pane_width + True """ - pane = self.window.select_pane(self._pane_id) - if pane is None: - raise exc.LibTmuxException(f"Pane not found: {self}") - return pane + return self.pane_width - def __repr__(self) -> str: - return "{}({} {})".format( - self.__class__.__name__, self.get("pane_id"), self.window - ) + # + # Legacy + # + def get(self, key: str, default: t.Optional[t.Any] = None) -> t.Any: + """ + .. deprecated:: 0.16 + """ + warnings.warn("Pane.get() is deprecated") + return getattr(self, key, default) + + def __getitem__(self, key: str) -> t.Any: + """ + .. deprecated:: 0.16 + """ + warnings.warn(f"Item lookups, e.g. pane['{key}'] is deprecated") + return getattr(self, key) diff --git a/src/libtmux/pytest_plugin.py b/src/libtmux/pytest_plugin.py index ce62da913..8a8124c1d 100644 --- a/src/libtmux/pytest_plugin.py +++ b/src/libtmux/pytest_plugin.py @@ -178,8 +178,8 @@ def session(request: pytest.FixtureRequest, server: Server) -> "Session": # find current sessions prefixed with tmuxp old_test_sessions = [] - for s in server._sessions: - old_name = s.get("session_name") + for s in server.sessions: + old_name = s.session_name if old_name is not None and old_name.startswith(TEST_SESSION_PREFIX): old_test_sessions.append(old_name) @@ -194,7 +194,7 @@ def session(request: pytest.FixtureRequest, server: Server) -> "Session": Make sure that tmuxp can :ref:`test_builder_visually` and switches to the newly created session for that testcase. """ - session_id = session.get("session_id") + session_id = session.session_id assert session_id is not None try: @@ -206,7 +206,7 @@ def session(request: pytest.FixtureRequest, server: Server) -> "Session": for old_test_session in old_test_sessions: logger.debug(f"Old test test session {old_test_session} found. Killing it.") server.kill_session(old_test_session) - assert TEST_SESSION_NAME == session.get("session_name") + assert TEST_SESSION_NAME == session.session_name assert TEST_SESSION_NAME != "tmuxp" return session diff --git a/src/libtmux/server.py b/src/libtmux/server.py index cfc0bda44..af3ed718a 100644 --- a/src/libtmux/server.py +++ b/src/libtmux/server.py @@ -9,16 +9,20 @@ import shutil import subprocess import typing as t +import warnings +from libtmux._internal.query_list import QueryList from libtmux.common import tmux_cmd +from libtmux.neo import fetch_objs +from libtmux.pane import Pane from libtmux.session import Session +from libtmux.window import Window from . import exc, formats from .common import ( EnvironmentMixin, PaneDict, SessionDict, - TmuxRelationalObject, WindowDict, has_gte_version, session_check_name, @@ -27,14 +31,13 @@ logger = logging.getLogger(__name__) -class Server(TmuxRelationalObject["Session", "SessionDict"], EnvironmentMixin): - +class Server(EnvironmentMixin): """ The :term:`tmux(1)` :term:`Server` [server_manual]_. - - :attr:`Server._sessions` [:class:`Session`, ...] + - :attr:`Server.sessions` [:class:`Session`, ...] - - :attr:`Session._windows` [:class:`Window`, ...] + - :attr:`Session.windows` [:class:`Window`, ...] - :attr:`Window._panes` [:class:`Pane`, ...] @@ -153,10 +156,18 @@ def raise_if_dead(self) -> None: subprocess.check_call([tmux_bin] + cmd_args) + # + # Command + # def cmd(self, *args: t.Any, **kwargs: t.Any) -> tmux_cmd: """ Execute tmux command and return output. + Examples + -------- + >>> server.cmd('display-message', 'hi') + + Returns ------- :class:`common.tmux_cmd` @@ -185,233 +196,35 @@ def cmd(self, *args: t.Any, **kwargs: t.Any) -> tmux_cmd: return tmux_cmd(*cmd_args, **kwargs) - def _list_sessions(self) -> t.List[SessionDict]: - """ - Return list of sessions in :py:obj:`dict` form. - - Retrieved from ``$ tmux(1) list-sessions`` stdout. - - The :py:obj:`list` is derived from ``stdout`` in - :class:`common.tmux_cmd` which wraps :py:class:`subprocess.Popen`. - - Returns - ------- - list of dict - """ - - sformats = formats.SESSION_FORMATS - tmux_formats = ["#{%s}" % f for f in sformats] - - tmux_args = ("-F%s" % formats.FORMAT_SEPARATOR.join(tmux_formats),) # output - - list_sessions_cmd = self.cmd("list-sessions", *tmux_args) - - 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_output = list_sessions_cmd.stdout - - # combine format keys with values returned from ``tmux list-sessions`` - sessions_formatters = [ - dict(zip(sformats, session.split(formats.FORMAT_SEPARATOR))) - for session in sessions_output - ] - - # clear up empty dict - sessions_formatters_filtered = [ - {k: v for k, v in session.items() if v} for session in sessions_formatters - ] - - return sessions_formatters_filtered - - @property - def _sessions(self) -> t.List[SessionDict]: - """Property / alias to return :meth:`~._list_sessions`.""" - - return self._list_sessions() - - def list_sessions(self) -> t.List[Session]: - """ - Return list of :class:`Session` from the ``tmux(1)`` session. - - Returns - ------- - list of :class:`Session` - """ - return [Session(server=self, **s) for s in self._sessions] - - @property - def sessions(self) -> t.List[Session]: - """Property / alias to return :meth:`~.list_sessions`.""" - try: - return self.list_sessions() - except Exception: - return [] - - #: Alias :attr:`sessions` for :class:`~libtmux.common.TmuxRelationalObject` - children = sessions # type: ignore - - def _list_windows(self) -> t.List[WindowDict]: - """ - Return list of windows in :py:obj:`dict` form. - - Retrieved from ``$ tmux(1) list-windows`` stdout. - - The :py:obj:`list` is derived from ``stdout`` in - :class:`common.tmux_cmd` which wraps :py:class:`subprocess.Popen`. - - Returns - ------- - list of dict - """ - - wformats = ["session_name", "session_id"] + formats.WINDOW_FORMATS - tmux_formats = ["#{%s}" % format for format in wformats] - - proc = self.cmd( - "list-windows", # ``tmux list-windows`` - "-a", - "-F%s" % formats.FORMAT_SEPARATOR.join(tmux_formats), # output - ) - - if proc.stderr: - raise exc.LibTmuxException(proc.stderr) - - window_output = proc.stdout - - wformats = ["session_name", "session_id"] + formats.WINDOW_FORMATS - - # combine format keys with values returned from ``tmux list-windows`` - window_formatters = [ - dict(zip(wformats, window.split(formats.FORMAT_SEPARATOR))) - for window in window_output - ] - - # clear up empty dict - 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 window_formatters_filtered: - if "window_id" not in w: - w["window_id"] = w["window_name"] - - if self._windows: - self._windows[:] = [] - - self._windows.extend(window_formatters_filtered) - - return self._windows - - def _update_windows(self) -> "Server": - """ - Update internal window data and return ``self`` for chainability. - - Returns - ------- - :class:`Server` - """ - self._list_windows() - return self - - def _list_panes(self) -> t.List[PaneDict]: - """ - Return list of panes in :py:obj:`dict` form. - - Retrieved from ``$ tmux(1) list-panes`` stdout. - - The :py:obj:`list` is derived from ``stdout`` in - :class:`util.tmux_cmd` which wraps :py:class:`subprocess.Popen`. - - Returns - ------- - list - """ - - pformats = [ - "session_name", - "session_id", - "window_index", - "window_id", - "window_name", - ] + formats.PANE_FORMATS - tmux_formats = [("#{%%s}%s" % formats.FORMAT_SEPARATOR) % f for f in pformats] - - proc = self.cmd("list-panes", "-a", "-F%s" % "".join(tmux_formats)) # output - - if proc.stderr: - raise exc.LibTmuxException(proc.stderr) - - pane_output = proc.stdout - - pformats = [ - "session_name", - "session_id", - "window_index", - "window_id", - "window_name", - ] + formats.PANE_FORMATS - - # combine format keys with values returned from ``tmux list-panes`` - pane_formatters = [ - dict(zip(pformats, formatter.split(formats.FORMAT_SEPARATOR))) - for formatter in pane_output - ] - - # Filter empty values - pane_formatters_filtered = [ - { - 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 formatter in pane_formatters - ] - - if self._panes: - self._panes[:] = [] - - self._panes.extend(pane_formatters_filtered) - - return self._panes - - def _update_panes(self) -> "Server": - """ - Update internal pane data and return ``self`` for chainability. - - Returns - ------- - :class:`Server` - """ - self._list_panes() - return self - @property def attached_sessions(self) -> t.List[Session]: """ Return active :class:`Session` objects. + Examples + -------- + >>> server.attached_sessions + [] + Returns ------- list of :class:`Session` """ - try: - sessions = self._sessions + sessions = self.sessions attached_sessions = list() for session in sessions: - attached = session.get("session_attached") + attached = session.session_attached # for now session_active is a unicode if attached != "0": - logger.debug(f"session {session.get('name')} attached") + logger.debug(f"session {session.name} attached") attached_sessions.append(session) else: continue - return [Session(server=self, **s) for s in attached_sessions] or [] + return attached_sessions + # return [Session(**s) for s in attached_sessions] or None except Exception: return [] @@ -610,9 +423,6 @@ def new_session( logger.debug(f"creating session {session_name}") - sformats = formats.SESSION_FORMATS - tmux_formats = ["#{%s}" % f for f in sformats] - env = os.environ.get("TMUX") if env: @@ -620,7 +430,7 @@ def new_session( tmux_args: t.Tuple[t.Union[str, int], ...] = ( "-P", - "-F%s" % formats.FORMAT_SEPARATOR.join(tmux_formats), # output + "-F#{session_id}", # output ) if session_name is not None: @@ -653,14 +463,172 @@ def new_session( if env: os.environ["TMUX"] = env - # Combine format keys with values returned from ``tmux list-windows`` - session_params = dict( - zip(sformats, session_stdout.split(formats.FORMAT_SEPARATOR)) + session_formatters = dict( + zip(["session_id"], session_stdout.split(formats.FORMAT_SEPARATOR)) + ) + + return Session.from_session_id( + server=self, session_id=session_formatters["session_id"] ) - # Filter empty values - session_params = {k: v for k, v in session_params.items() if v} + # + # Relations + # + @property + def sessions(self) -> QueryList[Session]: # type:ignore + """Sessions belonging server. + + Can be accessed via + :meth:`.sessions.get() ` and + :meth:`.sessions.filter() ` + """ + sessions: t.List["Session"] = [] + + try: + for obj in fetch_objs( + list_cmd="list-sessions", + server=self, + ): + sessions.append(Session(server=self, **obj)) + except Exception: + pass - session = Session(server=self, **session_params) + return QueryList(sessions) - return session + @property + def windows(self) -> QueryList[Window]: # type:ignore + """Windows belonging server. + + Can be accessed via + :meth:`.sessions.get() ` and + :meth:`.sessions.filter() ` + """ + windows: t.List["Window"] = [] + for obj in fetch_objs( + list_cmd="list-windows", + list_extra_args=("-a",), + server=self, + ): + windows.append(Window(server=self, **obj)) + + return QueryList(windows) + + @property + def panes(self) -> QueryList[Pane]: # type:ignore + """Panes belonging server. + + Can be accessed via + :meth:`.sessions.get() ` and + :meth:`.sessions.filter() ` + """ + panes: t.List["Pane"] = [] + for obj in fetch_objs( + list_cmd="list-panes", + list_extra_args=["-s"], + server=self, + ): + panes.append(Pane(server=self, **obj)) + + return QueryList(panes) + + def _list_panes(self) -> t.List[PaneDict]: + """ + Return list of panes in :py:obj:`dict` form. + + Retrieved from ``$ tmux(1) list-panes`` stdout. + + The :py:obj:`list` is derived from ``stdout`` in + :class:`util.tmux_cmd` which wraps :py:class:`subprocess.Popen`. + """ + return [p.__dict__ for p in self.panes] + + def _update_panes(self) -> "Server": + """ + Update internal pane data and return ``self`` for chainability. + + Returns + ------- + :class:`Server` + """ + self._list_panes() + return self + + # + # Legacy: Redundant stuff we want to remove + # + def get_by_id(self, id: str) -> t.Optional[Session]: + """ + .. deprecated:: 0.16 + """ + warnings.warn("Server.get_by_id() is deprecated") + return self.sessions.get(session_id=id, default=None) + + def where(self, kwargs: t.Dict[str, t.Any]) -> t.List[Session]: + """ + .. deprecated:: 0.16 + """ + warnings.warn("Server.find_where() is deprecated") + try: + return self.sessions.filter(**kwargs) + except IndexError: + return [] + + def find_where(self, kwargs: t.Dict[str, t.Any]) -> t.Optional[Session]: + """ + .. deprecated:: 0.16 + """ + warnings.warn("Server.find_where() is deprecated") + return self.sessions.get(default=None, **kwargs) + + def _list_windows(self) -> t.List[WindowDict]: + """Return list of windows in :py:obj:`dict` form. + + Retrieved from ``$ tmux(1) list-windows`` stdout. + + The :py:obj:`list` is derived from ``stdout`` in + :class:`common.tmux_cmd` which wraps :py:class:`subprocess.Popen`. + + .. deprecated:: 0.16 + """ + warnings.warn("Server._list_windows() is deprecated") + return [w.__dict__ for w in self.windows] + + def _update_windows(self) -> "Server": + """Update internal window data and return ``self`` for chainability. + + .. deprecated:: 0.16 + """ + warnings.warn("Server._update_windows() is deprecated") + self._list_windows() + return self + + @property + def _sessions(self) -> t.List[SessionDict]: + """Property / alias to return :meth:`~._list_sessions`. + + .. deprecated:: 0.16 + """ + warnings.warn("Server._sessions is deprecated") + return self._list_sessions() + + def _list_sessions(self) -> t.List["SessionDict"]: + """ + .. deprecated:: 0.16 + """ + warnings.warn("Server._list_sessions() is deprecated") + return [s.__dict__ for s in self.sessions] + + def list_sessions(self) -> t.List[Session]: + """Return list of :class:`Session` from the ``tmux(1)`` session. + + .. deprecated:: 0.16 + + Returns + ------- + list of :class:`Session` + """ + warnings.warn("Server.list_sessions is deprecated") + return self.sessions + + #: Alias :attr:`sessions` for :class:`~libtmux.common.TmuxRelationalObject` + children = sessions diff --git a/src/libtmux/session.py b/src/libtmux/session.py index 37cd8c122..b1ffabf0f 100644 --- a/src/libtmux/session.py +++ b/src/libtmux/session.py @@ -4,19 +4,22 @@ ~~~~~~~~~~~~~~~ """ +import dataclasses import logging import os import typing as t +import warnings +from libtmux._internal.query_list import QueryList from libtmux.common import tmux_cmd +from libtmux.formats import FORMAT_SEPARATOR +from libtmux.neo import Obj, fetch_obj, fetch_objs +from libtmux.pane import Pane from libtmux.window import Window -from . import exc, formats +from . import exc from .common import ( EnvironmentMixin, - SessionDict, - TmuxMappingObject, - TmuxRelationalObject, WindowDict, handle_option_error, has_gte_version, @@ -25,16 +28,48 @@ ) if t.TYPE_CHECKING: - from .pane import Pane from .server import Server logger = logging.getLogger(__name__) -class Session( - TmuxMappingObject, TmuxRelationalObject["Window", "WindowDict"], EnvironmentMixin -): +# Dataclasses define the schema in a typed annotation way. +# +# Users load through YAML, JSON, TOML, or any sort of data entry, these are parsed +# into dataclasses. Bi-directionals. +# +# This means the "intermediary language" is actually a first class, typed, declarative +# python syntax. It's so first class, it's used in tests, exposed as an experimental API +# with full intention for users to bootstrap sessions fast. +# +# Pitfalls: Lock / pin your versions. APIs may change. It's a Public API - but +# considered experimental in the sense we . +# +# The good news is, since it's typed, you will be able to refactor quickly. In fact, +# it's easier to factor these than a config where there's no safety. +# +# Session.build_workspace( +# windows=[ +# Window( +# panes=[ +# panes=[] +# ] +# +# ) +# ], +# settings=WorkspaceBuilderSettings( +# backend='libtmux.ext.workspace_builder.backends.default' +# ) +# ) +# WorkspaceBuilder: +# backend = 'libtmux.ext.workspace_builder.backends.default' +# + + +@dataclasses.dataclass() +class Session(Obj, EnvironmentMixin): + """ A :term:`tmux(1)` :term:`Session` [session_manual]_. @@ -71,40 +106,62 @@ class Session( https://man.openbsd.org/tmux.1#DESCRIPTION. Accessed April 1st, 2018. """ - child_id_attribute = "window_id" - """Unique child ID key for :class:`~libtmux.common.TmuxRelationalObject`""" - formatter_prefix = "session_" - """Namespace used for :class:`~libtmux.common.TmuxMappingObject`""" server: "Server" - """:class:`libtmux.server.Server` session is linked to""" - def __init__(self, server: "Server", session_id: str, **kwargs: t.Any) -> None: - EnvironmentMixin.__init__(self) - self.server = server + def refresh(self) -> None: + assert isinstance(self.session_id, str) + return super()._refresh( + obj_key="session_id", obj_id=self.session_id, list_cmd="list-sessions" + ) - self._session_id = session_id - self.server._update_windows() + @classmethod + def from_session_id(cls, server: "Server", session_id: str) -> "Session": + session = fetch_obj( + obj_key="session_id", + obj_id=session_id, + list_cmd="list-sessions", + server=server, + ) + return cls(server=server, **session) + # + # Relations + # @property - def _info(self) -> t.Optional[SessionDict]: # type: ignore # mypy#1362 - attrs = {"session_id": str(self._session_id)} - - def by(val: SessionDict) -> bool: - for key in attrs.keys(): - try: - if attrs[key] != val[key]: - return False - except KeyError: - return False - return True - - 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 windows(self) -> QueryList["Window"]: # type:ignore + """Windows belonging session. + + Can be accessed via + :meth:`.windows.get() ` and + :meth:`.windows.filter() ` + """ + windows: t.List["Window"] = [] + for obj in fetch_objs( + list_cmd="list-windows", + list_extra_args=["-t", str(self.session_id)], + server=self.server, + ): + if obj.get("session_id") == self.session_id: + windows.append(Window(server=self.server, **obj)) + + return QueryList(windows) + @property + def panes(self) -> QueryList["Pane"]: # type:ignore + panes: t.List["Pane"] = [] + for obj in fetch_objs( + list_cmd="list-panes", + list_extra_args=["-s", "-t", str(self.session_id)], + server=self.server, + ): + if obj.get("session_id") == self.session_id: + panes.append(Pane(server=self.server, **obj)) + + return QueryList(panes) + + # + # Command + # def cmd(self, *args: t.Any, **kwargs: t.Any) -> tmux_cmd: """ Return :meth:`server.cmd`. @@ -124,43 +181,258 @@ def cmd(self, *args: t.Any, **kwargs: t.Any) -> tmux_cmd: # insert -t immediately after 1st arg, as per tmux format new_args: t.Tuple[str, ...] = tuple() new_args += (args[0],) + assert isinstance(self.session_id, str) new_args += ( "-t", - self.id, + self.session_id, ) new_args += tuple(args[1:]) args = new_args return self.server.cmd(*args, **kwargs) - def attach_session(self) -> None: + # + # Commands (tmux-like) + # + def set_option( + self, option: str, value: t.Union[str, int], _global: bool = False + ) -> "Session": + """ + Set option ``$ tmux set-option