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