meerschaum.plugins

   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
  10
  11import pathlib
  12import functools
  13
  14import meerschaum as mrsm
  15from meerschaum.utils.typing import Callable, Any, Union, Optional, Dict, List, Tuple
  16from meerschaum.utils.threading import RLock
  17from meerschaum.core.Plugin import Plugin
  18
  19_api_plugins: Dict[str, List[Callable[['fastapi.App'], Any]]] = {}
  20_pre_sync_hooks: Dict[Union[str, None], List[Callable[[Any], Any]]] = {}
  21_post_sync_hooks: Dict[Union[str, None], List[Callable[[Any], Any]]] = {}
  22_actions_daemon_enabled: Dict[str, bool] = {}
  23_locks = {
  24    '_api_plugins': RLock(),
  25    '_dash_plugins': RLock(),
  26    '_pre_sync_hooks': RLock(),
  27    '_post_sync_hooks': RLock(),
  28    '_actions_daemon_enabled': RLock(),
  29    '__path__': RLock(),
  30    'sys.path': RLock(),
  31    'internal_plugins': RLock(),
  32    '_synced_symlinks': RLock(),
  33    'PLUGINS_INTERNAL_LOCK_PATH': RLock(),
  34}
  35__all__ = (
  36    "Plugin",
  37    "make_action",
  38    "api_plugin",
  39    "dash_plugin",
  40    "web_page",
  41    "import_plugins",
  42    "from_plugin_import",
  43    "reload_plugins",
  44    "get_plugins",
  45    "get_data_plugins",
  46    "add_plugin_argument",
  47    "pre_sync_hook",
  48    "post_sync_hook",
  49)
  50__pdoc__ = {
  51    'venvs': False,
  52    'data': False,
  53    'stack': False,
  54    'plugins': False,
  55}
  56
  57
  58def make_action(
  59    function: Optional[Callable[[Any], Any]] = None,
  60    shell: bool = False,
  61    activate: bool = True,
  62    deactivate: bool = True,
  63    debug: bool = False,
  64    daemon: bool = True,
  65    skip_if_loaded: bool = True,
  66    _plugin_name: Optional[str] = None,
  67) -> Callable[[Any], Any]:
  68    """
  69    Make a function a Meerschaum action. Useful for plugins that are adding multiple actions.
  70    
  71    Parameters
  72    ----------
  73    function: Callable[[Any], Any]
  74        The function to become a Meerschaum action. Must accept all keyword arguments.
  75        
  76    shell: bool, default False
  77        Not used.
  78        
  79    Returns
  80    -------
  81    Another function (this is a decorator function).
  82
  83    Examples
  84    --------
  85    >>> from meerschaum.plugins import make_action
  86    >>>
  87    >>> @make_action
  88    ... def my_action(**kw):
  89    ...     print('foo')
  90    ...     return True, "Success"
  91    >>>
  92    """
  93    def _decorator(func: Callable[[Any], Any]) -> Callable[[Any], Any]:
  94        from meerschaum.actions import actions, _custom_actions_plugins, _plugins_actions
  95        if skip_if_loaded and func.__name__ in actions:
  96            return func
  97
  98        from meerschaum.config.paths import PLUGINS_RESOURCES_PATH
  99        plugin_name = (
 100            func.__name__.split(f"{PLUGINS_RESOURCES_PATH.stem}.", maxsplit=1)[-1].split('.')[0]
 101        )
 102        plugin = Plugin(plugin_name) if plugin_name else None
 103
 104        if debug:
 105            from meerschaum.utils.debug import dprint
 106            dprint(
 107                f"Adding action '{func.__name__}' from plugin "
 108                f"'{plugin}'..."
 109            )
 110
 111        actions[func.__name__] = func
 112        _custom_actions_plugins[func.__name__] = plugin_name
 113        if plugin_name not in _plugins_actions:
 114            _plugins_actions[plugin_name] = []
 115        _plugins_actions[plugin_name].append(func.__name__)
 116        if not daemon:
 117            _actions_daemon_enabled[func.__name__] = False
 118        return func
 119
 120    if function is None:
 121        return _decorator
 122    return _decorator(function)
 123
 124
 125def pre_sync_hook(
 126    function: Callable[[Any], Any],
 127) -> Callable[[Any], Any]:
 128    """
 129    Register a function as a sync hook to be executed right before sync.
 130    
 131    Parameters
 132    ----------
 133    function: Callable[[Any], Any]
 134        The function to execute right before a sync.
 135        
 136    Returns
 137    -------
 138    Another function (this is a decorator function).
 139
 140    Examples
 141    --------
 142    >>> from meerschaum.plugins import pre_sync_hook
 143    >>>
 144    >>> @pre_sync_hook
 145    ... def log_sync(pipe, **kwargs):
 146    ...     print(f"About to sync {pipe} with kwargs:\n{kwargs}.")
 147    >>>
 148    """
 149    with _locks['_pre_sync_hooks']:
 150        plugin_name = _get_parent_plugin()
 151        try:
 152            if plugin_name not in _pre_sync_hooks:
 153                _pre_sync_hooks[plugin_name] = []
 154            _pre_sync_hooks[plugin_name].append(function)
 155        except Exception as e:
 156            from meerschaum.utils.warnings import warn
 157            warn(e)
 158    return function
 159
 160
 161def post_sync_hook(
 162    function: Callable[[Any], Any],
 163) -> Callable[[Any], Any]:
 164    """
 165    Register a function as a sync hook to be executed upon completion of a sync.
 166    
 167    Parameters
 168    ----------
 169    function: Callable[[Any], Any]
 170        The function to execute upon completion of a sync.
 171        
 172    Returns
 173    -------
 174    Another function (this is a decorator function).
 175
 176    Examples
 177    --------
 178    >>> from meerschaum.plugins import post_sync_hook
 179    >>> from meerschaum.utils.misc import interval_str
 180    >>> from datetime import timedelta
 181    >>>
 182    >>> @post_sync_hook
 183    ... def log_sync(pipe, success_tuple, duration=None, **kwargs):
 184    ...     duration_delta = timedelta(seconds=duration)
 185    ...     duration_text = interval_str(duration_delta)
 186    ...     print(f"It took {duration_text} to sync {pipe}.")
 187    >>>
 188    """
 189    with _locks['_post_sync_hooks']:
 190        try:
 191            plugin_name = _get_parent_plugin()
 192            if plugin_name not in _post_sync_hooks:
 193                _post_sync_hooks[plugin_name] = []
 194            _post_sync_hooks[plugin_name].append(function)
 195        except Exception as e:
 196            from meerschaum.utils.warnings import warn
 197            warn(e)
 198    return function
 199
 200
 201_plugin_endpoints_to_pages = {}
 202_plugins_web_pages = {}
 203def web_page(
 204    page: Union[str, None, Callable[[Any], Any]] = None,
 205    login_required: bool = True,
 206    skip_navbar: bool = False,
 207    page_group: Optional[str] = None,
 208    **kwargs
 209) -> Any:
 210    """
 211    Quickly add pages to the dash application.
 212
 213    Examples
 214    --------
 215    >>> import meerschaum as mrsm
 216    >>> from meerschaum.plugins import web_page
 217    >>> html = mrsm.attempt_import('dash.html')
 218    >>> 
 219    >>> @web_page('foo/bar', login_required=False)
 220    >>> def foo_bar():
 221    ...     return html.Div([html.H1("Hello, World!")])
 222    >>> 
 223    """
 224    page_str = None
 225
 226    def _decorator(_func: Callable[[Any], Any]) -> Callable[[Any], Any]:
 227        nonlocal page_str, page_group
 228
 229        @functools.wraps(_func)
 230        def wrapper(*_args, **_kwargs):
 231            return _func(*_args, **_kwargs)
 232
 233        if page_str is None:
 234            page_str = _func.__name__
 235
 236        page_str = page_str.lstrip('/').rstrip('/').strip()
 237        if not page_str.startswith('dash'):
 238            page_str = f'/dash/{page_str}'
 239        page_key = (
 240            ' '.join(
 241                [
 242                    word.capitalize()
 243                    for word in (
 244                        page_str.replace('/dash', '').lstrip('/').rstrip('/').strip()
 245                        .replace('-', ' ').replace('_', ' ').split(' ')
 246                    )
 247                ]
 248            )
 249        )
 250 
 251        plugin_name = _get_parent_plugin()
 252        page_group = page_group or plugin_name
 253        if page_group not in _plugin_endpoints_to_pages:
 254            _plugin_endpoints_to_pages[page_group] = {}
 255        _plugin_endpoints_to_pages[page_group][page_str] = {
 256            'function': _func,
 257            'login_required': login_required,
 258            'skip_navbar': skip_navbar,
 259            'page_key': page_key,
 260        }
 261        if plugin_name not in _plugins_web_pages:
 262            _plugins_web_pages[plugin_name] = []
 263        _plugins_web_pages[plugin_name].append(_func)
 264        return wrapper
 265
 266    if callable(page):
 267        decorator_to_return = _decorator(page)
 268        page_str = page.__name__
 269    else:
 270        decorator_to_return = _decorator
 271        page_str = page
 272
 273    return decorator_to_return
 274
 275
 276_dash_plugins = {}
 277def dash_plugin(function: Callable[[Any], Any]) -> Callable[[Any], Any]:
 278    """
 279    Execute the function when starting the Dash application.
 280    """
 281    with _locks['_dash_plugins']:
 282        plugin_name = _get_parent_plugin()
 283        try:
 284            if plugin_name not in _dash_plugins:
 285                _dash_plugins[plugin_name] = []
 286            _dash_plugins[plugin_name].append(function)
 287        except Exception as e:
 288            from meerschaum.utils.warnings import warn
 289            warn(e)
 290    return function
 291
 292
 293def api_plugin(function: Callable[[Any], Any]) -> Callable[[Any], Any]:
 294    """
 295    Execute the function when initializing the Meerschaum API module.
 296    Useful for lazy-loading heavy plugins only when the API is started,
 297    such as when editing the `meerschaum.api.app` FastAPI app.
 298    
 299    The FastAPI app will be passed as the only parameter.
 300    
 301    Examples
 302    --------
 303    >>> from meerschaum.plugins import api_plugin
 304    >>>
 305    >>> @api_plugin
 306    >>> def initialize_plugin(app):
 307    ...     @app.get('/my/new/path')
 308    ...     def new_path():
 309    ...         return {'message': 'It works!'}
 310    >>>
 311    """
 312    with _locks['_api_plugins']:
 313        try:
 314            if function.__module__ not in _api_plugins:
 315                _api_plugins[function.__module__] = []
 316            _api_plugins[function.__module__].append(function)
 317        except Exception as e:
 318            from meerschaum.utils.warnings import warn
 319            warn(e)
 320    return function
 321
 322
 323_synced_symlinks: bool = False
 324_injected_plugin_symlinks = set()
 325def sync_plugins_symlinks(debug: bool = False, warn: bool = True) -> None:
 326    """
 327    Update the plugins' internal symlinks. 
 328    """
 329    from meerschaum.utils.warnings import error, warn as _warn, dprint
 330    global _synced_symlinks
 331    with _locks['_synced_symlinks']:
 332        if _synced_symlinks:
 333            if debug:
 334                dprint("Skip syncing symlinks...")
 335            return
 336
 337    import os
 338    import pathlib
 339    import time
 340    from collections import defaultdict
 341    from meerschaum.utils.misc import flatten_list, make_symlink, is_symlink
 342    from meerschaum._internal.static import STATIC_CONFIG
 343    from meerschaum.config._paths import (
 344        PLUGINS_RESOURCES_PATH,
 345        PLUGINS_INIT_PATH,
 346        PLUGINS_DIR_PATHS,
 347        PLUGINS_INTERNAL_LOCK_PATH,
 348        PLUGINS_INJECTED_RESOURCES_PATH,
 349    )
 350
 351    ### If the lock file exists, sleep for up to a second or until it's removed before continuing.
 352    with _locks['PLUGINS_INTERNAL_LOCK_PATH']:
 353        if PLUGINS_INTERNAL_LOCK_PATH.exists():
 354            lock_sleep_total     = STATIC_CONFIG['plugins']['lock_sleep_total']
 355            lock_sleep_increment = STATIC_CONFIG['plugins']['lock_sleep_increment']
 356            lock_start = time.perf_counter()
 357            while (
 358                (time.perf_counter() - lock_start) < lock_sleep_total
 359            ):
 360                time.sleep(lock_sleep_increment)
 361                if not PLUGINS_INTERNAL_LOCK_PATH.exists():
 362                    break
 363                try:
 364                    PLUGINS_INTERNAL_LOCK_PATH.unlink()
 365                except Exception as e:
 366                    if warn:
 367                        _warn(f"Error while removing lockfile {PLUGINS_INTERNAL_LOCK_PATH}:\n{e}")
 368                    break
 369
 370        ### Begin locking from other processes.
 371        try:
 372            PLUGINS_INTERNAL_LOCK_PATH.touch()
 373        except Exception as e:
 374            if warn:
 375                _warn(f"Unable to create lockfile {PLUGINS_INTERNAL_LOCK_PATH}:\n{e}")
 376
 377    with _locks['internal_plugins']:
 378
 379        try:
 380            from importlib.metadata import entry_points
 381        except ImportError:
 382            importlib_metadata = mrsm.attempt_import('importlib_metadata', lazy=False)
 383            entry_points = importlib_metadata.entry_points
 384
 385        ### NOTE: Allow plugins to be installed via `pip`.
 386        packaged_plugin_paths = []
 387        try:
 388            discovered_packaged_plugins_eps = entry_points(group='meerschaum.plugins')
 389        except TypeError:
 390            discovered_packaged_plugins_eps = []
 391
 392        for ep in discovered_packaged_plugins_eps:
 393            module_name = ep.name
 394            for package_file_path in ep.dist.files:
 395                if package_file_path.suffix != '.py':
 396                    continue
 397                if str(package_file_path) == f'{module_name}.py':
 398                    packaged_plugin_paths.append(package_file_path.locate())
 399                elif str(package_file_path) == f'{module_name}/__init__.py':
 400                    packaged_plugin_paths.append(package_file_path.locate().parent)
 401
 402        if is_symlink(PLUGINS_RESOURCES_PATH) or not PLUGINS_RESOURCES_PATH.exists():
 403            try:
 404                PLUGINS_RESOURCES_PATH.unlink()
 405            except Exception:
 406                pass
 407
 408        PLUGINS_RESOURCES_PATH.mkdir(exist_ok=True)
 409
 410        existing_symlinked_paths = {
 411            _existing_symlink: pathlib.Path(os.path.realpath(_existing_symlink))
 412            for item in os.listdir(PLUGINS_RESOURCES_PATH)
 413            if is_symlink(_existing_symlink := (PLUGINS_RESOURCES_PATH / item))
 414        }
 415        injected_symlinked_paths = {
 416            _injected_symlink: pathlib.Path(os.path.realpath(_injected_symlink))
 417            for item in os.listdir(PLUGINS_INJECTED_RESOURCES_PATH)
 418            if is_symlink(_injected_symlink := (PLUGINS_INJECTED_RESOURCES_PATH / item))
 419        }
 420        plugins_to_be_symlinked = list(flatten_list(
 421            [
 422                [
 423                    pathlib.Path(os.path.realpath(plugins_path / item))
 424                    for item in os.listdir(plugins_path)
 425                    if (
 426                        not item.startswith('.')
 427                    ) and (item not in ('__pycache__', '__init__.py'))
 428                ]
 429                for plugins_path in PLUGINS_DIR_PATHS
 430                if plugins_path.exists()
 431            ]
 432        ))
 433        plugins_to_be_symlinked.extend(packaged_plugin_paths)
 434
 435        ### Check for duplicates.
 436        seen_plugins = defaultdict(lambda: 0)
 437        for plugin_path in plugins_to_be_symlinked:
 438            plugin_name = plugin_path.stem
 439            seen_plugins[plugin_name] += 1
 440        for plugin_name, plugin_count in seen_plugins.items():
 441            if plugin_count > 1:
 442                if warn:
 443                    _warn(f"Found duplicate plugins named '{plugin_name}'.")
 444
 445        for plugin_symlink_path, real_path in existing_symlinked_paths.items():
 446
 447            ### Remove invalid symlinks.
 448            if real_path not in plugins_to_be_symlinked:
 449                if plugin_symlink_path in _injected_plugin_symlinks:
 450                    continue
 451                if plugin_symlink_path in injected_symlinked_paths:
 452                    continue
 453                if real_path in injected_symlinked_paths.values():
 454                    continue
 455                try:
 456                    plugin_symlink_path.unlink()
 457                except Exception:
 458                    pass
 459
 460            ### Remove valid plugins from the to-be-symlinked list.
 461            else:
 462                plugins_to_be_symlinked.remove(real_path)
 463
 464        for plugin_path in plugins_to_be_symlinked:
 465            plugin_symlink_path = PLUGINS_RESOURCES_PATH / plugin_path.name
 466            try:
 467                ### There might be duplicate folders (e.g. __pycache__).
 468                if (
 469                    plugin_symlink_path.exists()
 470                    and
 471                    plugin_symlink_path.is_dir()
 472                    and
 473                    not is_symlink(plugin_symlink_path)
 474                ):
 475                    continue
 476                success, msg = make_symlink(plugin_path, plugin_symlink_path)
 477            except Exception as e:
 478                success, msg = False, str(e)
 479            if not success:
 480                if warn:
 481                    _warn(
 482                        f"Failed to create symlink {plugin_symlink_path} "
 483                        + f"to {plugin_path}:\n    {msg}"
 484                    )
 485
 486    ### Release symlink lock file in case other processes need it.
 487    with _locks['PLUGINS_INTERNAL_LOCK_PATH']:
 488        try:
 489            if PLUGINS_INTERNAL_LOCK_PATH.exists():
 490                PLUGINS_INTERNAL_LOCK_PATH.unlink()
 491        ### Sometimes competing threads will delete the lock file at the same time.
 492        except FileNotFoundError:
 493            pass
 494        except Exception as e:
 495            if warn:
 496                _warn(f"Error cleaning up lockfile {PLUGINS_INTERNAL_LOCK_PATH}:\n{e}")
 497
 498        try:
 499            if not PLUGINS_INIT_PATH.exists():
 500                PLUGINS_INIT_PATH.touch()
 501        except Exception as e:
 502            error(f"Failed to create the file '{PLUGINS_INIT_PATH}':\n{e}")
 503
 504    with _locks['__path__']:
 505        if str(PLUGINS_RESOURCES_PATH.parent) not in __path__:
 506            __path__.append(str(PLUGINS_RESOURCES_PATH.parent))
 507
 508    with _locks['_synced_symlinks']:
 509        _synced_symlinks = True
 510
 511
 512def import_plugins(
 513    *plugins_to_import: Union[str, List[str], None],
 514    warn: bool = True,
 515) -> Union[
 516    'ModuleType', Tuple['ModuleType', None]
 517]:
 518    """
 519    Import the Meerschaum plugins directory.
 520
 521    Parameters
 522    ----------
 523    plugins_to_import: Union[str, List[str], None]
 524        If provided, only import the specified plugins.
 525        Otherwise import the entire plugins module. May be a string, list, or `None`.
 526        Defaults to `None`.
 527
 528    Returns
 529    -------
 530    A module of list of modules, depening on the number of plugins provided.
 531
 532    """
 533    import sys
 534    import importlib
 535    from meerschaum.utils.misc import flatten_list
 536    from meerschaum.config._paths import PLUGINS_RESOURCES_PATH
 537    from meerschaum.utils.venv import is_venv_active, activate_venv, deactivate_venv, Venv
 538    from meerschaum.utils.warnings import warn as _warn
 539    plugins_to_import = list(plugins_to_import)
 540    prepended_sys_path = False
 541    with _locks['sys.path']:
 542
 543        ### Since plugins may depend on other plugins,
 544        ### we need to activate the virtual environments for library plugins.
 545        ### This logic exists in `Plugin.activate_venv()`,
 546        ### but that code requires the plugin's module to already be imported.
 547        ### It's not a guarantee of correct activation order,
 548        ### e.g. if a library plugin pins a specific package and another 
 549        plugins_names = get_plugins_names()
 550        already_active_venvs = [is_venv_active(plugin_name) for plugin_name in plugins_names]
 551
 552        if not sys.path or sys.path[0] != str(PLUGINS_RESOURCES_PATH.parent):
 553            prepended_sys_path = True
 554            sys.path.insert(0, str(PLUGINS_RESOURCES_PATH.parent))
 555
 556        if not plugins_to_import:
 557            for plugin_name in plugins_names:
 558                activate_venv(plugin_name)
 559            try:
 560                imported_plugins = importlib.import_module(PLUGINS_RESOURCES_PATH.stem)
 561            except ImportError as e:
 562                _warn(f"Failed to import the plugins module:\n    {e}")
 563                import traceback
 564                traceback.print_exc()
 565                imported_plugins = None
 566            for plugin_name in plugins_names:
 567                if plugin_name in already_active_venvs:
 568                    continue
 569                deactivate_venv(plugin_name)
 570
 571        else:
 572            imported_plugins = []
 573            for plugin_name in flatten_list(plugins_to_import):
 574                plugin = Plugin(plugin_name)
 575                try:
 576                    with Venv(plugin, init_if_not_exists=False):
 577                        imported_plugins.append(
 578                            importlib.import_module(
 579                                f'{PLUGINS_RESOURCES_PATH.stem}.{plugin_name}'
 580                            )
 581                        )
 582                except Exception as e:
 583                    _warn(
 584                        f"Failed to import plugin '{plugin_name}':\n    "
 585                        + f"{e}\n\nHere's a stacktrace:",
 586                        stack = False,
 587                    )
 588                    from meerschaum.utils.formatting import get_console
 589                    get_console().print_exception(
 590                        suppress = [
 591                            'meerschaum/plugins/__init__.py',
 592                            importlib,
 593                            importlib._bootstrap,
 594                        ]
 595                    )
 596                    imported_plugins.append(None)
 597
 598        if imported_plugins is None and warn:
 599            _warn("Failed to import plugins.", stacklevel=3)
 600
 601        if prepended_sys_path and str(PLUGINS_RESOURCES_PATH.parent) in sys.path:
 602            sys.path.remove(str(PLUGINS_RESOURCES_PATH.parent))
 603
 604    if isinstance(imported_plugins, list):
 605        return (imported_plugins[0] if len(imported_plugins) == 1 else tuple(imported_plugins))
 606    return imported_plugins
 607
 608
 609def from_plugin_import(plugin_import_name: str, *attrs: str) -> Any:
 610    """
 611    Emulate the `from module import x` behavior.
 612
 613    Parameters
 614    ----------
 615    plugin_import_name: str
 616        The import name of the plugin's module.
 617        Separate submodules with '.' (e.g. 'compose.utils.pipes')
 618
 619    attrs: str
 620        Names of the attributes to return.
 621
 622    Returns
 623    -------
 624    Objects from a plugin's submodule.
 625    If multiple objects are provided, return a tuple.
 626
 627    Examples
 628    --------
 629    >>> init = from_plugin_import('compose.utils', 'init')
 630    >>> with mrsm.Venv('compose'):
 631    ...     cf = init()
 632    >>> build_parent_pipe, get_defined_pipes = from_plugin_import(
 633    ...     'compose.utils.pipes',
 634    ...     'build_parent_pipe',
 635    ...     'get_defined_pipes',
 636    ... )
 637    >>> parent_pipe = build_parent_pipe(cf)
 638    >>> defined_pipes = get_defined_pipes(cf)
 639    """
 640    import importlib
 641    from meerschaum.config._paths import PLUGINS_RESOURCES_PATH
 642    from meerschaum.utils.warnings import warn as _warn
 643    if plugin_import_name.startswith('plugins.'):
 644        plugin_import_name = plugin_import_name[len('plugins.'):]
 645    plugin_import_parts = plugin_import_name.split('.')
 646    plugin_root_name = plugin_import_parts[0]
 647    plugin = mrsm.Plugin(plugin_root_name)
 648
 649    submodule_import_name = '.'.join(
 650        [PLUGINS_RESOURCES_PATH.stem]
 651        + plugin_import_parts
 652    )
 653    if len(attrs) == 0:
 654        raise ValueError(f"Provide which attributes to return from '{submodule_import_name}'.")
 655
 656    attrs_to_return = []
 657    with mrsm.Venv(plugin):
 658        if plugin.module is None:
 659            raise ImportError(f"Unable to import plugin '{plugin}'.")
 660
 661        try:
 662            submodule = importlib.import_module(submodule_import_name)
 663        except ImportError as e:
 664            _warn(
 665                f"Failed to import plugin '{submodule_import_name}':\n    "
 666                + f"{e}\n\nHere's a stacktrace:",
 667                stack=False,
 668            )
 669            from meerschaum.utils.formatting import get_console
 670            get_console().print_exception(
 671                suppress=[
 672                    'meerschaum/plugins/__init__.py',
 673                    importlib,
 674                    importlib._bootstrap,
 675                ]
 676            )
 677            return None
 678
 679        for attr in attrs:
 680            try:
 681                attrs_to_return.append(getattr(submodule, attr))
 682            except Exception:
 683                _warn(f"Failed to access '{attr}' from '{submodule_import_name}'.")
 684                attrs_to_return.append(None)
 685        
 686        if len(attrs) == 1:
 687            return attrs_to_return[0]
 688
 689        return tuple(attrs_to_return)
 690
 691
 692_loaded_plugins: bool = False
 693def load_plugins(
 694    skip_if_loaded: bool = True,
 695    shell: bool = False,
 696    debug: bool = False,
 697) -> None:
 698    """
 699    Import Meerschaum plugins and update the actions dictionary.
 700    """
 701    global _loaded_plugins
 702    from meerschaum.utils.warnings import dprint
 703
 704    if skip_if_loaded and _loaded_plugins:
 705        if debug:
 706            dprint("Skip loading plugins...")
 707        return
 708
 709    from inspect import isfunction, getmembers
 710    from meerschaum.actions import __all__ as _all, modules
 711    from meerschaum.config._paths import PLUGINS_RESOURCES_PATH
 712    from meerschaum.utils.packages import get_modules_from_package
 713
 714    _plugins_names, plugins_modules = get_modules_from_package(
 715        import_plugins(),
 716        names = True,
 717        recursive = True,
 718        modules_venvs = True
 719    )
 720
 721    ### I'm appending here to keep from redefining the modules list.
 722    new_modules = (
 723        [
 724            mod
 725            for mod in modules
 726            if not mod.__name__.startswith(PLUGINS_RESOURCES_PATH.stem + '.')
 727        ]
 728        + plugins_modules
 729    )
 730    n_mods = len(modules)
 731    for mod in new_modules:
 732        modules.append(mod)
 733    for i in range(n_mods):
 734        modules.pop(0)
 735
 736    for module in plugins_modules:
 737        for name, func in getmembers(module):
 738            if not isfunction(func):
 739                continue
 740            if name == module.__name__.split('.')[-1]:
 741                make_action(
 742                    func,
 743                    **{'shell': shell, 'debug': debug},
 744                    _plugin_name=name,
 745                    skip_if_loaded=True,
 746                )
 747
 748    _loaded_plugins = True
 749
 750
 751def unload_custom_actions(plugins: Optional[List[str]] = None, debug: bool = False) -> None:
 752    """
 753    Unload the custom actions added by plugins.
 754    """
 755    from meerschaum.actions import (
 756        actions,
 757        _custom_actions_plugins,
 758        _plugins_actions,
 759    )
 760    from meerschaum._internal.entry import _shell
 761    import meerschaum._internal.shell as shell_pkg
 762
 763    plugins = plugins if plugins is not None else list(_plugins_actions)
 764
 765    for plugin in plugins:
 766        action_names = _plugins_actions.get(plugin, [])
 767        actions_to_remove = {
 768            action_name: actions.get(action_name, None)
 769            for action_name in action_names
 770        }
 771        for action_name in action_names:
 772            _ = actions.pop(action_name, None)
 773            _ = _custom_actions_plugins.pop(action_name, None)
 774            _ = _actions_daemon_enabled.pop(action_name, None)
 775
 776        _ = _plugins_actions.pop(plugin, None)
 777        shell_pkg._remove_shell_actions(
 778            _shell=_shell,
 779            actions=actions_to_remove,
 780        )
 781    
 782
 783def unload_plugins(
 784    plugins: Optional[List[str]] = None,
 785    remove_symlinks: bool = True,
 786    debug: bool = False,
 787) -> None:
 788    """
 789    Unload the specified plugins from memory.
 790    """
 791    global _loaded_plugins, _synced_symlinks
 792    import sys
 793    from meerschaum.config.paths import PLUGINS_RESOURCES_PATH, PLUGINS_INJECTED_RESOURCES_PATH
 794    from meerschaum.connectors import unload_plugin_connectors
 795    if debug:
 796        from meerschaum.utils.warnings import dprint
 797
 798    _loaded_plugins = False
 799    _synced_symlinks = False
 800
 801    all_plugins = get_plugins_names()
 802    plugins = plugins if plugins is not None else all_plugins
 803    if debug:
 804        dprint(f"Unloading plugins: {plugins}")
 805
 806    unload_custom_actions(plugins, debug=debug)
 807    unload_plugin_connectors(plugins, debug=debug)
 808
 809    module_prefix = f"{PLUGINS_RESOURCES_PATH.stem}."
 810    loaded_modules = [mod_name for mod_name in sys.modules if mod_name.startswith(module_prefix)]
 811
 812    root_plugins_mod = (
 813        sys.modules.get(PLUGINS_RESOURCES_PATH.stem, None)
 814        if sorted(plugins) != sorted(all_plugins)
 815        else sys.modules.pop(PLUGINS_RESOURCES_PATH, None)
 816    )
 817
 818    for plugin_name in plugins:
 819        for mod_name in loaded_modules:
 820            if (
 821                mod_name[len(PLUGINS_RESOURCES_PATH.stem):].startswith(plugin_name + '.')
 822                or mod_name[len(PLUGINS_RESOURCES_PATH.stem):] == plugin_name
 823            ):
 824                _ = sys.modules.pop(mod_name, None)
 825
 826        if root_plugins_mod is not None and plugin_name in root_plugins_mod.__dict__:
 827            try:
 828                delattr(root_plugins_mod, plugin_name)
 829            except Exception:
 830                pass
 831
 832        ### Unload sync hooks.
 833        _ = _pre_sync_hooks.pop(plugin_name, None)
 834        _ = _post_sync_hooks.pop(plugin_name, None)
 835
 836        ### Unload API endpoints and pages.
 837        _ = _dash_plugins.pop(plugin_name, None)
 838        web_page_funcs = _plugins_web_pages.pop(plugin_name, None) or []
 839        page_groups_to_pop = []
 840        for page_group, page_functions in _plugin_endpoints_to_pages.items():
 841            page_functions_to_pop = [
 842                page_str
 843                for page_str, page_payload in page_functions.items()
 844                if page_payload.get('function', None) in web_page_funcs
 845            ]
 846            for page_str in page_functions_to_pop:
 847                page_functions.pop(page_str, None)
 848            if not page_functions:
 849                page_groups_to_pop.append(page_group)
 850        
 851        for page_group in page_groups_to_pop:
 852            _plugin_endpoints_to_pages.pop(page_group, None)
 853
 854        ### Remove all but injected symlinks.
 855        if remove_symlinks:
 856            dir_symlink_path = PLUGINS_RESOURCES_PATH / plugin_name
 857            dir_symlink_injected_path = PLUGINS_INJECTED_RESOURCES_PATH / plugin_name
 858            file_symlink_path = PLUGINS_RESOURCES_PATH / f"{plugin_name}.py"
 859            file_symlink_injected_path = PLUGINS_INJECTED_RESOURCES_PATH / f"{plugin_name}.py"
 860
 861            try:
 862                if dir_symlink_path.exists() and not dir_symlink_injected_path.exists():
 863                    dir_symlink_path.unlink()
 864            except Exception:
 865                pass
 866
 867            try:
 868                if file_symlink_path.exists() and not file_symlink_injected_path.exists():
 869                    file_symlink_path.unlink()
 870            except Exception:
 871                pass
 872
 873
 874def reload_plugins(plugins: Optional[List[str]] = None, debug: bool = False) -> None:
 875    """
 876    Reload plugins back into memory.
 877
 878    Parameters
 879    ----------
 880    plugins: Optional[List[str]], default None
 881        The plugins to reload. `None` will reload all plugins.
 882
 883    """
 884    global _synced_symlinks
 885    unload_plugins(plugins, debug=debug)
 886    _synced_symlinks = False
 887    sync_plugins_symlinks(debug=debug)
 888    load_plugins(skip_if_loaded=False, debug=debug)
 889
 890
 891def get_plugins(*to_load, try_import: bool = True) -> Union[Tuple[Plugin], Plugin]:
 892    """
 893    Return a list of `Plugin` objects.
 894
 895    Parameters
 896    ----------
 897    to_load:
 898        If specified, only load specific plugins.
 899        Otherwise return all plugins.
 900
 901    try_import: bool, default True
 902        If `True`, allow for plugins to be imported.
 903    """
 904    from meerschaum.config._paths import PLUGINS_RESOURCES_PATH
 905    import os
 906    sync_plugins_symlinks()
 907    _plugins = [
 908        Plugin(name)
 909        for name in (
 910            to_load or [
 911                (
 912                    name if (PLUGINS_RESOURCES_PATH / name).is_dir()
 913                    else name[:-3]
 914                )
 915                for name in os.listdir(PLUGINS_RESOURCES_PATH)
 916                if name != '__init__.py'
 917            ]
 918        )
 919    ]
 920    plugins = tuple(plugin for plugin in _plugins if plugin.is_installed(try_import=try_import))
 921    if len(to_load) == 1:
 922        if len(plugins) == 0:
 923            raise ValueError(f"Plugin '{to_load[0]}' is not installed.")
 924        return plugins[0]
 925    return plugins
 926
 927
 928def get_plugins_names(*to_load, **kw) -> List[str]:
 929    """
 930    Return a list of installed plugins.
 931    """
 932    return [plugin.name for plugin in get_plugins(*to_load, **kw)]
 933
 934
 935def get_plugins_modules(*to_load, **kw) -> List['ModuleType']:
 936    """
 937    Return a list of modules for the installed plugins, or `None` if things break.
 938    """
 939    return [plugin.module for plugin in get_plugins(*to_load, **kw)]
 940
 941
 942def get_data_plugins() -> List[Plugin]:
 943    """
 944    Only return the modules of plugins with either `fetch()` or `sync()` functions.
 945    """
 946    import inspect
 947    plugins = get_plugins()
 948    data_names = {'sync', 'fetch'}
 949    data_plugins = []
 950    for plugin in plugins:
 951        for name, ob in inspect.getmembers(plugin.module):
 952            if not inspect.isfunction(ob):
 953                continue
 954            if name not in data_names:
 955                continue
 956            data_plugins.append(plugin)
 957    return data_plugins
 958
 959
 960def add_plugin_argument(*args, **kwargs) -> None:
 961    """
 962    Add argparse arguments under the 'Plugins options' group.
 963    Takes the same parameters as the regular argparse `add_argument()` function.
 964
 965    Examples
 966    --------
 967    >>> add_plugin_argument('--foo', type=int, help="This is my help text!")
 968    >>> 
 969    """
 970    from meerschaum._internal.arguments._parser import groups, _seen_plugin_args, parser
 971    from meerschaum.utils.warnings import warn
 972    _parent_plugin_name = _get_parent_plugin()
 973    title = f"Plugin '{_parent_plugin_name}' options" if _parent_plugin_name else 'Custom options'
 974    group_key = 'plugin_' + (_parent_plugin_name or '')
 975    if group_key not in groups:
 976        groups[group_key] = parser.add_argument_group(
 977            title = title,
 978        )
 979        _seen_plugin_args[group_key] = set()
 980    try:
 981        if str(args) not in _seen_plugin_args[group_key]:
 982            groups[group_key].add_argument(*args, **kwargs)
 983            _seen_plugin_args[group_key].add(str(args))
 984    except Exception as e:
 985        warn(e)
 986
 987
 988def inject_plugin_path(
 989    plugin_path: pathlib.Path,
 990    plugins_resources_path: Optional[pathlib.Path] = None) -> None:
 991    """
 992    Inject a plugin as a symlink into the internal `plugins` directory.
 993
 994    Parameters
 995    ----------
 996    plugin_path: pathlib.Path
 997        The path to the plugin's source module.
 998    """
 999    from meerschaum.utils.misc import make_symlink
1000    if plugins_resources_path is None:
1001        from meerschaum.config.paths import PLUGINS_RESOURCES_PATH, PLUGINS_INJECTED_RESOURCES_PATH
1002        plugins_resources_path = PLUGINS_RESOURCES_PATH
1003        plugins_injected_resources_path = PLUGINS_INJECTED_RESOURCES_PATH
1004    else:
1005        plugins_injected_resources_path = plugins_resources_path / '.injected'
1006
1007    if plugin_path.is_dir():
1008        plugin_name = plugin_path.name
1009        dest_path = plugins_resources_path / plugin_name
1010        injected_path = plugins_injected_resources_path / plugin_name
1011    elif plugin_path.name == '__init__.py':
1012        plugin_name = plugin_path.parent.name
1013        dest_path = plugins_resources_path / plugin_name
1014        injected_path = plugins_injected_resources_path / plugin_name
1015    elif plugin_path.name.endswith('.py'):
1016        plugin_name = plugin_path.name[:(-1 * len('.py'))]
1017        dest_path = plugins_resources_path / plugin_path.name
1018        injected_path = plugins_injected_resources_path / plugin_path.name
1019    else:
1020        raise ValueError(f"Cannot deduce plugin name from path '{plugin_path}'.")
1021
1022    _injected_plugin_symlinks.add(dest_path)
1023    make_symlink(plugin_path, dest_path)
1024    make_symlink(plugin_path, injected_path)
1025
1026
1027def _get_parent_plugin(stacklevel: Union[int, Tuple[int, ...]] = (1, 2, 3, 4)) -> Union[str, None]:
1028    """If this function is called from outside a Meerschaum plugin, it will return None."""
1029    import inspect
1030    if not isinstance(stacklevel, tuple):
1031        stacklevel = (stacklevel,)
1032
1033    for _level in stacklevel:
1034        try:
1035            parent_globals = inspect.stack()[_level][0].f_globals
1036            global_name = parent_globals.get('__name__', '')
1037            if global_name.startswith('meerschaum.'):
1038                continue
1039            plugin_name = global_name.replace('plugins.', '').split('.')[0]
1040            if plugin_name.startswith('_') or plugin_name == 'importlib':
1041                continue
1042            return plugin_name
1043        except Exception:
1044            continue
1045
1046    return None