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