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