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