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