Add configurable REPL settings in bohort

[test-312-latest]
[pypi]

aioipfs/__init__.py

@@ -1,4 +1,4 @@
-__version__ = '0.7.0'
+__version__ = '0.7.1'
 
 from yarl import URL
 from distutils.version import StrictVersion  # type: ignore

aioipfs/scripts/bohort/__init__.py

@@ -6,6 +6,7 @@
 import inspect
 
 from dataclasses import dataclass
+from importlib import resources
 from pathlib import Path
 from typing import Union, List, Dict
 from ptpython import embed
@@ -16,12 +17,13 @@
 import appdirs  # type: ignore
 
 from omegaconf import OmegaConf
-from omegaconf import DictConfig
+from omegaconf import DictConfig, ListConfig
 
 from .cli import configure
+from .cli import pf_text
 
 
-__version__ = '0.2.0'
+__version__ = '0.2.1'
 
 
 @dataclass
@@ -30,8 +32,8 @@ class Context:
     interactive: bool = False
 
 
-def rpc_method_config(config: DictConfig, method: str) -> Union[DictConfig,
-                                                                None]:
+def rpc_method_config(config: DictConfig,
+                      method: str) -> Union[DictConfig, None]:
     rpc = config.get('rpc_methods')
 
     if isinstance(rpc, DictConfig):
@@ -42,35 +44,49 @@ def rpc_method_config(config: DictConfig, method: str) -> Union[DictConfig,
 
 async def _cmd_wrapper(config: DictConfig,
                        ctx: Context, method: str, *args, **kwargs):
+    timeout: float = 0
     try:
         meth = operator.attrgetter(method)(ctx.client)
         assert meth
 
         rpc_cfg = rpc_method_config(config, method)
-        if rpc_cfg and 'defaults' in rpc_cfg:
-            defaults = OmegaConf.to_container(rpc_cfg.defaults)  # type: ignore
-
-            if isinstance(defaults, dict):
-                for key, value in defaults.items():
-                    if key not in kwargs and isinstance(value,
-                                                        (int, float, str)):
-                        kwargs[key] = value  # type: ignore
-
-        if inspect.isasyncgenfunction(meth):
-            # async generator
-            _entries: List = []
-
-            async for entry in meth(*args, **kwargs):
-                _entries.append(entry)
-
-            return _entries
-        else:
-            # coroutine
-            return await meth(*args, **kwargs)
+        if rpc_cfg:
+            timeout = rpc_cfg.get('timeout', 0)
+            assert isinstance(timeout, (int, float)), \
+                f"Invalid timeout config for method: {method}"
+
+            if 'defaults' in rpc_cfg:
+                defaults = OmegaConf.to_container(
+                    rpc_cfg.defaults)  # type: ignore
+
+                if isinstance(defaults, dict):
+                    for key, value in defaults.items():
+                        if key not in kwargs and isinstance(value,
+                                                            (int, float, str)):
+                            kwargs[key] = value  # type: ignore
+
+        async with ctx.client.timeout(
+                timeout if timeout > 0 else None):  # type: ignore
+            if inspect.isasyncgenfunction(meth):
+                # async generator
+                _entries: List = []
+
+                async for entry in meth(*args, **kwargs):
+                    _entries.append(entry)
+
+                return _entries
+            else:
+                # coroutine
+
+                return await meth(*args, **kwargs)
     except aioipfs.RPCAccessDenied:
         print('Access denied for this RPC endpoint! Check your credentials.')
     except (aioipfs.APIError, aioipfs.UnknownAPIError) as aerr:
         print(f'API error {aerr.code}: {aerr.message}')
+    except asyncio.TimeoutError:
+        pf_text(f'Timeout for method: {method} ({timeout} secs)')
+    except asyncio.CancelledError as err:
+        pf_text(f'Method {method} cancelled: {err}')
     except AttributeError:
         print(f'No such client method: {method}')
     except BaseException:
@@ -90,19 +106,26 @@ def get_auth_helper(creds: str) -> Union[aioipfs.BasicAuth,
     raise ValueError(f'Invalid RPC credentials value: {creds}')
 
 
+def save_config(cfg_path: Path, config: Union[DictConfig, ListConfig]):
+    with open(cfg_path, 'wt') as f:
+        OmegaConf.save(config, f)
+
+
 async def start(args, cfg_dir: Path, data_dir: Path) -> None:
     cfg_path = cfg_dir.joinpath('bohort.yaml')
 
     if not cfg_path.exists():
-        with open(cfg_path, 'wt') as f:
-            OmegaConf.save(OmegaConf.create({
-                'nodes': {},
-                'rpc_methods': {}
-            }), f)
+        cfg_path.touch()
 
     with open(cfg_path, 'rt') as f:
         cfg = OmegaConf.load(f)
 
+    with resources.files(__name__).joinpath(
+            'default_config.yaml').open('r') as f:
+        cfg = OmegaConf.merge(OmegaConf.load(f), cfg)
+
+        save_config(cfg_path, cfg)
+
     if args.save_node:
         assert re.match(r'^[\w_-]+$', args.save_node), \
             "Invalid node name format"
@@ -122,8 +145,7 @@ async def start(args, cfg_dir: Path, data_dir: Path) -> None:
         })
         cfg = OmegaConf.merge(ncfg, cfg)
 
-        with open(cfg_path, 'wt') as f:
-            OmegaConf.save(cfg, f)
+        save_config(cfg_path, cfg)
 
     if args.node:
         node, credid = tuple(args.node.split(
@@ -201,7 +223,7 @@ async def start(args, cfg_dir: Path, data_dir: Path) -> None:
                 locals=clocals,
                 return_asyncio_coroutine=True,
                 patch_stdout=True,
-                configure=configure,
+                configure=functools.partial(configure, cfg),
                 history_filename=args.history_path if
                 not args.no_history else None
             )  # type: ignore

aioipfs/scripts/bohort/cli.py

@@ -1,34 +1,48 @@
 from ptpython.prompt_style import PromptStyle
 from ptpython.layout import CompletionVisualisation
-from prompt_toolkit.formatted_text import HTML
+from prompt_toolkit import print_formatted_text, HTML
+from omegaconf import DictConfig
 
 
-def configure(repl):
+def configure(config: DictConfig, repl) -> None:
     class CustomPrompt(PromptStyle):
         def in_prompt(self):
-            return HTML("<ansigreen>bohort [%s]</ansigreen>: ") % (
-                repl.current_statement_index
-            )
+            return HTML("<{ipcolor}>bohort [{idx}]</{ipcolor}>: ".format(
+                ipcolor=config.repl.input_prompt_color,
+                idx=repl.current_statement_index
+            ))
 
         def in2_prompt(self, width):
             return "...: ".rjust(width)
 
         def out_prompt(self):
-            return HTML("<ansicyan>Result[%s]</ansicyan>: ") % (
-                repl.current_statement_index
-            )
+            return HTML("<{opcolor}>Result[{idx}]</{opcolor}>: ".format(
+                opcolor=config.repl.output_prompt_color,
+                idx=repl.current_statement_index
+            ))
 
     repl.all_prompt_styles["custom"] = CustomPrompt()
-    repl.cursor_shape_config = "Blink block"
+    repl.cursor_shape_config = config.repl.cursor_shape
     repl.insert_blank_line_after_output = True
-    repl.use_code_colorscheme('native')
     repl.prompt_style = "custom"
-    repl.show_signature = True
-    repl.enable_history_search = True
-    repl.enable_auto_suggest = True
+    repl.show_signature = config.repl.show_signature
+    repl.enable_history_search = config.repl.enable_history_search
+    repl.enable_auto_suggest = config.repl.enable_auto_suggest
     repl.title = 'bohort'
-    repl.confirm_exit = False
+    repl.confirm_exit = config.repl.confirm_exit
     repl.vi_keep_last_used_mode = True
-    repl.completion_visualisation = CompletionVisualisation.POP_UP
     repl.completion_menu_scroll_offset = 0
-    repl.complete_while_typing = False
+    repl.complete_while_typing = config.repl.complete_while_typing
+    repl.enable_input_validation = True
+
+    cvisual = config.repl.get('completion_visualisation')
+
+    if cvisual in ['TOOLBAR', 'POP_UP', 'MULTI_COLUMN']:
+        repl.completion_visualisation = getattr(CompletionVisualisation,
+                                                cvisual)
+
+    repl.use_code_colorscheme(config.repl.color_scheme)
+
+
+def pf_text(text: str, color: str = 'ansiyellow') -> None:
+    print_formatted_text(HTML(f'<{color}>{text}</{color}>'))

aioipfs/scripts/bohort/default_config.yaml

@@ -0,0 +1,14 @@
+nodes: {}
+rpc_methods: {}
+
+repl:
+  cursor_shape: 'Blink block'
+  input_prompt_color: 'ansigreen'
+  output_prompt_color: 'ansiyellow'
+  completion_visualisation: POP_UP
+  color_scheme: default
+  show_signature: true
+  enable_history_search: true
+  enable_auto_suggest: true
+  complete_while_typing: false
+  confirm_exit: false

docs/bohort.rst

@@ -90,6 +90,9 @@ The configuration file location (on Posix platforms) is: **~/.config/aioipfs/boh
 RPC methods
 -----------
 
+RPC params
+^^^^^^^^^^
+
 You can set the default params that will be passed to specific RPC methods
 by defining the default coroutine keyword arguments for each method:
 
@@ -111,6 +114,41 @@ by defining the default coroutine keyword arguments for each method:
 If you pass a parameter for which you've set a default in the config, the default
 value won't be used.
 
+Timeout
+^^^^^^^
+
+You can set a timeout (in seconds) for each RPC method:
+
+.. code-block:: yaml
+
+    rpc_methods:
+      core.ls:
+        timeout: 60
+
+REPL settings
+^^^^^^^^^^^^^
+
+.. code-block:: yaml
+
+    repl:
+      cursor_shape: Blink block
+      input_prompt_color: ansigreen
+      output_prompt_color: ansiyellow
+
+      # Possible values: POP_UP, MULTI_COLUMN, TOOLBAR or NONE
+      completion_visualisation: POP_UP
+
+      color_scheme: default
+      show_signature: true
+      enable_history_search: true
+      enable_auto_suggest: true
+      complete_while_typing: false
+      confirm_exit: false
+
+Check out ptpython's
+`config.py example <https://github.com/prompt-toolkit/ptpython/blob/master/examples/ptpython_config/config.py>`_ for a description of all the settings.
+
+
 REPL toolkit documentation
 --------------------------
 

pyproject.toml

@@ -77,5 +77,8 @@ include-package-data = false
 exclude = ["tests"]
 namespaces = false
 
+[tool.setuptools.package-data]
+"aioipfs.scripts.bohort" = ["*.yaml"]
+
 [project.scripts]
 bohort = "aioipfs.scripts.bohort:run_bohort"