meerschaum.plugins

Expose plugin management APIs from the meerschaum.plugins module.

View Source

  1#! /usr/bin/env python
  2# -*- coding: utf-8 -*-
  3# vim:fenc=utf-8
  4
  5"""
  6Expose plugin management APIs from the `meerschaum.plugins` module.
  7"""
  8
  9from __future__ import annotations
 10from meerschaum.utils.typing import Callable, Any, Union, Optional, Dict, List, Tuple
 11from meerschaum.utils.threading import Lock, RLock
 12from meerschaum.plugins._Plugin import Plugin
 13
 14_api_plugins: Dict[str, List[Callable[['fastapi.App'], Any]]] = {}
 15_pre_sync_hooks: Dict[str, List[Callable[[Any], Any]]] = {}
 16_post_sync_hooks: Dict[str, List[Callable[[Any], Any]]] = {}
 17_locks = {
 18    '_api_plugins': RLock(),
 19    '_pre_sync_hooks': RLock(),
 20    '_post_sync_hooks': RLock(),
 21    '__path__': RLock(),
 22    'sys.path': RLock(),
 23    'internal_plugins': RLock(),
 24    '_synced_symlinks': RLock(),
 25    'PLUGINS_INTERNAL_LOCK_PATH': RLock(),
 26}
 27__all__ = (
 28    "Plugin", "make_action", "api_plugin", "import_plugins",
 29    "reload_plugins", "get_plugins", "get_data_plugins", "add_plugin_argument",
 30    "pre_sync_hook", "post_sync_hook",
 31)
 32__pdoc__ = {
 33    'venvs': False, 'data': False, 'stack': False, 'plugins': False,
 34}
 35
 36def make_action(
 37        function: Callable[[Any], Any],
 38        shell: bool = False,
 39        activate: bool = True,
 40        deactivate: bool = True,
 41        debug: bool = False
 42    ) -> Callable[[Any], Any]:
 43    """
 44    Make a function a Meerschaum action. Useful for plugins that are adding multiple actions.
 45    
 46    Parameters
 47    ----------
 48    function: Callable[[Any], Any]
 49        The function to become a Meerschaum action. Must accept all keyword arguments.
 50        
 51    shell: bool, default False
 52        Not used.
 53        
 54    Returns
 55    -------
 56    Another function (this is a decorator function).
 57
 58    Examples
 59    --------
 60    >>> from meerschaum.plugins import make_action
 61    >>>
 62    >>> @make_action
 63    ... def my_action(**kw):
 64    ...     print('foo')
 65    ...     return True, "Success"
 66    >>>
 67    """
 68
 69    from meerschaum.actions import actions
 70    from meerschaum.utils.formatting import pprint
 71    package_name = function.__globals__['__name__']
 72    plugin_name = (
 73        package_name.split('.')[1]
 74        if package_name.startswith('plugins.') else None
 75    )
 76    plugin = Plugin(plugin_name) if plugin_name else None
 77
 78    if debug:
 79        from meerschaum.utils.debug import dprint
 80        dprint(
 81            f"Adding action '{function.__name__}' from plugin " +
 82            f"'{plugin}'..."
 83        )
 84
 85    actions[function.__name__] = function
 86    return function
 87
 88
 89def pre_sync_hook(
 90        function: Callable[[Any], Any],
 91    ) -> Callable[[Any], Any]:
 92    """
 93    Register a function as a sync hook to be executed right before sync.
 94    
 95    Parameters
 96    ----------
 97    function: Callable[[Any], Any]
 98        The function to execute right before a sync.
 99        
100    Returns
101    -------
102    Another function (this is a decorator function).
103
104    Examples
105    --------
106    >>> from meerschaum.plugins import pre_sync_hook
107    >>>
108    >>> @pre_sync_hook
109    ... def log_sync(pipe, **kwargs):
110    ...     print(f"About to sync {pipe} with kwargs:\n{kwargs}.")
111    >>>
112    """
113    with _locks['_pre_sync_hooks']:
114        try:
115            if function.__module__ not in _pre_sync_hooks:
116                _pre_sync_hooks[function.__module__] = []
117            _pre_sync_hooks[function.__module__].append(function)
118        except Exception as e:
119            from meerschaum.utils.warnings import warn
120            warn(e)
121    return function
122
123
124def post_sync_hook(
125        function: Callable[[Any], Any],
126    ) -> Callable[[Any], Any]:
127    """
128    Register a function as a sync hook to be executed upon completion of a sync.
129    
130    Parameters
131    ----------
132    function: Callable[[Any], Any]
133        The function to execute upon completion of a sync.
134        
135    Returns
136    -------
137    Another function (this is a decorator function).
138
139    Examples
140    --------
141    >>> from meerschaum.plugins import post_sync_hook
142    >>>
143    >>> @post_sync_hook
144    ... def log_sync(pipe, success_tuple, duration=None, **kwargs):
145    ...     print(f"It took {round(duration, 2)} seconds to sync {pipe}.")
146    >>>
147    """
148    with _locks['_post_sync_hooks']:
149        try:
150            if function.__module__ not in _post_sync_hooks:
151                _post_sync_hooks[function.__module__] = []
152            _post_sync_hooks[function.__module__].append(function)
153        except Exception as e:
154            from meerschaum.utils.warnings import warn
155            warn(e)
156    return function
157
158
159def api_plugin(function: Callable[[Any], Any]) -> Callable[[Any], Any]:
160    """
161    Execute the function when initializing the Meerschaum API module.
162    Useful for lazy-loading heavy plugins only when the API is started,
163    such as when editing the `meerschaum.api.app` FastAPI app.
164    
165    The FastAPI app will be passed as the only parameter.
166    
167    Parameters
168    ----------
169    function: Callable[[Any, Any]]
170        The function to be called before starting the Meerschaum API.
171        
172    Returns
173    -------
174    Another function (decorator function).
175
176    Examples
177    --------
178    >>> from meerschaum.plugins import api_plugin
179    >>>
180    >>> @api_plugin
181    >>> def initialize_plugin(app):
182    ...     @app.get('/my/new/path')
183    ...     def new_path():
184    ...         return {'message': 'It works!'}
185    >>>
186    """
187    with _locks['_api_plugins']:
188        try:
189            if function.__module__ not in _api_plugins:
190                _api_plugins[function.__module__] = []
191            _api_plugins[function.__module__].append(function)
192        except Exception as e:
193            from meerschaum.utils.warnings import warn
194            warn(e)
195    return function
196
197
198_synced_symlinks: bool = False
199def sync_plugins_symlinks(debug: bool = False, warn: bool = True) -> None:
200    """
201    Update the plugins 
202    """
203    global _synced_symlinks
204    with _locks['_synced_symlinks']:
205        if _synced_symlinks:
206            return
207
208    import sys, os, pathlib, time
209    from collections import defaultdict
210    import importlib.util
211    from meerschaum.utils.misc import flatten_list, make_symlink, is_symlink
212    from meerschaum.utils.warnings import error, warn as _warn
213    from meerschaum.config.static import STATIC_CONFIG
214    from meerschaum.utils.venv import Venv, activate_venv, deactivate_venv, is_venv_active
215    from meerschaum.config._paths import (
216        PLUGINS_RESOURCES_PATH,
217        PLUGINS_ARCHIVES_RESOURCES_PATH,
218        PLUGINS_INIT_PATH,
219        PLUGINS_DIR_PATHS,
220        PLUGINS_INTERNAL_LOCK_PATH,
221    )
222
223    ### If the lock file exists, sleep for up to a second or until it's removed before continuing.
224    with _locks['PLUGINS_INTERNAL_LOCK_PATH']:
225        if PLUGINS_INTERNAL_LOCK_PATH.exists():
226            lock_sleep_total     = STATIC_CONFIG['plugins']['lock_sleep_total']
227            lock_sleep_increment = STATIC_CONFIG['plugins']['lock_sleep_increment']
228            lock_start = time.perf_counter()
229            while (
230                (time.perf_counter() - lock_start) < lock_sleep_total
231            ):
232                time.sleep(lock_sleep_increment)
233                if not PLUGINS_INTERNAL_LOCK_PATH.exists():
234                    break
235                try:
236                    PLUGINS_INTERNAL_LOCK_PATH.unlink()
237                except Exception as e:
238                    if warn:
239                        _warn(f"Error while removing lockfile {PLUGINS_INTERNAL_LOCK_PATH}:\n{e}")
240                    break
241
242        ### Begin locking from other processes.
243        try:
244            PLUGINS_INTERNAL_LOCK_PATH.touch()
245        except Exception as e:
246            if warn:
247                _warn(f"Unable to create lockfile {PLUGINS_INTERNAL_LOCK_PATH}:\n{e}")
248
249    with _locks['internal_plugins']:
250        if is_symlink(PLUGINS_RESOURCES_PATH) or not PLUGINS_RESOURCES_PATH.exists():
251            try:
252                PLUGINS_RESOURCES_PATH.unlink()
253            except Exception as e:
254                pass
255
256        PLUGINS_RESOURCES_PATH.mkdir(exist_ok=True)
257
258
259        existing_symlinked_paths = [
260            (PLUGINS_RESOURCES_PATH / item) 
261            for item in os.listdir(PLUGINS_RESOURCES_PATH)
262        ]
263        for plugins_path in PLUGINS_DIR_PATHS:
264            if not plugins_path.exists():
265                plugins_path.mkdir(exist_ok=True, parents=True)
266        plugins_to_be_symlinked = list(flatten_list(
267            [
268                [
269                    (plugins_path / item)
270                    for item in os.listdir(plugins_path)
271                    if (
272                        not item.startswith('.')
273                    ) and (item not in ('__pycache__', '__init__.py'))
274                ]
275                for plugins_path in PLUGINS_DIR_PATHS
276            ]
277        ))
278
279        ### Check for duplicates.
280        seen_plugins = defaultdict(lambda: 0)
281        for plugin_path in plugins_to_be_symlinked:
282            plugin_name = plugin_path.stem
283            seen_plugins[plugin_name] += 1
284        for plugin_name, plugin_count in seen_plugins.items():
285            if plugin_count > 1:
286                if warn:
287                    _warn(f"Found duplicate plugins named '{plugin_name}'.")
288
289        for plugin_symlink_path in existing_symlinked_paths:
290            real_path = pathlib.Path(os.path.realpath(plugin_symlink_path))
291
292            ### Remove invalid symlinks.
293            if real_path not in plugins_to_be_symlinked:
294                if not is_symlink(plugin_symlink_path):
295                    continue
296                try:
297                    plugin_symlink_path.unlink()
298                except Exception as e:
299                    pass
300
301            ### Remove valid plugins from the to-be-symlinked list.
302            else:
303                plugins_to_be_symlinked.remove(real_path)
304
305        for plugin_path in plugins_to_be_symlinked:
306            plugin_symlink_path = PLUGINS_RESOURCES_PATH / plugin_path.name
307            try:
308                ### There might be duplicate folders (e.g. __pycache__).
309                if (
310                    plugin_symlink_path.exists()
311                    and
312                    plugin_symlink_path.is_dir()
313                    and
314                    not is_symlink(plugin_symlink_path)
315                ):
316                    continue
317                success, msg = make_symlink(plugin_path, plugin_symlink_path)
318            except Exception as e:
319                success, msg = False, str(e)
320            if not success:
321                if warn:
322                    _warn(
323                        f"Failed to create symlink {plugin_symlink_path} "
324                        + f"to {plugin_path}:\n    {msg}"
325                    )
326
327    ### Release symlink lock file in case other processes need it.
328    with _locks['PLUGINS_INTERNAL_LOCK_PATH']:
329        try:
330            if PLUGINS_INTERNAL_LOCK_PATH.exists():
331                PLUGINS_INTERNAL_LOCK_PATH.unlink()
332        ### Sometimes competing threads will delete the lock file at the same time.
333        except FileNotFoundError:
334            pass
335        except Exception as e:
336            if warn:
337                _warn(f"Error cleaning up lockfile {PLUGINS_INTERNAL_LOCK_PATH}:\n{e}")
338
339        try:
340            if not PLUGINS_INIT_PATH.exists():
341                PLUGINS_INIT_PATH.touch()
342        except Exception as e:
343            error(f"Failed to create the file '{PLUGINS_INIT_PATH}':\n{e}")
344
345    with _locks['__path__']:
346        if str(PLUGINS_RESOURCES_PATH.parent) not in __path__:
347            __path__.append(str(PLUGINS_RESOURCES_PATH.parent))
348
349    with _locks['_synced_symlinks']:
350        _synced_symlinks = True
351
352
353def import_plugins(
354        *plugins_to_import: Union[str, List[str], None],
355        warn: bool = True,
356    ) -> Union[
357        'ModuleType', Tuple['ModuleType', None]
358    ]:
359    """
360    Import the Meerschaum plugins directory.
361
362    Parameters
363    ----------
364    plugins_to_import: Union[str, List[str], None]
365        If provided, only import the specified plugins.
366        Otherwise import the entire plugins module. May be a string, list, or `None`.
367        Defaults to `None`.
368
369    Returns
370    -------
371    A module of list of modules, depening on the number of plugins provided.
372
373    """
374    import sys
375    import os
376    import importlib
377    from meerschaum.utils.misc import flatten_list
378    from meerschaum.config._paths import PLUGINS_RESOURCES_PATH
379    from meerschaum.utils.venv import is_venv_active, activate_venv, deactivate_venv, Venv
380    from meerschaum.utils.warnings import warn as _warn
381    plugins_to_import = list(plugins_to_import)
382    with _locks['sys.path']:
383
384        ### Since plugins may depend on other plugins,
385        ### we need to activate the virtual environments for library plugins.
386        ### This logic exists in `Plugin.activate_venv()`,
387        ### but that code requires the plugin's module to already be imported.
388        ### It's not a guarantee of correct activation order,
389        ### e.g. if a library plugin pins a specific package and another 
390        plugins_names = get_plugins_names()
391        already_active_venvs = [is_venv_active(plugin_name) for plugin_name in plugins_names]
392
393        if not sys.path or sys.path[0] != str(PLUGINS_RESOURCES_PATH.parent):
394            sys.path.insert(0, str(PLUGINS_RESOURCES_PATH.parent))
395
396        if not plugins_to_import:
397            for plugin_name in plugins_names:
398                activate_venv(plugin_name)
399            try:
400                imported_plugins = importlib.import_module(PLUGINS_RESOURCES_PATH.stem)
401            except ImportError as e:
402                _warn(f"Failed to import the plugins module:\n    {e}")
403                import traceback
404                traceback.print_exc()
405                imported_plugins = None
406            for plugin_name in plugins_names:
407                if plugin_name in already_active_venvs:
408                    continue
409                deactivate_venv(plugin_name)
410
411        else:
412            imported_plugins = []
413            for plugin_name in flatten_list(plugins_to_import):
414                plugin = Plugin(plugin_name)
415                try:
416                    with Venv(plugin):
417                        imported_plugins.append(
418                            importlib.import_module(
419                                f'{PLUGINS_RESOURCES_PATH.stem}.{plugin_name}'
420                            )
421                        )
422                except Exception as e:
423                    _warn(
424                        f"Failed to import plugin '{plugin_name}':\n    "
425                        + f"{e}\n\nHere's a stacktrace:",
426                        stack = False,
427                    )
428                    from meerschaum.utils.formatting import get_console
429                    get_console().print_exception(
430                        suppress = [
431                            'meerschaum/plugins/__init__.py',
432                            importlib,
433                            importlib._bootstrap,
434                        ]
435                    )
436                    imported_plugins.append(None)
437
438        if imported_plugins is None and warn:
439            _warn(f"Failed to import plugins.", stacklevel=3)
440
441        if str(PLUGINS_RESOURCES_PATH.parent) in sys.path:
442            sys.path.remove(str(PLUGINS_RESOURCES_PATH.parent))
443
444    if isinstance(imported_plugins, list):
445        return (imported_plugins[0] if len(imported_plugins) == 1 else tuple(imported_plugins))
446    return imported_plugins
447
448
449def load_plugins(debug: bool = False, shell: bool = False) -> None:
450    """
451    Import Meerschaum plugins and update the actions dictionary.
452    """
453    from inspect import isfunction, getmembers
454    from meerschaum.actions import __all__ as _all, modules
455    from meerschaum.config._paths import PLUGINS_RESOURCES_PATH
456    from meerschaum.utils.packages import get_modules_from_package
457    if debug:
458        from meerschaum.utils.debug import dprint
459
460    _plugins_names, plugins_modules = get_modules_from_package(
461        import_plugins(),
462        names = True,
463        recursive = True,
464        modules_venvs = True
465    )
466    ### I'm appending here to keep from redefining the modules list.
467    new_modules = (
468        [
469            mod for mod in modules
470            if not mod.__name__.startswith(PLUGINS_RESOURCES_PATH.stem + '.')
471        ]
472        + plugins_modules
473    )
474    n_mods = len(modules)
475    for mod in new_modules:
476        modules.append(mod)
477    for i in range(n_mods):
478        modules.pop(0)
479
480    for module in plugins_modules:
481        for name, func in getmembers(module):
482            if not isfunction(func):
483                continue
484            if name == module.__name__.split('.')[-1]:
485                make_action(func, **{'shell': shell, 'debug': debug})
486
487
488def reload_plugins(plugins: Optional[List[str]] = None, debug: bool = False) -> None:
489    """
490    Reload plugins back into memory.
491
492    Parameters
493    ----------
494    plugins: Optional[List[str]], default None
495        The plugins to reload. `None` will reload all plugins.
496
497    """
498    import sys
499    if debug:
500        from meerschaum.utils.debug import dprint
501
502    if not plugins:
503        plugins = get_plugins_names()
504    for plugin_name in plugins:
505        if debug:
506            dprint(f"Reloading plugin '{plugin_name}'...")
507        mod_name = 'plugins.' + str(plugin_name)
508        if mod_name in sys.modules:
509            del sys.modules[mod_name]
510    load_plugins(debug=debug)
511
512
513def get_plugins(*to_load, try_import: bool = True) -> Union[Tuple[Plugin], Plugin]:
514    """
515    Return a list of `Plugin` objects.
516
517    Parameters
518    ----------
519    to_load:
520        If specified, only load specific plugins.
521        Otherwise return all plugins.
522
523    try_import: bool, default True
524        If `True`, allow for plugins to be imported.
525    """
526    from meerschaum.config._paths import PLUGINS_RESOURCES_PATH
527    import os
528    sync_plugins_symlinks()
529    _plugins = [
530        Plugin(name) for name in (
531            to_load or [
532                (
533                    name if (PLUGINS_RESOURCES_PATH / name).is_dir()
534                    else name[:-3]
535                ) for name in os.listdir(PLUGINS_RESOURCES_PATH) if name != '__init__.py'
536            ]
537        )
538    ]
539    plugins = tuple(plugin for plugin in _plugins if plugin.is_installed(try_import=try_import))
540    if len(to_load) == 1:
541        return plugins[0]
542    return plugins
543
544
545def get_plugins_names(*to_load, **kw) -> List[str]:
546    """
547    Return a list of installed plugins.
548    """
549    return [plugin.name for plugin in get_plugins(*to_load, **kw)]
550
551
552def get_plugins_modules(*to_load, **kw) -> List['ModuleType']:
553    """
554    Return a list of modules for the installed plugins, or `None` if things break.
555    """
556    return [plugin.module for plugin in get_plugins(*to_load, **kw)]
557
558
559def get_data_plugins() -> List[Plugin]:
560    """
561    Only return the modules of plugins with either `fetch()` or `sync()` functions.
562    """
563    import inspect
564    plugins = get_plugins()
565    data_names = {'sync', 'fetch'}
566    data_plugins = []
567    for plugin in plugins:
568        for name, ob in inspect.getmembers(plugin.module):
569            if not inspect.isfunction(ob):
570                continue
571            if name not in data_names:
572                continue
573            data_plugins.append(plugin)
574    return data_plugins
575
576
577def add_plugin_argument(*args, **kwargs) -> None:
578    """
579    Add argparse arguments under the 'Plugins options' group.
580    Takes the same parameters as the regular argparse `add_argument()` function.
581
582    Examples
583    --------
584    >>> add_plugin_argument('--foo', type=int, help="This is my help text!")
585    >>> 
586    """
587    from meerschaum._internal.arguments._parser import groups, _seen_plugin_args, parser
588    from meerschaum.utils.warnings import warn, error
589    _parent_plugin_name = _get_parent_plugin(2)
590    title = f"Plugin '{_parent_plugin_name}' options" if _parent_plugin_name else 'Custom options'
591    group_key = 'plugin_' + (_parent_plugin_name or '')
592    if group_key not in groups:
593        groups[group_key] = parser.add_argument_group(
594            title = title,
595        )
596        _seen_plugin_args[group_key] = set()
597    try:
598        if str(args) not in _seen_plugin_args[group_key]:
599            groups[group_key].add_argument(*args, **kwargs)
600            _seen_plugin_args[group_key].add(str(args))
601    except Exception as e:
602        warn(e)
603
604
605def _get_parent_plugin(stacklevel: int = 1) -> Union[str, None]:
606    """If this function is called from outside a Meerschaum plugin, it will return None."""
607    import inspect, re
608    try:
609        parent_globals = inspect.stack()[stacklevel][0].f_globals
610        parent_file = parent_globals.get('__file__', '')
611        return parent_globals['__name__'].replace('plugins.', '').split('.')[0]
612    except Exception as e:
613        return None

def make_action( function: Callable[[Any], Any], shell: bool = False, activate: bool = True, deactivate: bool = True, debug: bool = False) -> Callable[[Any], Any]: View Source

37def make_action(
38        function: Callable[[Any], Any],
39        shell: bool = False,
40        activate: bool = True,
41        deactivate: bool = True,
42        debug: bool = False
43    ) -> Callable[[Any], Any]:
44    """
45    Make a function a Meerschaum action. Useful for plugins that are adding multiple actions.
46    
47    Parameters
48    ----------
49    function: Callable[[Any], Any]
50        The function to become a Meerschaum action. Must accept all keyword arguments.
51        
52    shell: bool, default False
53        Not used.
54        
55    Returns
56    -------
57    Another function (this is a decorator function).
58
59    Examples
60    --------
61    >>> from meerschaum.plugins import make_action
62    >>>
63    >>> @make_action
64    ... def my_action(**kw):
65    ...     print('foo')
66    ...     return True, "Success"
67    >>>
68    """
69
70    from meerschaum.actions import actions
71    from meerschaum.utils.formatting import pprint
72    package_name = function.__globals__['__name__']
73    plugin_name = (
74        package_name.split('.')[1]
75        if package_name.startswith('plugins.') else None
76    )
77    plugin = Plugin(plugin_name) if plugin_name else None
78
79    if debug:
80        from meerschaum.utils.debug import dprint
81        dprint(
82            f"Adding action '{function.__name__}' from plugin " +
83            f"'{plugin}'..."
84        )
85
86    actions[function.__name__] = function
87    return function

Make a function a Meerschaum action. Useful for plugins that are adding multiple actions.

Parameters

function (Callable[[Any], Any]): The function to become a Meerschaum action. Must accept all keyword arguments.
shell (bool, default False): Not used.

Returns

Another function (this is a decorator function).

Examples

>>> from meerschaum.plugins import make_action
>>>
>>> @make_action
... def my_action(**kw):
...     print('foo')
...     return True, "Success"
>>>

def api_plugin(function: Callable[[Any], Any]) -> Callable[[Any], Any]: View Source

160def api_plugin(function: Callable[[Any], Any]) -> Callable[[Any], Any]:
161    """
162    Execute the function when initializing the Meerschaum API module.
163    Useful for lazy-loading heavy plugins only when the API is started,
164    such as when editing the `meerschaum.api.app` FastAPI app.
165    
166    The FastAPI app will be passed as the only parameter.
167    
168    Parameters
169    ----------
170    function: Callable[[Any, Any]]
171        The function to be called before starting the Meerschaum API.
172        
173    Returns
174    -------
175    Another function (decorator function).
176
177    Examples
178    --------
179    >>> from meerschaum.plugins import api_plugin
180    >>>
181    >>> @api_plugin
182    >>> def initialize_plugin(app):
183    ...     @app.get('/my/new/path')
184    ...     def new_path():
185    ...         return {'message': 'It works!'}
186    >>>
187    """
188    with _locks['_api_plugins']:
189        try:
190            if function.__module__ not in _api_plugins:
191                _api_plugins[function.__module__] = []
192            _api_plugins[function.__module__].append(function)
193        except Exception as e:
194            from meerschaum.utils.warnings import warn
195            warn(e)
196    return function

Execute the function when initializing the Meerschaum API module. Useful for lazy-loading heavy plugins only when the API is started, such as when editing the meerschaum.api.app FastAPI app.

The FastAPI app will be passed as the only parameter.

Parameters

function (Callable[[Any, Any]]): The function to be called before starting the Meerschaum API.

Returns

Another function (decorator function).

Examples

>>> from meerschaum.plugins import api_plugin
>>>
>>> @api_plugin
>>> def initialize_plugin(app):
...     @app.get('/my/new/path')
...     def new_path():
...         return {'message': 'It works!'}
>>>

def import_plugins( *plugins_to_import: Union[str, List[str], NoneType], warn: bool = True) -> "Union['ModuleType', Tuple['ModuleType', None]]": View Source

354def import_plugins(
355        *plugins_to_import: Union[str, List[str], None],
356        warn: bool = True,
357    ) -> Union[
358        'ModuleType', Tuple['ModuleType', None]
359    ]:
360    """
361    Import the Meerschaum plugins directory.
362
363    Parameters
364    ----------
365    plugins_to_import: Union[str, List[str], None]
366        If provided, only import the specified plugins.
367        Otherwise import the entire plugins module. May be a string, list, or `None`.
368        Defaults to `None`.
369
370    Returns
371    -------
372    A module of list of modules, depening on the number of plugins provided.
373
374    """
375    import sys
376    import os
377    import importlib
378    from meerschaum.utils.misc import flatten_list
379    from meerschaum.config._paths import PLUGINS_RESOURCES_PATH
380    from meerschaum.utils.venv import is_venv_active, activate_venv, deactivate_venv, Venv
381    from meerschaum.utils.warnings import warn as _warn
382    plugins_to_import = list(plugins_to_import)
383    with _locks['sys.path']:
384
385        ### Since plugins may depend on other plugins,
386        ### we need to activate the virtual environments for library plugins.
387        ### This logic exists in `Plugin.activate_venv()`,
388        ### but that code requires the plugin's module to already be imported.
389        ### It's not a guarantee of correct activation order,
390        ### e.g. if a library plugin pins a specific package and another 
391        plugins_names = get_plugins_names()
392        already_active_venvs = [is_venv_active(plugin_name) for plugin_name in plugins_names]
393
394        if not sys.path or sys.path[0] != str(PLUGINS_RESOURCES_PATH.parent):
395            sys.path.insert(0, str(PLUGINS_RESOURCES_PATH.parent))
396
397        if not plugins_to_import:
398            for plugin_name in plugins_names:
399                activate_venv(plugin_name)
400            try:
401                imported_plugins = importlib.import_module(PLUGINS_RESOURCES_PATH.stem)
402            except ImportError as e:
403                _warn(f"Failed to import the plugins module:\n    {e}")
404                import traceback
405                traceback.print_exc()
406                imported_plugins = None
407            for plugin_name in plugins_names:
408                if plugin_name in already_active_venvs:
409                    continue
410                deactivate_venv(plugin_name)
411
412        else:
413            imported_plugins = []
414            for plugin_name in flatten_list(plugins_to_import):
415                plugin = Plugin(plugin_name)
416                try:
417                    with Venv(plugin):
418                        imported_plugins.append(
419                            importlib.import_module(
420                                f'{PLUGINS_RESOURCES_PATH.stem}.{plugin_name}'
421                            )
422                        )
423                except Exception as e:
424                    _warn(
425                        f"Failed to import plugin '{plugin_name}':\n    "
426                        + f"{e}\n\nHere's a stacktrace:",
427                        stack = False,
428                    )
429                    from meerschaum.utils.formatting import get_console
430                    get_console().print_exception(
431                        suppress = [
432                            'meerschaum/plugins/__init__.py',
433                            importlib,
434                            importlib._bootstrap,
435                        ]
436                    )
437                    imported_plugins.append(None)
438
439        if imported_plugins is None and warn:
440            _warn(f"Failed to import plugins.", stacklevel=3)
441
442        if str(PLUGINS_RESOURCES_PATH.parent) in sys.path:
443            sys.path.remove(str(PLUGINS_RESOURCES_PATH.parent))
444
445    if isinstance(imported_plugins, list):
446        return (imported_plugins[0] if len(imported_plugins) == 1 else tuple(imported_plugins))
447    return imported_plugins

Import the Meerschaum plugins directory.

Parameters

plugins_to_import (Union[str, List[str], None]): If provided, only import the specified plugins. Otherwise import the entire plugins module. May be a string, list, or None. Defaults to None.

Returns

A module of list of modules, depening on the number of plugins provided.

def reload_plugins(plugins: Optional[List[str]] = None, debug: bool = False) -> None: View Source

489def reload_plugins(plugins: Optional[List[str]] = None, debug: bool = False) -> None:
490    """
491    Reload plugins back into memory.
492
493    Parameters
494    ----------
495    plugins: Optional[List[str]], default None
496        The plugins to reload. `None` will reload all plugins.
497
498    """
499    import sys
500    if debug:
501        from meerschaum.utils.debug import dprint
502
503    if not plugins:
504        plugins = get_plugins_names()
505    for plugin_name in plugins:
506        if debug:
507            dprint(f"Reloading plugin '{plugin_name}'...")
508        mod_name = 'plugins.' + str(plugin_name)
509        if mod_name in sys.modules:
510            del sys.modules[mod_name]
511    load_plugins(debug=debug)

Reload plugins back into memory.

Parameters

plugins (Optional[List[str]], default None): The plugins to reload. None will reload all plugins.

def get_plugins( *to_load, try_import: bool = True) -> Union[Tuple[Plugin], Plugin]: View Source

514def get_plugins(*to_load, try_import: bool = True) -> Union[Tuple[Plugin], Plugin]:
515    """
516    Return a list of `Plugin` objects.
517
518    Parameters
519    ----------
520    to_load:
521        If specified, only load specific plugins.
522        Otherwise return all plugins.
523
524    try_import: bool, default True
525        If `True`, allow for plugins to be imported.
526    """
527    from meerschaum.config._paths import PLUGINS_RESOURCES_PATH
528    import os
529    sync_plugins_symlinks()
530    _plugins = [
531        Plugin(name) for name in (
532            to_load or [
533                (
534                    name if (PLUGINS_RESOURCES_PATH / name).is_dir()
535                    else name[:-3]
536                ) for name in os.listdir(PLUGINS_RESOURCES_PATH) if name != '__init__.py'
537            ]
538        )
539    ]
540    plugins = tuple(plugin for plugin in _plugins if plugin.is_installed(try_import=try_import))
541    if len(to_load) == 1:
542        return plugins[0]
543    return plugins

Return a list of Plugin objects.

Parameters

to_load:: If specified, only load specific plugins. Otherwise return all plugins.
try_import (bool, default True): If True, allow for plugins to be imported.

def get_data_plugins() -> List[Plugin]: View Source

560def get_data_plugins() -> List[Plugin]:
561    """
562    Only return the modules of plugins with either `fetch()` or `sync()` functions.
563    """
564    import inspect
565    plugins = get_plugins()
566    data_names = {'sync', 'fetch'}
567    data_plugins = []
568    for plugin in plugins:
569        for name, ob in inspect.getmembers(plugin.module):
570            if not inspect.isfunction(ob):
571                continue
572            if name not in data_names:
573                continue
574            data_plugins.append(plugin)
575    return data_plugins

Only return the modules of plugins with either fetch() or sync() functions.

def add_plugin_argument(*args, **kwargs) -> None: View Source

578def add_plugin_argument(*args, **kwargs) -> None:
579    """
580    Add argparse arguments under the 'Plugins options' group.
581    Takes the same parameters as the regular argparse `add_argument()` function.
582
583    Examples
584    --------
585    >>> add_plugin_argument('--foo', type=int, help="This is my help text!")
586    >>> 
587    """
588    from meerschaum._internal.arguments._parser import groups, _seen_plugin_args, parser
589    from meerschaum.utils.warnings import warn, error
590    _parent_plugin_name = _get_parent_plugin(2)
591    title = f"Plugin '{_parent_plugin_name}' options" if _parent_plugin_name else 'Custom options'
592    group_key = 'plugin_' + (_parent_plugin_name or '')
593    if group_key not in groups:
594        groups[group_key] = parser.add_argument_group(
595            title = title,
596        )
597        _seen_plugin_args[group_key] = set()
598    try:
599        if str(args) not in _seen_plugin_args[group_key]:
600            groups[group_key].add_argument(*args, **kwargs)
601            _seen_plugin_args[group_key].add(str(args))
602    except Exception as e:
603        warn(e)

Add argparse arguments under the 'Plugins options' group. Takes the same parameters as the regular argparse add_argument() function.

Examples

>>> add_plugin_argument('--foo', type=int, help="This is my help text!")
>>>

def pre_sync_hook(function: Callable[[Any], Any]) -> Callable[[Any], Any]: View Source

 90def pre_sync_hook(
 91        function: Callable[[Any], Any],
 92    ) -> Callable[[Any], Any]:
 93    """
 94    Register a function as a sync hook to be executed right before sync.
 95    
 96    Parameters
 97    ----------
 98    function: Callable[[Any], Any]
 99        The function to execute right before a sync.
100        
101    Returns
102    -------
103    Another function (this is a decorator function).
104
105    Examples
106    --------
107    >>> from meerschaum.plugins import pre_sync_hook
108    >>>
109    >>> @pre_sync_hook
110    ... def log_sync(pipe, **kwargs):
111    ...     print(f"About to sync {pipe} with kwargs:\n{kwargs}.")
112    >>>
113    """
114    with _locks['_pre_sync_hooks']:
115        try:
116            if function.__module__ not in _pre_sync_hooks:
117                _pre_sync_hooks[function.__module__] = []
118            _pre_sync_hooks[function.__module__].append(function)
119        except Exception as e:
120            from meerschaum.utils.warnings import warn
121            warn(e)
122    return function

Parameters
----------
function: Callable[[Any], Any]
    The function to execute right before a sync.

Returns
-------
Another function (this is a decorator function).

Examples
--------
>>> from meerschaum.plugins import pre_sync_hook
>>>
>>> @pre_sync_hook
... def log_sync(pipe, **kwargs):
...     print(f"About to sync {pipe} with kwargs:

{kwargs}.")

>

def post_sync_hook(function: Callable[[Any], Any]) -> Callable[[Any], Any]: View Source

125def post_sync_hook(
126        function: Callable[[Any], Any],
127    ) -> Callable[[Any], Any]:
128    """
129    Register a function as a sync hook to be executed upon completion of a sync.
130    
131    Parameters
132    ----------
133    function: Callable[[Any], Any]
134        The function to execute upon completion of a sync.
135        
136    Returns
137    -------
138    Another function (this is a decorator function).
139
140    Examples
141    --------
142    >>> from meerschaum.plugins import post_sync_hook
143    >>>
144    >>> @post_sync_hook
145    ... def log_sync(pipe, success_tuple, duration=None, **kwargs):
146    ...     print(f"It took {round(duration, 2)} seconds to sync {pipe}.")
147    >>>
148    """
149    with _locks['_post_sync_hooks']:
150        try:
151            if function.__module__ not in _post_sync_hooks:
152                _post_sync_hooks[function.__module__] = []
153            _post_sync_hooks[function.__module__].append(function)
154        except Exception as e:
155            from meerschaum.utils.warnings import warn
156            warn(e)
157    return function

Parameters

function (Callable[[Any], Any]): The function to execute upon completion of a sync.

Returns

Another function (this is a decorator function).

Examples

>>> from meerschaum.plugins import post_sync_hook
>>>
>>> @post_sync_hook
... def log_sync(pipe, success_tuple, duration=None, **kwargs):
...     print(f"It took {round(duration, 2)} seconds to sync {pipe}.")
>>>