meerschaum.actions

  1#! /usr/bin/env python
  2# -*- coding: utf-8 -*-
  3# vim:fenc=utf-8
  4
  5"""
  6Default actions available to the mrsm CLI.
  7"""
  8
  9from __future__ import annotations
 10from meerschaum.utils.typing import Callable, Any, Optional, Union, List, Dict, SuccessTuple
 11from meerschaum.utils.packages import get_modules_from_package
 12_custom_actions = []
 13
 14def get_subactions(
 15        action: Union[str, List[str]],
 16        _actions: Optional[Dict[str, Callable[[Any], Any]]] = None,
 17    ) -> Dict[str, Callable[[Any], Any]]:
 18    """
 19    Return a dictionary of an action's sub-action functions.
 20
 21    Examples
 22    --------
 23    >>> get_subactions('install').keys()
 24    dict_keys(['packages', 'plugins', 'required'])
 25    >>> 
 26    """
 27    import importlib, inspect
 28    subactions = {}
 29    if isinstance(action, str):
 30        action = [action]
 31    action_function = get_action(action[0], _actions=_actions)
 32    if action_function is None:
 33        return subactions
 34    try:
 35        action_module = importlib.import_module(action_function.__module__)
 36    except ImportError:
 37        action_module = None
 38    if action_module is None:
 39        return subactions
 40    for name, f in inspect.getmembers(action_module):
 41        if not inspect.isfunction(f):
 42            continue
 43        if action_function.__name__ + '_' in name and not name.lstrip('_').startswith('complete'):
 44            _name = name.replace(action_function.__name__, '')
 45            _name = _name.lstrip('_')
 46            subactions[_name] = f
 47    return subactions
 48
 49
 50def get_action(
 51        action: Union[str, List[str]],
 52        _actions: Optional[Dict[str, Callable[[Any], Any]]] = None,
 53    ) -> Union[Callable[[Any], Any], None]:
 54    """
 55    Return a function corresponding to the given action list.
 56    This may be a custom action with an underscore, in which case, allow for underscores.
 57    This may also be a subactions, which is handled by `get_subactions()`
 58    """
 59    if _actions is None:
 60        _actions = actions
 61    if isinstance(action, str):
 62        action = [action]
 63
 64    if not any(action):
 65        return None
 66
 67    ### Simple case, e.g. ['show']
 68    if len(action) == 1:
 69        if action[0] in _actions:
 70            return _actions[action[0]]
 71
 72        ### e.g. ['foo'] (and no custom action available)
 73        return None
 74
 75    ### Last case: it could be a custom action with an underscore in the name.
 76    action_name_with_underscores = '_'.join(action)
 77    candidates = []
 78    for action_key, action_function in _actions.items():
 79        if not '_' in action_key:
 80            continue
 81        if action_name_with_underscores.startswith(action_key):
 82            leftovers = action_name_with_underscores.replace(action_key, '')
 83            candidates.append((len(leftovers), action_function))
 84    if len(candidates) > 0:
 85        return sorted(candidates)[0][1]
 86
 87    ### Might be dealing with a subaction.
 88    if action[0] in _actions:
 89        subactions = get_subactions([action[0]], _actions=_actions)
 90        if action[1] not in subactions:
 91            return _actions[action[0]]
 92        return subactions[action[1]]
 93
 94    return None
 95
 96
 97def get_main_action_name(
 98        action: Union[List[str], str],
 99        _actions: Optional[Dict[str, Callable[[Any], SuccessTuple]]] = None,
100    ) -> Union[str, None]:
101    """
102    Given an action list, return the name of the main function.
103    For subactions, this will return the root function.
104    If no action function can be found, return `None`.
105
106    Parameters
107    ----------
108    action: Union[List[str], str]
109        The standard action list.
110
111    Examples
112    --------
113    >>> get_main_action_name(['show', 'pipes'])
114    'show'
115    >>> 
116    """
117    if _actions is None:
118        _actions = actions
119    action_function = get_action(action, _actions=_actions)
120    if action_function is None:
121        return None
122    clean_name = action_function.__name__.lstrip('_')
123    if clean_name in _actions:
124        return clean_name
125    words = clean_name.split('_')
126    first_word = words[0]
127    if first_word in _actions:
128        return first_word
129    ### We might be dealing with shell `do_` functions.
130    second_word = words[1] if len(words) > 1 else None
131    if second_word and second_word in _actions:
132        return second_word
133    return None
134
135
136def get_completer(
137        action: Union[List[str], str],
138        _actions: Optional[Dict[str, Callable[[Any], SuccessTuple]]] = None,
139    ) -> Union[
140        Callable[['meerschaum._internal.shell.Shell', str, str, int, int], List[str]], None
141    ]:
142    """Search for a custom completer function for an action."""
143    import importlib, inspect
144    if isinstance(action, str):
145        action = [action]
146    action_function = get_action(action)
147    if action_function is None:
148        return None
149    try:
150        action_module = importlib.import_module(action_function.__module__)
151    except ImportError:
152        action_module = None
153    if action_module is None:
154        return None
155    candidates = []
156    for name, f in inspect.getmembers(action_module):
157        if not inspect.isfunction(f):
158            continue
159        name_lstrip = name.lstrip('_')
160        if not name_lstrip.startswith('complete_'):
161            continue
162        if name_lstrip == 'complete_' + action_function.__name__:
163            candidates.append((name_lstrip, f))
164            continue
165        if name_lstrip == 'complete_' + action_function.__name__.split('_')[0]:
166            candidates.append((name_lstrip, f))
167    if len(candidates) == 1:
168        return candidates[0][1]
169    if len(candidates) > 1:
170        return sorted(candidates)[-1][1]
171    
172    return None
173
174
175def choose_subaction(
176        action: Optional[List[str]] = None,
177        options: Optional[Dict[str, Any]] = None,
178        **kw
179    ) -> SuccessTuple:
180    """
181    Given a dictionary of options and the standard Meerschaum actions list,
182    check if choice is valid and execute chosen function, else show available
183    options and return False
184
185    Parameters
186    ----------
187    action: Optional[List[str]], default None
188        A list of subactions (e.g. `show pipes` -> ['pipes']).
189
190    options: Optional[Dict[str, Any]], default None
191        Available options to execute.
192        option (key) -> function (value)
193        Functions must accept **kw keyword arguments
194        and return a tuple of success bool and message.
195
196    Returns
197    -------
198    The return value of the chosen subaction (assumed to be a `SuccessTuple`).
199
200    """
201    from meerschaum.utils.warnings import warn, info
202    import inspect
203    if action is None:
204        action = []
205    if options is None:
206        options = {}
207    parent_action = inspect.stack()[1][3]
208    if len(action) == 0:
209        action = ['']
210    choice = action[0]
211
212    def valid_choice(_choice : str, _options : dict):
213        if _choice in _options:
214            return _choice
215        if (_choice + 's') in options:
216            return _choice + 's'
217        return None
218
219    parsed_choice = valid_choice(choice, options)
220    if parsed_choice is None:
221        warn(f"Cannot {parent_action} '{choice}'. Choose one:", stack=False)
222        for option in sorted(options):
223            print(f"  - {parent_action} {option}")
224        return (False, f"Invalid choice '{choice}'.")
225    ### remove parent sub-action
226    kw['action'] = list(action)
227    del kw['action'][0]
228    return options[parsed_choice](**kw)
229
230
231def _get_subaction_names(action: str, globs: dict = None) -> List[str]:
232    """Use `meerschaum.actions.get_subactions()` instead."""
233    if globs is None:
234        import importlib
235        module = importlib.import_module(f'meerschaum.actions.{action}')
236        globs = vars(module)
237    subactions = []
238    for item in globs:
239        if f'_{action}' in item and 'complete' not in item.lstrip('_'):
240            subactions.append(globs[item])
241    return subactions
242
243
244def choices_docstring(action: str, globs : Optional[Dict[str, Any]] = None) -> str:
245    """
246    Append the an action's available options to the module docstring.
247    This function is to be placed at the bottom of each action module.
248
249    Parameters
250    ----------
251    action: str
252        The name of the action module (e.g. 'install').
253        
254    globs: Optional[Dict[str, Any]], default None
255        An optional dictionary of global variables.
256
257    Returns
258    -------
259    The generated docstring for the module.
260
261    Examples
262    --------
263    >>> from meerschaum.utils.misc import choices_docstring as _choices_docstring
264    >>> install.__doc__ += _choices_docstring('install')
265
266    """
267    options_str = f"\n    Options:\n        `{action} "
268    subactions = _get_subaction_names(action, globs=globs)
269    options_str += "["
270    sa_names = []
271    for sa in subactions:
272        try:
273            sa_names.append(sa.__name__[len(f"_{action}") + 1:])
274        except Exception as e:
275            print(e)
276            return ""
277    for sa_name in sorted(sa_names):
278        options_str += f"{sa_name}, "
279    options_str = options_str[:-2] + "]`"
280    return options_str
281
282### build __all__ from other .py files in this package
283import sys
284modules = get_modules_from_package(
285    sys.modules[__name__],
286    names = False,
287)
288__all__ = ['actions', 'get_subactions', 'get_action', 'get_main_action_name', 'get_completer']
289
290### Build the actions dictionary by importing all
291### functions that do not begin with '_' from all submodules.
292from inspect import getmembers, isfunction
293actions = {}
294
295for module in modules:
296    ### A couple important things happening here:
297    ### 1. Find all functions in all modules in `actions` package
298    ###     (skip functions that begin with '_')
299    ### 2. Add them as members to the Shell class
300    ###     - Original definition : meerschaum._internal.shell.Shell
301    ###     - New definition      : meerschaum._internal.Shell
302    ### 3. Populate the actions dictionary with function names and functions
303    ###
304    ### UPDATE:
305    ### Shell modifications have been deferred to get_shell in order to improve lazy loading.
306
307    actions.update(
308        dict(
309            [
310                ### __name__ and new function pointer
311                (ob[0], ob[1])
312                    for ob in getmembers(module)
313                        if isfunction(ob[1])
314                            ### check that the function belongs to the module
315                            and ob[0] == module.__name__.replace('_', '').split('.')[-1]
316                            ### skip functions that start with '_'
317                            and ob[0][0] != '_'
318            ]
319        )
320    )
321
322original_actions = actions.copy()
323from meerschaum._internal.entry import entry, get_shell
324import meerschaum.plugins
325make_action = meerschaum.plugins.make_action
326pre_sync_hook = meerschaum.plugins.pre_sync_hook
327post_sync_hook = meerschaum.plugins.post_sync_hook
328
329### Instruct pdoc to skip the `meerschaum.actions.plugins` subdirectory.
330__pdoc__ = {
331    'plugins': False,
332    'arguments': True,
333    'shell': False,
334    'actions': (
335        "Access functions of the standard Meerschaum actions.\n\n"
336        + "Visit the [actions reference page](https://meerschaum.io/reference/actions/) "
337        + "for documentation about each action.\n\n"
338        + """\n\nExamples
339--------
340>>> actions['show'](['pipes'])
341
342"""
343        + "Keys\n----\n"
344        + ', '.join(sorted([f'`{a}`' for a in actions]))
345    )
346}
347for a in actions:
348    __pdoc__[a] = False
349meerschaum.plugins.load_plugins()