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