Module meerschaum.utils.packages

Functions for managing packages and virtual environments reside here.

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8

"""
Functions for managing packages and virtual environments reside here.
"""

from __future__ import annotations

import importlib.util, os, pathlib
from meerschaum.utils.typing import Any, List, SuccessTuple, Optional, Union, Tuple, Dict
from meerschaum.utils.threading import Lock, RLock
from meerschaum.utils.packages._packages import packages, all_packages, get_install_names
from meerschaum.utils.venv import (
    activate_venv,
    deactivate_venv,
    venv_executable,
    venv_exec,
    venv_exists,
    venv_target_path,
    inside_venv,
    Venv,
    init_venv,
)

_import_module = importlib.import_module
_import_hook_venv = None
_locks = {
    '_pkg_resources_get_distribution': RLock(),
    'import_versions': RLock(),
    '_checked_for_updates': RLock(),
    '_is_installed_first_check': RLock(),
}
_checked_for_updates = set()
_is_installed_first_check: Dict[str, bool] = {}

def get_module_path(
        import_name: str,
        venv: Optional[str] = 'mrsm',
        debug: bool = False,
        _try_install_name_on_fail: bool = True,
    ) -> Union[pathlib.Path, None]:
    """
    Get a module's path without importing.
    """
    if debug:
        from meerschaum.utils.debug import dprint
    if not _try_install_name_on_fail:
        install_name = _import_to_install_name(import_name, with_version=False)
        install_name_lower = install_name.lower().replace('-', '_')
        import_name_lower = install_name_lower
    else:
        import_name_lower = import_name.lower().replace('-', '_')
    vtp = venv_target_path(venv, allow_nonexistent=True, debug=debug)
    if not vtp.exists():
        if debug:
            dprint(f"Venv '{venv}' does not exist, cannot import '{import_name}'.", color=False)
        return None
    candidates = []
    for file_name in os.listdir(vtp):
        file_name_lower = file_name.lower().replace('-', '_')
        if not file_name_lower.startswith(import_name_lower):
            continue
        if file_name.endswith('dist_info'):
            continue
        file_path = vtp / file_name

        ### Most likely: Is a directory with __init__.py
        if file_name_lower == import_name_lower and file_path.is_dir():
            init_path = file_path / '__init__.py'
            if init_path.exists():
                candidates.append(init_path)

        ### May be a standalone .py file.
        elif file_name_lower == import_name_lower + '.py':
            candidates.append(file_path)

        ### Compiled wheels (e.g. pyodbc)
        elif file_name_lower.startswith(import_name_lower + '.'):
            candidates.append(file_path)

    if len(candidates) == 1:
        return candidates[0]

    if not candidates:
        if _try_install_name_on_fail:
            return get_module_path(
                import_name, venv=venv, debug=debug,
                _try_install_name_on_fail=False
            )
        if debug:
            dprint(f"No candidates found for '{import_name}' in venv '{venv}'.", color=False)
        return None

    specs_paths = []
    for candidate_path in candidates:
        spec = importlib.util.spec_from_file_location(import_name, str(candidate_path))
        if spec is not None:
            return candidate_path
    
    if debug:
        dprint(
            f"Was unable to find a file location for '{import_name}' in venv '{venv}'.",
            color = False,
        )
    return None


def manually_import_module(
        import_name: str,
        venv: Optional[str] = 'mrsm',
        check_update: bool = True,
        check_pypi: bool = False,
        install: bool = True,
        split: bool = True,
        warn: bool = True,
        color: bool = True,
        debug: bool = False,
        use_sys_modules: bool = True,
    ) -> Union['ModuleType', None]:
    """
    Manually import a module from a virtual environment (or the base environment).

    Parameters
    ----------
    import_name: str
        The name of the module.
        
    venv: Optional[str], default 'mrsm'
        The virtual environment to read from.

    check_update: bool, default True
        If `True`, examine whether the available version of the package meets the required version.

    check_pypi: bool, default False
        If `True`, check PyPI for updates before importing.

    install: bool, default True
        If `True`, install the package if it's not installed or needs an update.

    split: bool, default True
        If `True`, split `import_name` on periods to get the package name.

    warn: bool, default True
        If `True`, raise a warning if the package cannot be imported.

    color: bool, default True
        If `True`, use color output for debug and warning text.

    debug: bool, default False
        Verbosity toggle.

    use_sys_modules: bool, default True
        If `True`, return the module in `sys.modules` if it exists.
        Otherwise continue with manually importing.

    Returns
    -------
    The specified module or `None` if it can't be imported.

    """
    import sys
    _previously_imported = import_name in sys.modules
    if _previously_imported and use_sys_modules:
        return sys.modules[import_name]
    if debug:
        from meerschaum.utils.debug import dprint
    from meerschaum.utils.warnings import warn as warn_function
    import warnings
    root_name = import_name.split('.')[0] if split else import_name
    install_name = _import_to_install_name(root_name)

    root_path = get_module_path(root_name, venv=venv)
    if root_path is None:
        return None
    mod_path = root_path
    if mod_path.is_dir():
        for _dir in import_name.split('.')[:-1]:
            mod_path = mod_path / _dir
            possible_end_module_filename = import_name.split('.')[-1] + '.py'
            try:
                mod_path = (
                    (mod_path / possible_end_module_filename)
                    if possible_end_module_filename in os.listdir(mod_path)
                    else (
                        mod_path / import_name.split('.')[-1] / '__init__.py'
                    )
                )
            except Exception as e:
                mod_path = None

    spec = (
        importlib.util.find_spec(import_name) if mod_path is None or not mod_path.exists()
        else importlib.util.spec_from_file_location(import_name, str(mod_path))
    )
    root_spec = (
        importlib.util.find_spec(root_name) if not root_path.exists()
        else importlib.util.spec_from_file_location(root_name, str(root_path))
    )

    ### Check for updates before importing.
    _version = (
        determine_version(
            pathlib.Path(root_spec.origin),
            import_name=root_name, venv=venv, debug=debug
        ) if root_spec is not None and root_spec.origin is not None else None
    )

    if debug:
        dprint(f'root_name: {root_name}', color=color)
    if _version is not None:
        if debug:
            dprint(f'_version: {_version}', color=color)
        if check_update:
            if need_update(
                None,
                import_name = root_name,
                version = _version,
                check_pypi = check_pypi,
                debug = debug,
            ):
                if install:
                    if not pip_install(
                        root_name,
                        venv = venv,
                        split = False,
                        check_update = check_update,
                        color = color,
                        debug = debug
                    ) and warn:
                        warn_function(
                            f"There's an update available for '{install_name}', "
                            + "but it failed to install. "
                            + "Try installig via Meerschaum with "
                            + "`install packages '{install_name}'`.",
                            ImportWarning,
                            stacklevel = 3,
                            color = False,
                        )
                elif warn:
                    warn_function(
                        f"There's an update available for '{root_name}'.",
                        stack = False,
                        color = False,
                    )
                spec = (
                    importlib.util.find_spec(import_name)
                    if mod_path is None or not mod_path.exists()
                    else importlib.util.spec_from_file_location(import_name, str(mod_path))
                )


    if spec is None:
        try:
            mod = _import_module(import_name)
        except Exception as e:
            mod = None
        return mod

    with Venv(venv, debug=debug):
        mod = importlib.util.module_from_spec(spec)
        old_sys_mod = sys.modules.get(import_name, None)
        sys.modules[import_name] = mod

        try:
            with warnings.catch_warnings():
                warnings.filterwarnings('ignore', 'The NumPy')
                spec.loader.exec_module(mod)
        except Exception as e:
            pass
        mod = _import_module(import_name)
        if old_sys_mod is not None:
            sys.modules[import_name] = old_sys_mod
        else:
            del sys.modules[import_name]
    
    return mod


def _import_to_install_name(import_name: str, with_version: bool = True) -> str:
    """
    Try to translate an import name to an installation name.
    """
    install_name = all_packages.get(import_name, import_name)
    if with_version:
        return install_name
    return get_install_no_version(install_name)


def _import_to_dir_name(import_name: str) -> str:
    """
    Translate an import name to the package name in the sites-packages directory.
    """
    import re
    return re.split(
        f'[<>=\[]', all_packages.get(import_name, import_name)
    )[0].replace('-', '_').lower() 


def _install_to_import_name(install_name: str) -> str:
    """
    Translate an installation name to a package's import name.
    """
    _install_no_version = get_install_no_version(install_name)
    return get_install_names().get(_install_no_version, _install_no_version)


def get_install_no_version(install_name: str) -> str:
    """
    Strip the version information from the install name.
    """
    import re
    return re.split('[\[=<>,! \]]', install_name)[0]


import_versions = {}
def determine_version(
        path: pathlib.Path,
        import_name: Optional[str] = None,
        venv: Optional[str] = 'mrsm',
        search_for_metadata: bool = True,
        split: bool = True,
        warn: bool = False,
        debug: bool = False,
    ) -> Union[str, None]:
    """
    Determine a module's `__version__` string from its filepath.
    
    First it searches for pip metadata, then it attempts to import the module in a subprocess.

    Parameters
    ----------
    path: pathlib.Path
        The file path of the module.

    import_name: Optional[str], default None
        The name of the module. If omitted, it will be determined from the file path.
        Defaults to `None`.

    venv: Optional[str], default 'mrsm'
        The virtual environment of the Python interpreter to use if importing is necessary.

    search_for_metadata: bool, default True
        If `True`, search the pip site_packages directory (assumed to be the parent)
        for the corresponding dist-info directory.

    warn: bool, default True
        If `True`, raise a warning if the module fails to import in the subprocess.

    split: bool, default True
        If `True`, split the determined import name by periods to get the room name.

    Returns
    -------
    The package's version string if available or `None`.
    If multiple versions are found, it will trigger an import in a subprocess.

    """
    with _locks['import_versions']:
        if venv not in import_versions:
            import_versions[venv] = {}
    import re, os
    old_cwd = os.getcwd()
    if debug:
        from meerschaum.utils.debug import dprint
    from meerschaum.utils.warnings import warn as warn_function
    if import_name is None:
        import_name = path.parent.stem if path.stem == '__init__' else path.stem
        import_name = import_name.split('.')[0] if split else import_name
    if import_name in import_versions[venv]:
        return import_versions[venv][import_name]
    _version = None
    module_parent_dir = (
        path.parent.parent if path.stem == '__init__' else path.parent
    ) if path is not None else venv_target_path(venv, debug=debug)

    installed_dir_name = _import_to_dir_name(import_name)
    clean_installed_dir_name = installed_dir_name.lower().replace('-', '_')

    ### First, check if a dist-info directory exists.
    _found_versions = []
    if search_for_metadata:
        for filename in os.listdir(module_parent_dir):
            if not filename.endswith('.dist-info'):
                continue
            filename_lower = filename.lower()
            if not filename_lower.startswith(clean_installed_dir_name + '-'):
                continue
            _v = filename.replace('.dist-info', '').split("-")[-1]
            _found_versions.append(_v)

    if len(_found_versions) == 1:
        _version = _found_versions[0]
        with _locks['import_versions']:
            import_versions[venv][import_name] = _version
        if debug:
            print(f"Found version {_version} for {import_name}.")
        return _found_versions[0]

    if not _found_versions:
        try:
            import importlib.metadata as importlib_metadata
        except ImportError:
            importlib_metadata = attempt_import(
                'importlib_metadata',
                debug=debug, check_update=False, precheck=False,
                color=False, check_is_installed=False, lazy=False,
            )
        try:
            os.chdir(module_parent_dir)
            _version = importlib_metadata.metadata(import_name)['Version']
        except Exception as e:
            _version = None
        finally:
            os.chdir(old_cwd)

        if _version is not None:
            with _locks['import_versions']:
                import_versions[venv][import_name] = _version
            return _version

    if debug:
        print(f'Found multiple versions for {import_name}: {_found_versions}')

    module_parent_dir_str = module_parent_dir.as_posix()

    ### Not a pip package, so let's try importing the module directly (in a subprocess).
    _no_version_str = 'no-version'
    code = (
        f"import sys, importlib; sys.path.insert(0, '{module_parent_dir_str}');\n"
        + f"module = importlib.import_module('{import_name}');\n"
        + "try:\n"
        + "  print(module.__version__ , end='')\n"
        + "except:\n"
        + f"  print('{_no_version_str}', end='')"
    )
    exit_code, stdout_bytes, stderr_bytes = venv_exec(
        code, venv=venv, with_extras=True, debug=debug
    )
    stdout, stderr = stdout_bytes.decode('utf-8'), stderr_bytes.decode('utf-8')
    _version = stdout.split('\n')[-1] if exit_code == 0 else None
    _version = _version if _version != _no_version_str else None

    if _version is None:
        _version = _get_package_metadata(import_name, venv).get('version', None)
    if _version is None and warn:
        warn_function(
            f"Failed to determine a version for '{import_name}':\n{stderr}",
            stack = False
        )

    ### If `__version__` doesn't exist, return `None`.
    import_versions[venv][import_name] = _version
    return _version


def _get_package_metadata(import_name: str, venv: Optional[str]) -> Dict[str, str]:
    """
    Get a package's metadata from pip.
    This is useful for getting a version when no `__version__` is defined
    and multiple versions are installed.

    Parameters
    ----------
    import_name: str
        The package's import or installation name.

    venv: Optional[str]
        The virtual environment which contains the package.

    Returns
    -------
    A dictionary of metadata from pip.
    """
    import re
    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
    install_name = _import_to_install_name(import_name)
    _args = ['show', install_name]
    if venv is not None:
        cache_dir_path = VIRTENV_RESOURCES_PATH / venv / 'cache'
        _args += ['--cache-dir', str(cache_dir_path)]
    proc = run_python_package(
        'pip', _args,
        capture_output=True, as_proc=True, venv=venv, universal_newlines=True,
    )
    outs, errs = proc.communicate()
    lines = outs.split('\n')
    meta = {}
    for line in lines:
        vals = line.split(": ")
        if len(vals) != 2:
            continue
        k, v = vals[0].lower(), vals[1]
        if v and 'UNKNOWN' not in v:
            meta[k] = v
    return meta


def need_update(
        package: Optional['ModuleType'] = None,
        install_name: Optional[str] = None,
        import_name: Optional[str] = None,
        version: Optional[str] = None,
        check_pypi: bool = False,
        split: bool = True,
        color: bool = True,
        debug: bool = False,
        _run_determine_version: bool = True,
    ) -> bool:
    """
    Check if a Meerschaum dependency needs an update.
    Returns a bool for whether or not a package needs to be updated.

    Parameters
    ----------
    package: 'ModuleType'
        The module of the package to be updated.

    install_name: Optional[str], default None
        If provided, use this string to determine the required version.
        Otherwise use the install name defined in `meerschaum.utils.packages._packages`.

    import_name:
        If provided, override the package's `__name__` string.

    version: Optional[str], default None
        If specified, override the package's `__version__` string.

    check_pypi: bool, default False
        If `True`, check pypi.org for updates.
        Defaults to `False`.

    split: bool, default True
        If `True`, split the module's name on periods to detrive the root name.
        Defaults to `True`.

    color: bool, default True
        If `True`, format debug output.
        Defaults to `True`.

    debug: bool, default True
        Verbosity toggle.

    Returns
    -------
    A bool indicating whether the package requires an update.

    """
    if debug:
        from meerschaum.utils.debug import dprint
    from meerschaum.utils.warnings import warn as warn_function
    import re
    root_name = (
        package.__name__.split('.')[0] if split else package.__name__
    ) if import_name is None else (
        import_name.split('.')[0] if split else import_name
    )
    install_name = install_name or _import_to_install_name(root_name)
    with _locks['_checked_for_updates']:
        if install_name in _checked_for_updates:
            return False
        _checked_for_updates.add(install_name)

    _install_no_version = get_install_no_version(install_name)
    if debug:
        dprint(f"_install_no_version: {_install_no_version}", color=color)
    required_version = install_name.replace(_install_no_version, '')
    if ']' in required_version:
        required_version = required_version.split(']')[1]

    ### No minimum version was specified, and we're not going to check PyPI.
    if not required_version and not check_pypi:
        return False

    if debug:
        dprint(f"required_version: {required_version}", color=color)

    try:
        if not version:
            if not _run_determine_version:
                version = determine_version(
                    pathlib.Path(package.__file__),
                    import_name=root_name, warn=False, debug=debug
                )
        if debug:
            dprint(f"version: {version}", color=color)
        if version is None:
            return False
    except Exception as e:
        if debug:
            dprint(str(e), color=color)
            dprint("No version could be determined from the installed package.", color=color)
        return False
    split_version = version.split('.')
    last_part = split_version[-1]
    if len(split_version) == 2:
        version = '.'.join(split_version) + '.0'
    elif 'dev' in last_part or 'rc' in last_part:
        tag = 'dev' if 'dev' in last_part else 'rc'
        last_sep = '-'
        if not last_part.startswith(tag):
            last_part = f'-{tag}'.join(last_part.split(tag))
            last_sep = '.'
        version = '.'.join(split_version[:-1]) + last_sep + last_part
    elif len(split_version) > 3:
        version = '.'.join(split_version[:3])

    packaging_version = attempt_import(
        'packaging.version', check_update=False, lazy=False, debug=debug,
    )

    ### Get semver if necessary
    if required_version:
        semver_path = get_module_path('semver', debug=debug)
        if semver_path is None:
            pip_install(_import_to_install_name('semver'), debug=debug)
        semver = attempt_import('semver', check_update=False, lazy=False, debug=debug)
    if check_pypi:
        ### Check PyPI for updates
        update_checker = attempt_import(
            'update_checker', lazy=False, check_update=False, debug=debug
        )
        checker = update_checker.UpdateChecker()
        result = checker.check(_install_no_version, version)
    else:
        ### Skip PyPI and assume we can't be sure.
        result = None

    ### Compare PyPI's version with our own.
    if result is not None:
        if debug:
            dprint(f"Available version: {result.available_version}", color=color)
            dprint(f"Required version: {required_version}", color=color)

        ### We have a result from PyPI and a stated required version.
        if required_version:
            try:
                return semver.Version.parse(result.available_version).match(required_version)
            except AttributeError as e:
                pip_install(_import_to_install_name('semver'), venv='mrsm', debug=debug)
                semver = manually_import_module('semver', venv='mrsm')
                return semver.Version.parse(version).match(required_version)
            except Exception as e:
                if debug:
                    dprint(f"Failed to match versions with exception:\n{e}", color=color)
                return False

        ### If `check_pypi` and we don't have a required version, check if PyPI's version
        ### is newer than the installed version.
        else:
            return (
                packaging_version.parse(result.available_version) > 
                packaging_version.parse(version)
            )

    ### We might be depending on a prerelease.
    ### Sanity check that the required version is not greater than the installed version.
    try:
        return (
            (not semver.Version.parse(version).match(required_version))
            if required_version else False
        )
    except AttributeError as e:
        pip_install(_import_to_install_name('semver'), venv='mrsm', debug=debug)
        semver = manually_import_module('semver', venv='mrsm', debug=debug)
        return (
            (not semver.Version.parse(version).match(required_version))
            if required_version else False
        )
    except Exception as e:
        print(f"Unable to parse version ({version}) for package '{import_name}'.")
        print(e)
        #  import traceback
        #  traceback.print_exc()
        if debug:
            dprint(e)
        return False
    try:
        return (
            packaging_version.parse(version) > 
            packaging_version.parse(required_version)
        )
    except Exception as e:
        if debug:
            dprint(e)
        return False
    return False


def get_pip(venv: Optional[str] = 'mrsm', debug: bool=False) -> bool:
    """
    Download and run the get-pip.py script.

    Parameters
    ----------
    debug: bool, default False
        Verbosity toggle.

    Returns
    -------
    A bool indicating success.

    """
    import sys, subprocess
    from meerschaum.utils.misc import wget
    from meerschaum.config._paths import CACHE_RESOURCES_PATH
    from meerschaum.config.static import STATIC_CONFIG
    url = STATIC_CONFIG['system']['urls']['get-pip.py']
    dest = CACHE_RESOURCES_PATH / 'get-pip.py'
    try:
        wget(url, dest, color=False, debug=debug)
    except Exception as e:
        print(f"Failed to fetch pip from '{url}'. Please install pip and restart Meerschaum.") 
        sys.exit(1)
    if venv is not None:
        init_venv(venv=venv, debug=debug)
    cmd_list = [venv_executable(venv=venv), dest.as_posix()] 
    return subprocess.call(cmd_list, env=_get_pip_os_env()) == 0


def pip_install(
        *install_names: str,
        args: Optional[List[str]] = None,
        requirements_file_path: Union[pathlib.Path, str, None] = None,
        venv: Optional[str] = 'mrsm',
        split: bool = False,
        check_update: bool = True,
        check_pypi: bool = True,
        check_wheel: bool = True,
        _uninstall: bool = False,
        color: bool = True,
        silent: bool = False,
        debug: bool = False,
    ) -> bool:
    """
    Install packages from PyPI with `pip`.

    Parameters
    ----------
    *install_names: str
        The installation names of packages to be installed.
        This includes version restrictions.
        Use `_import_to_install_name()` to get the predefined `install_name` for a package
        from its import name.
        
    args: Optional[List[str]], default None
        A list of command line arguments to pass to `pip`.
        If not provided, default to `['--upgrade']` if `_uninstall` is `False`, else `[]`.

    requirements_file_path: Optional[pathlib.Path, str], default None
        If provided, append `['-r', '/path/to/requirements.txt']` to `args`.

    venv: str, default 'mrsm'
        The virtual environment to install into.

    split: bool, default False
        If `True`, split on periods and only install the root package name.

    check_update: bool, default True
        If `True`, check if the package requires an update.

    check_pypi: bool, default True
        If `True` and `check_update` is `True`, check PyPI for the latest version.

    check_wheel: bool, default True
        If `True`, check if `wheel` is available.

    _uninstall: bool, default False
        If `True`, uninstall packages instead.

    color: bool, default True
        If `True`, include color in debug text.

    silent: bool, default False
        If `True`, skip printing messages.

    debug: bool, default False
        Verbosity toggle.

    Returns
    -------
    A bool indicating success.

    """
    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
    from meerschaum.utils.warnings import warn
    if args is None:
        args = ['--upgrade'] if not _uninstall else []
    if color:
        ANSI, UNICODE = True, True
    else:
        ANSI, UNICODE = False, False
    if check_wheel:
        have_wheel = venv_contains_package('wheel', venv=venv, debug=debug)

    _args = list(args)
    have_pip = venv_contains_package('pip', venv=venv, debug=debug)
    import sys
    if not have_pip:
        if not get_pip(venv=venv, debug=debug):
            import sys
            minor = sys.version_info.minor
            print(
                "\nFailed to import `pip` and `ensurepip`.\n"
                + "If you are running Ubuntu/Debian, "
                + "you might need to install `python3.{minor}-distutils`:\n\n"
                + f"    sudo apt install python3.{minor}-pip python3.{minor}-venv\n\n"
                + "Please install pip and restart Meerschaum.\n\n"
                + "You can find instructions on installing `pip` here:\n"
                + "https://pip.pypa.io/en/stable/installing/"
            )
            sys.exit(1)
    
    with Venv(venv, debug=debug):
        if venv is not None:
            if '--ignore-installed' not in args and '-I' not in _args and not _uninstall:
                _args += ['--ignore-installed']
            if '--cache-dir' not in args and not _uninstall:
                cache_dir_path = VIRTENV_RESOURCES_PATH / venv / 'cache'
                _args += ['--cache-dir', str(cache_dir_path)]

        if 'pip' not in ' '.join(_args):
            if check_update and not _uninstall:
                pip = attempt_import('pip', venv=venv, install=False, debug=debug, lazy=False)
                if need_update(pip, check_pypi=check_pypi, debug=debug):
                    _args.append(all_packages['pip'])
        
        _args = (['install'] if not _uninstall else ['uninstall']) + _args

        if check_wheel and not _uninstall:
            if not have_wheel:
                if not pip_install(
                    'setuptools', 'wheel',
                    venv = venv,
                    check_update = False, check_pypi = False,
                    check_wheel = False, debug = debug,
                ):
                    warn(
                        f"Failed to install `setuptools` and `wheel` for virtual environment '{venv}'.",
                        color=False,
                    )

        if requirements_file_path is not None:
            _args.append('-r')
            _args.append(str(pathlib.Path(requirements_file_path).resolve()))

        if not ANSI and '--no-color' not in _args:
            _args.append('--no-color')

        if '--no-input' not in _args:
            _args.append('--no-input')

        if _uninstall and '-y' not in _args:
            _args.append('-y')

        if '--no-warn-conflicts' not in _args and not _uninstall:
            _args.append('--no-warn-conflicts')

        if '--disable-pip-version-check' not in _args:
            _args.append('--disable-pip-version-check')

        if '--target' not in _args and '-t' not in _args and not _uninstall:
            if venv is not None:
                _args += ['--target', venv_target_path(venv, debug=debug)]
        elif (
            '--target' not in _args
                and '-t' not in _args
                and not inside_venv()
                and not _uninstall
        ):
            _args += ['--user']

        if debug:
            if '-v' not in _args or '-vv' not in _args or '-vvv' not in _args:
                pass
        else:
            if '-q' not in _args or '-qq' not in _args or '-qqq' not in _args:
                pass

        _packages = [
            (install_name if not _uninstall else get_install_no_version(install_name))
            for install_name in install_names
        ]
        msg = "Installing packages:" if not _uninstall else "Uninstalling packages:"
        for p in _packages:
            msg += f'\n  - {p}'
        if not silent:
            print(msg)

        if not _uninstall:
            for install_name in _packages:
                _install_no_version = get_install_no_version(install_name)
                if _install_no_version in ('pip', 'wheel'):
                    continue
                if not completely_uninstall_package(
                    _install_no_version,
                    venv=venv, debug=debug,
                ):
                    warn(
                        f"Failed to clean up package '{_install_no_version}'.",
                    )

        success = run_python_package(
            'pip',
            _args + _packages,
            venv = venv,
            env = _get_pip_os_env(),
            debug = debug,
        ) == 0

    msg = (
        "Successfully " + ('un' if _uninstall else '') + "installed packages." if success 
        else "Failed to " + ('un' if _uninstall else '') + "install packages."
    )
    if not silent:
        print(msg)
    if debug:
        print('pip ' + ('un' if _uninstall else '') + 'install returned:', success)
    return success


def completely_uninstall_package(
        install_name: str,
        venv: str = 'mrsm',
        debug: bool = False,
    ) -> bool:
    """
    Continue calling `pip uninstall` until a package is completely
    removed from a virtual environment. 
    This is useful for dealing with multiple installed versions of a package.
    """
    attempts = 0
    _install_no_version = get_install_no_version(install_name)
    clean_install_no_version = _install_no_version.lower().replace('-', '_')
    installed_versions = []
    vtp = venv_target_path(venv, allow_nonexistent=True, debug=debug)
    if not vtp.exists():
        return True

    for file_name in os.listdir(vtp):
        if not file_name.endswith('.dist-info'):
            continue
        clean_dist_info = file_name.replace('-', '_').lower()
        if not clean_dist_info.startswith(clean_install_no_version):
            continue
        installed_versions.append(file_name)

    max_attempts = len(installed_versions) + 1
    while attempts < max_attempts:
        if not venv_contains_package(
            _install_to_import_name(_install_no_version),
            venv=venv, debug=debug,
        ):
            return True
        if not pip_uninstall(
            _install_no_version,
            venv=venv,
            silent=(not debug), debug=debug
        ):
            return False
        attempts += 1
    return False


def pip_uninstall(
        *args, **kw
    ) -> bool:
    """
    Uninstall Python packages.
    This function is a wrapper around `pip_install()` but with `_uninstall` enforced as `True`.
    """
    return pip_install(*args, _uninstall=True, **{k: v for k, v in kw.items() if k != '_uninstall'})


def run_python_package(
        package_name: str,
        args: Optional[List[str]] = None,
        venv: Optional[str] = 'mrsm',
        cwd: Optional[str] = None,
        foreground: bool = False,
        as_proc: bool = False,
        capture_output: bool = False,
        debug: bool = False,
        **kw: Any,
    ) -> Union[int, subprocess.Popen]:
    """
    Runs an installed python package.
    E.g. Translates to `/usr/bin/python -m [package]`

    Parameters
    ----------
    package_name: str
        The Python module to be executed.

    args: Optional[List[str]], default None
        Additional command line arguments to be appended after `-m [package]`.

    venv: Optional[str], default 'mrsm'
        If specified, execute the Python interpreter from a virtual environment.

    cwd: Optional[str], default None
        If specified, change directories before starting the process.
        Defaults to `None`.

    as_proc: bool, default False
        If `True`, return a `subprocess.Popen` object.

    capture_output: bool, default False
        If `as_proc` is `True`, capture stdout and stderr.

    foreground: bool, default False
        If `True`, start the subprocess as a foreground process.
        Defaults to `False`.

    kw: Any
        Additional keyword arguments to pass to `meerschaum.utils.process.run_process()`
        and by extension `subprocess.Popen()`.

    Returns
    -------
    Either a return code integer or a `subprocess.Popen` object

    """
    import sys, platform
    import subprocess
    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
    from meerschaum.utils.process import run_process
    from meerschaum.utils.warnings import warn
    if args is None:
        args = []
    old_cwd = os.getcwd()
    if cwd is not None:
        os.chdir(cwd)
    executable = venv_executable(venv=venv)
    command = [executable, '-m', str(package_name)] + [str(a) for a in args]
    import traceback
    if debug:
        print(command, file=sys.stderr)
    try:
        to_return = run_process(
            command,
            foreground = foreground,
            as_proc = as_proc,
            capture_output = capture_output,
            **kw
        )
    except Exception as e:
        msg = f"Failed to execute {command}, will try again:\n{traceback.format_exc()}"
        warn(msg, color=False)
        stdout, stderr = (
            (None, None)
            if not capture_output
            else (subprocess.PIPE, subprocess.PIPE)
        )
        proc = subprocess.Popen(
            command,
            stdout = stdout,
            stderr = stderr,
            env = kw.get('env', os.environ),
        )
        to_return = proc if as_proc else proc.wait()
    except KeyboardInterrupt:
        to_return = 1 if not as_proc else None
    os.chdir(old_cwd)
    return to_return


def attempt_import(
        *names: List[str],
        lazy: bool = True,
        warn: bool = True,
        install: bool = True,
        venv: Optional[str] = 'mrsm',
        precheck: bool = True,
        split: bool = True,
        check_update: bool = False,
        check_pypi: bool = False,
        check_is_installed: bool = True,
        color: bool = True,
        debug: bool = False
    ) -> Union['ModuleType', Tuple['ModuleType']]:
    """
    Raise a warning if packages are not installed; otherwise import and return modules.
    If `lazy` is `True`, return lazy-imported modules.
    
    Returns tuple of modules if multiple names are provided, else returns one module.
    
    Parameters
    ----------
    names: List[str]
        The packages to be imported.

    lazy: bool, default True
        If `True`, lazily load packages.

    warn: bool, default True
        If `True`, raise a warning if a package cannot be imported.

    install: bool, default True
        If `True`, attempt to install a missing package into the designated virtual environment.
        If `check_update` is True, install updates if available.

    venv: Optional[str], default 'mrsm'
        The virtual environment in which to search for packages and to install packages into.

    precheck: bool, default True
        If `True`, attempt to find module before importing (necessary for checking if modules exist
        and retaining lazy imports), otherwise assume lazy is `False`.

    split: bool, default True
        If `True`, split packages' names on `'.'`.

    check_update: bool, default False
        If `True` and `install` is `True`, install updates if the required minimum version
        does not match.

    check_pypi: bool, default False
        If `True` and `check_update` is `True`, check PyPI when determining whether
        an update is required.

    check_is_installed: bool, default True
        If `True`, check if the package is contained in the virtual environment.

    Returns
    -------
    The specified modules. If they're not available and `install` is `True`, it will first
    download them into a virtual environment and return the modules.

    Examples
    --------
    >>> pandas, sqlalchemy = attempt_import('pandas', 'sqlalchemy')
    >>> pandas = attempt_import('pandas')

    """

    import importlib.util

    ### to prevent recursion, check if parent Meerschaum package is being imported
    if names == ('meerschaum',):
        return _import_module('meerschaum')

    if venv == 'mrsm' and _import_hook_venv is not None:
        if debug:
            print(f"Import hook for virtual environment '{_import_hook_venv}' is active.")
        venv = _import_hook_venv

    _warnings = _import_module('meerschaum.utils.warnings')
    warn_function = _warnings.warn

    def do_import(_name: str, **kw) -> Union['ModuleType', None]:
        with Venv(venv=venv, debug=debug):
            ### determine the import method (lazy vs normal)
            from meerschaum.utils.misc import filter_keywords
            import_method = (
                _import_module if not lazy
                else lazy_import
            )
            try:
                mod = import_method(_name, **(filter_keywords(import_method, **kw)))
            except Exception as e:
                if warn:
                    import traceback
                    traceback.print_exception(type(e), e, e.__traceback__)
                    warn_function(
                        f"Failed to import module '{_name}'.\nException:\n{e}",
                        ImportWarning,
                        stacklevel = (5 if lazy else 4),
                        color = False,
                    )
                mod = None
        return mod

    modules = []
    for name in names:
        ### Check if package is a declared dependency.
        root_name = name.split('.')[0] if split else name
        install_name = _import_to_install_name(root_name)

        if install_name is None:
            install_name = root_name
            if warn and root_name != 'plugins':
                warn_function(
                    f"Package '{root_name}' is not declared in meerschaum.utils.packages.",
                    ImportWarning,
                    stacklevel = 3,
                    color = False
                )

        ### Determine if the package exists.
        if precheck is False:
            found_module = (
                do_import(
                    name, debug=debug, warn=False, venv=venv, color=color,
                    check_update=False, check_pypi=False, split=split,
                ) is not None
            )
        else:
            if check_is_installed:
                with _locks['_is_installed_first_check']:
                    if not _is_installed_first_check.get(name, False):
                        package_is_installed = is_installed(
                            name,
                            venv = venv,
                            split = split,
                            debug = debug,
                        )
                        _is_installed_first_check[name] = package_is_installed
                    else:
                        package_is_installed = _is_installed_first_check[name]
            else:
                package_is_installed = _is_installed_first_check.get(
                    name,
                    venv_contains_package(name, venv=venv, split=split, debug=debug)
                )
            found_module = package_is_installed

        if not found_module:
            if install:
                if not pip_install(
                    install_name,
                    venv = venv,
                    split = False,
                    check_update = check_update,
                    color = color,
                    debug = debug
                ) and warn:
                    warn_function(
                        f"Failed to install '{install_name}'.",
                        ImportWarning,
                        stacklevel = 3,
                        color = False,
                    )
            elif warn:
                ### Raise a warning if we can't find the package and install = False.
                warn_function(
                    (f"\n\nMissing package '{name}' from virtual environment '{venv}'; "
                     + "some features will not work correctly."
                     + f"\n\nSet install=True when calling attempt_import.\n"),
                    ImportWarning,
                    stacklevel = 3,
                    color = False,
                )

        ### Do the import. Will be lazy if lazy=True.
        m = do_import(
            name, debug=debug, warn=warn, venv=venv, color=color,
            check_update=check_update, check_pypi=check_pypi, install=install, split=split,
        )
        modules.append(m)

    modules = tuple(modules)
    if len(modules) == 1:
        return modules[0]
    return modules


def lazy_import(
        name: str,
        local_name: str = None,
        **kw
    ) -> meerschaum.utils.packages.lazy_loader.LazyLoader:
    """
    Lazily import a package.
    """
    from meerschaum.utils.packages.lazy_loader import LazyLoader
    if local_name is None:
        local_name = name
    return LazyLoader(
        local_name,
        globals(),
        name,
        **kw
    )


def import_pandas(
        debug: bool = False,
        lazy: bool = False,
        **kw
    ) -> 'ModuleType':
    """
    Quality-of-life function to attempt to import the configured version of `pandas`.
    """
    import sys
    from meerschaum.config import get_config
    pandas_module_name = get_config('system', 'connectors', 'all', 'pandas', patch=True)
    ### NOTE: modin does NOT currently work!
    if pandas_module_name == 'modin':
        pandas_module_name = 'modin.pandas'

    pytz = attempt_import('pytz', debug=debug, lazy=False, **kw)
    pd = attempt_import(pandas_module_name, debug=debug, lazy=lazy, **kw)
    return pd


def import_rich(
        lazy: bool = True,
        debug: bool = False,
        **kw : Any
    ) -> 'ModuleType':
    """
    Quality of life function for importing `rich`.
    """
    from meerschaum.utils.formatting import ANSI, UNICODE
    if not ANSI and not UNICODE:
        return None

    ## need typing_extensions for `from rich import box`
    typing_extensions = attempt_import(
        'typing_extensions', lazy=False, debug=debug
    )
    pygments = attempt_import(
        'pygments', lazy=False,
    )
    rich = attempt_import(
        'rich', lazy=lazy, **kw)
    return rich


def _dash_less_than_2(**kw) -> bool:
    dash = attempt_import('dash', **kw)
    if dash is None:
        return None
    packaging_version = attempt_import('packaging.version', **kw)
    return (
        packaging_version.parse(dash.__version__) < 
        packaging_version.parse('2.0.0')
    )


def import_dcc(warn=False, **kw) -> 'ModuleType':
    """
    Import Dash Core Components (`dcc`).
    """
    return (
        attempt_import('dash_core_components', warn=warn, **kw)
        if _dash_less_than_2(warn=warn, **kw) else attempt_import('dash.dcc', warn=warn, **kw)
    )


def import_html(warn=False, **kw) -> 'ModuleType':
    """
    Import Dash HTML Components (`html`).
    """
    return (
        attempt_import('dash_html_components', warn=warn, **kw)
        if _dash_less_than_2(warn=warn, **kw)
        else attempt_import('dash.html', warn=warn, **kw)
    )


def get_modules_from_package(
        package: 'package',
        names: bool = False,
        recursive: bool = False,
        lazy: bool = False,
        modules_venvs: bool = False,
        debug: bool = False
    ):
    """
    Find and import all modules in a package.
    
    Returns
    -------
    Either list of modules or tuple of lists.

    """
    from os.path import dirname, join, isfile, isdir, basename
    import glob

    pattern = '*' if recursive else '*.py'
    module_names = glob.glob(join(dirname(package.__file__), pattern), recursive=recursive)
    _all = [
        basename(f)[:-3] if isfile(f) else basename(f)
        for f in module_names
            if ((isfile(f) and f.endswith('.py')) or isdir(f))
               and not f.endswith('__init__.py')
               and not f.endswith('__pycache__')
    ]

    if debug:
        from meerschaum.utils.debug import dprint
        dprint(str(_all))
    modules = []
    for module_name in [package.__name__ + "." + mod_name for mod_name in _all]:
        ### there's probably a better way than a try: catch but it'll do for now
        try:
            ### if specified, activate the module's virtual environment before importing.
            ### NOTE: this only considers the filename, so two modules from different packages
            ### may end up sharing virtual environments.
            if modules_venvs:
                activate_venv(module_name.split('.')[-1], debug=debug)
            m = lazy_import(module_name, debug=debug) if lazy else _import_module(module_name)
            modules.append(m)
        except Exception as e:
            if debug:
                dprint(e)
        finally:
            if modules_venvs:
                deactivate_venv(module_name.split('.')[-1], debug=debug)
    if names:
        return _all, modules

    return modules


def import_children(
        package: Optional['ModuleType'] = None,
        package_name: Optional[str] = None,
        types : Optional[List[str]] = None,
        lazy: bool = True,
        recursive: bool = False,
        debug: bool = False
    ) -> List['ModuleType']:
    """
    Import all functions in a package to its `__init__`.

    Parameters
    ----------
    package: Optional[ModuleType], default None
        Package to import its functions into.
        If `None` (default), use parent.

    package_name: Optional[str], default None
        Name of package to import its functions into
        If None (default), use parent.

    types: Optional[List[str]], default None
        Types of members to return.
        Defaults are `['method', 'builtin', 'class', 'function', 'package', 'module']`

    Returns
    -------
    A list of modules.
    """
    import sys, inspect

    if types is None:
        types = ['method', 'builtin', 'function', 'class', 'module']

    ### if package_name and package are None, use parent
    if package is None and package_name is None:
        package_name = inspect.stack()[1][0].f_globals['__name__']

    ### populate package or package_name from other other
    if package is None:
        package = sys.modules[package_name]
    elif package_name is None:
        package_name = package.__name__

    ### Set attributes in sys module version of package.
    ### Kinda like setting a dictionary
    ###   functions[name] = func
    modules = get_modules_from_package(package, recursive=recursive, lazy=lazy, debug=debug)
    _all, members = [], []
    objects = []
    for module in modules:
        _objects = []
        for ob in inspect.getmembers(module):
            for t in types:
                ### ob is a tuple of (name, object)
                if getattr(inspect, 'is' + t)(ob[1]):
                    _objects.append(ob)

        if 'module' in types:
            _objects.append((module.__name__.split('.')[0], module))
        objects += _objects
    for ob in objects:
        setattr(sys.modules[package_name], ob[0], ob[1])
        _all.append(ob[0])
        members.append(ob[1])

    if debug:
        from meerschaum.utils.debug import dprint
        dprint(str(_all))
    ### set __all__ for import *
    setattr(sys.modules[package_name], '__all__', _all)
    return members


def reload_package(
        package: str,
        lazy: bool = False,
        debug: bool = False,
        **kw: Any
    ) -> 'ModuleType':
    """
    Recursively load a package's subpackages, even if they were not previously loaded.
    """
    import pydoc
    if isinstance(package, str):
        package_name = package
    else:
        try:
            package_name = package.__name__
        except Exception as e:
            package_name = str(package)
    return pydoc.safeimport(package_name, forceload=1)


def is_installed(
        import_name: str,
        venv: Optional[str] = 'mrsm',
        split: bool = True,
        debug: bool = False,
    ) -> bool:
    """
    Check whether a package is installed.
    """
    if debug:
        from meerschaum.utils.debug import dprint
    root_name = import_name.split('.')[0] if split else import_name
    import importlib.util
    with Venv(venv, debug=debug):
        try:
            spec_path = pathlib.Path(
                get_module_path(root_name, venv=venv, debug=debug)
                or
                importlib.util.find_spec(root_name).origin
            )
        except (ModuleNotFoundError, ValueError, AttributeError, TypeError) as e:
            spec_path = None

        if debug:
            if spec_path is not None:
                dprint(f"Found a path for '{root_name}':\n{spec_path}", color=False)
            else:
                dprint(f"Could not find a path for '{root_name}':\n{spec_path}", color=False)

        found = (
            not need_update(
                None, import_name = root_name,
                _run_determine_version = False,
                check_pypi = False,
                version = determine_version(
                    spec_path, venv=venv, debug=debug, import_name=root_name
                ),
                debug = debug,
            )
        ) if spec_path is not None else False

    return found


def venv_contains_package(
        import_name: str,
        venv: Optional[str] = 'mrsm',
        split: bool = True,
        debug: bool = False,
    ) -> bool:
    """
    Search the contents of a virtual environment for a package.
    """
    root_name = import_name.split('.')[0] if split else import_name
    return get_module_path(root_name, venv=venv, debug=debug) is not None


def package_venv(package: 'ModuleType') -> Union[str, None]:
    """
    Inspect a package and return the virtual environment in which it presides.
    """
    import os
    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
    if str(VIRTENV_RESOURCES_PATH) not in package.__file__:
        return None
    return package.__file__.split(str(VIRTENV_RESOURCES_PATH))[1].split(os.path.sep)[1]


def ensure_readline() -> 'ModuleType':
    """Make sure that the `readline` package is able to be imported."""
    import sys
    try:
        import readline
    except ImportError:
        readline = None

    if readline is None:
        import platform
        rl_name = "gnureadline" if platform.system() != 'Windows' else "pyreadline3"
        try:
            rl = attempt_import(
                rl_name,
                lazy = False,
                install = True,
                venv = None,
                warn = False,
            )
        except (ImportError, ModuleNotFoundError):
            if not pip_install(rl_name, args=['--upgrade', '--ignore-installed'], venv=None):
                print(f"Unable to import {rl_name}!", file=sys.stderr)
                sys.exit(1)

    sys.modules['readline'] = readline
    return readline

_pkg_resources_get_distribution = None
_custom_distributions = {}
def _monkey_patch_get_distribution(_dist: str, _version: str) -> None:
    """
    Monkey patch `pkg_resources.get_distribution` to allow for importing `flask_compress`.
    """
    import pkg_resources
    from collections import namedtuple
    global _pkg_resources_get_distribution
    with _locks['_pkg_resources_get_distribution']:
        _pkg_resources_get_distribution = pkg_resources.get_distribution
    _custom_distributions[_dist] = _version
    _Dist = namedtuple('_Dist', ['version'])
    def _get_distribution(dist):
        """Hack for flask-compress."""
        if dist in _custom_distributions:
            return _Dist(_custom_distributions[dist])
        return _pkg_resources_get_distribution(dist)
    pkg_resources.get_distribution = _get_distribution


def _get_pip_os_env():
    """
    Return the environment variables context in which `pip` should be run.
    See PEP 668 for why we are overriding the environment.
    """
    import os
    pip_os_env = os.environ.copy()
    pip_os_env.update({
        'PIP_BREAK_SYSTEM_PACKAGES': 'true',
    })
    return pip_os_env

Sub-modules

meerschaum.utils.packages.lazy_loader

A LazyLoader class.

Functions

def attempt_import(*names: List[str], lazy: bool = True, warn: bool = True, install: bool = True, venv: Optional[str] = 'mrsm', precheck: bool = True, split: bool = True, check_update: bool = False, check_pypi: bool = False, check_is_installed: bool = True, color: bool = True, debug: bool = False) ‑> Union['ModuleType', Tuple['ModuleType']]

Raise a warning if packages are not installed; otherwise import and return modules. If lazy is True, return lazy-imported modules.

Returns tuple of modules if multiple names are provided, else returns one module.

Parameters

names : List[str]

The packages to be imported.

lazy : bool, default True

If True, lazily load packages.

warn : bool, default True

If True, raise a warning if a package cannot be imported.

install : bool, default True

If True, attempt to install a missing package into the designated virtual environment. If check_update is True, install updates if available.

venv : Optional[str], default 'mrsm'

The virtual environment in which to search for packages and to install packages into.

precheck : bool, default True

If True, attempt to find module before importing (necessary for checking if modules exist and retaining lazy imports), otherwise assume lazy is False.

split : bool, default True

If True, split packages' names on '.'.

check_update : bool, default False

If True and install is True, install updates if the required minimum version does not match.

check_pypi : bool, default False

If True and check_update is True, check PyPI when determining whether an update is required.

check_is_installed : bool, default True

If True, check if the package is contained in the virtual environment.

Returns

The specified modules. If they're not available and install is True, it will first download them into a virtual environment and return the modules.

Examples

>>> pandas, sqlalchemy = attempt_import('pandas', 'sqlalchemy')
>>> pandas = attempt_import('pandas')

Expand source code

def attempt_import(
        *names: List[str],
        lazy: bool = True,
        warn: bool = True,
        install: bool = True,
        venv: Optional[str] = 'mrsm',
        precheck: bool = True,
        split: bool = True,
        check_update: bool = False,
        check_pypi: bool = False,
        check_is_installed: bool = True,
        color: bool = True,
        debug: bool = False
    ) -> Union['ModuleType', Tuple['ModuleType']]:
    """
    Raise a warning if packages are not installed; otherwise import and return modules.
    If `lazy` is `True`, return lazy-imported modules.
    
    Returns tuple of modules if multiple names are provided, else returns one module.
    
    Parameters
    ----------
    names: List[str]
        The packages to be imported.

    lazy: bool, default True
        If `True`, lazily load packages.

    warn: bool, default True
        If `True`, raise a warning if a package cannot be imported.

    install: bool, default True
        If `True`, attempt to install a missing package into the designated virtual environment.
        If `check_update` is True, install updates if available.

    venv: Optional[str], default 'mrsm'
        The virtual environment in which to search for packages and to install packages into.

    precheck: bool, default True
        If `True`, attempt to find module before importing (necessary for checking if modules exist
        and retaining lazy imports), otherwise assume lazy is `False`.

    split: bool, default True
        If `True`, split packages' names on `'.'`.

    check_update: bool, default False
        If `True` and `install` is `True`, install updates if the required minimum version
        does not match.

    check_pypi: bool, default False
        If `True` and `check_update` is `True`, check PyPI when determining whether
        an update is required.

    check_is_installed: bool, default True
        If `True`, check if the package is contained in the virtual environment.

    Returns
    -------
    The specified modules. If they're not available and `install` is `True`, it will first
    download them into a virtual environment and return the modules.

    Examples
    --------
    >>> pandas, sqlalchemy = attempt_import('pandas', 'sqlalchemy')
    >>> pandas = attempt_import('pandas')

    """

    import importlib.util

    ### to prevent recursion, check if parent Meerschaum package is being imported
    if names == ('meerschaum',):
        return _import_module('meerschaum')

    if venv == 'mrsm' and _import_hook_venv is not None:
        if debug:
            print(f"Import hook for virtual environment '{_import_hook_venv}' is active.")
        venv = _import_hook_venv

    _warnings = _import_module('meerschaum.utils.warnings')
    warn_function = _warnings.warn

    def do_import(_name: str, **kw) -> Union['ModuleType', None]:
        with Venv(venv=venv, debug=debug):
            ### determine the import method (lazy vs normal)
            from meerschaum.utils.misc import filter_keywords
            import_method = (
                _import_module if not lazy
                else lazy_import
            )
            try:
                mod = import_method(_name, **(filter_keywords(import_method, **kw)))
            except Exception as e:
                if warn:
                    import traceback
                    traceback.print_exception(type(e), e, e.__traceback__)
                    warn_function(
                        f"Failed to import module '{_name}'.\nException:\n{e}",
                        ImportWarning,
                        stacklevel = (5 if lazy else 4),
                        color = False,
                    )
                mod = None
        return mod

    modules = []
    for name in names:
        ### Check if package is a declared dependency.
        root_name = name.split('.')[0] if split else name
        install_name = _import_to_install_name(root_name)

        if install_name is None:
            install_name = root_name
            if warn and root_name != 'plugins':
                warn_function(
                    f"Package '{root_name}' is not declared in meerschaum.utils.packages.",
                    ImportWarning,
                    stacklevel = 3,
                    color = False
                )

        ### Determine if the package exists.
        if precheck is False:
            found_module = (
                do_import(
                    name, debug=debug, warn=False, venv=venv, color=color,
                    check_update=False, check_pypi=False, split=split,
                ) is not None
            )
        else:
            if check_is_installed:
                with _locks['_is_installed_first_check']:
                    if not _is_installed_first_check.get(name, False):
                        package_is_installed = is_installed(
                            name,
                            venv = venv,
                            split = split,
                            debug = debug,
                        )
                        _is_installed_first_check[name] = package_is_installed
                    else:
                        package_is_installed = _is_installed_first_check[name]
            else:
                package_is_installed = _is_installed_first_check.get(
                    name,
                    venv_contains_package(name, venv=venv, split=split, debug=debug)
                )
            found_module = package_is_installed

        if not found_module:
            if install:
                if not pip_install(
                    install_name,
                    venv = venv,
                    split = False,
                    check_update = check_update,
                    color = color,
                    debug = debug
                ) and warn:
                    warn_function(
                        f"Failed to install '{install_name}'.",
                        ImportWarning,
                        stacklevel = 3,
                        color = False,
                    )
            elif warn:
                ### Raise a warning if we can't find the package and install = False.
                warn_function(
                    (f"\n\nMissing package '{name}' from virtual environment '{venv}'; "
                     + "some features will not work correctly."
                     + f"\n\nSet install=True when calling attempt_import.\n"),
                    ImportWarning,
                    stacklevel = 3,
                    color = False,
                )

        ### Do the import. Will be lazy if lazy=True.
        m = do_import(
            name, debug=debug, warn=warn, venv=venv, color=color,
            check_update=check_update, check_pypi=check_pypi, install=install, split=split,
        )
        modules.append(m)

    modules = tuple(modules)
    if len(modules) == 1:
        return modules[0]
    return modules

def completely_uninstall_package(install_name: str, venv: str = 'mrsm', debug: bool = False) ‑> bool

Continue calling pip uninstall until a package is completely removed from a virtual environment. This is useful for dealing with multiple installed versions of a package.

Expand source code

def completely_uninstall_package(
        install_name: str,
        venv: str = 'mrsm',
        debug: bool = False,
    ) -> bool:
    """
    Continue calling `pip uninstall` until a package is completely
    removed from a virtual environment. 
    This is useful for dealing with multiple installed versions of a package.
    """
    attempts = 0
    _install_no_version = get_install_no_version(install_name)
    clean_install_no_version = _install_no_version.lower().replace('-', '_')
    installed_versions = []
    vtp = venv_target_path(venv, allow_nonexistent=True, debug=debug)
    if not vtp.exists():
        return True

    for file_name in os.listdir(vtp):
        if not file_name.endswith('.dist-info'):
            continue
        clean_dist_info = file_name.replace('-', '_').lower()
        if not clean_dist_info.startswith(clean_install_no_version):
            continue
        installed_versions.append(file_name)

    max_attempts = len(installed_versions) + 1
    while attempts < max_attempts:
        if not venv_contains_package(
            _install_to_import_name(_install_no_version),
            venv=venv, debug=debug,
        ):
            return True
        if not pip_uninstall(
            _install_no_version,
            venv=venv,
            silent=(not debug), debug=debug
        ):
            return False
        attempts += 1
    return False

def determine_version(path: pathlib.Path, import_name: Optional[str] = None, venv: Optional[str] = 'mrsm', search_for_metadata: bool = True, split: bool = True, warn: bool = False, debug: bool = False) ‑> Optional[str]

Determine a module's __version__ string from its filepath.

First it searches for pip metadata, then it attempts to import the module in a subprocess.

Parameters

path : pathlib.Path

The file path of the module.

import_name : Optional[str], default None

The name of the module. If omitted, it will be determined from the file path. Defaults to None.

venv : Optional[str], default 'mrsm'

The virtual environment of the Python interpreter to use if importing is necessary.

search_for_metadata : bool, default True

If True, search the pip site_packages directory (assumed to be the parent) for the corresponding dist-info directory.

warn : bool, default True

If True, raise a warning if the module fails to import in the subprocess.

split : bool, default True

If True, split the determined import name by periods to get the room name.

Returns

The package's version string if available or None. If multiple versions are found, it will trigger an import in a subprocess.

Expand source code

def determine_version(
        path: pathlib.Path,
        import_name: Optional[str] = None,
        venv: Optional[str] = 'mrsm',
        search_for_metadata: bool = True,
        split: bool = True,
        warn: bool = False,
        debug: bool = False,
    ) -> Union[str, None]:
    """
    Determine a module's `__version__` string from its filepath.
    
    First it searches for pip metadata, then it attempts to import the module in a subprocess.

    Parameters
    ----------
    path: pathlib.Path
        The file path of the module.

    import_name: Optional[str], default None
        The name of the module. If omitted, it will be determined from the file path.
        Defaults to `None`.

    venv: Optional[str], default 'mrsm'
        The virtual environment of the Python interpreter to use if importing is necessary.

    search_for_metadata: bool, default True
        If `True`, search the pip site_packages directory (assumed to be the parent)
        for the corresponding dist-info directory.

    warn: bool, default True
        If `True`, raise a warning if the module fails to import in the subprocess.

    split: bool, default True
        If `True`, split the determined import name by periods to get the room name.

    Returns
    -------
    The package's version string if available or `None`.
    If multiple versions are found, it will trigger an import in a subprocess.

    """
    with _locks['import_versions']:
        if venv not in import_versions:
            import_versions[venv] = {}
    import re, os
    old_cwd = os.getcwd()
    if debug:
        from meerschaum.utils.debug import dprint
    from meerschaum.utils.warnings import warn as warn_function
    if import_name is None:
        import_name = path.parent.stem if path.stem == '__init__' else path.stem
        import_name = import_name.split('.')[0] if split else import_name
    if import_name in import_versions[venv]:
        return import_versions[venv][import_name]
    _version = None
    module_parent_dir = (
        path.parent.parent if path.stem == '__init__' else path.parent
    ) if path is not None else venv_target_path(venv, debug=debug)

    installed_dir_name = _import_to_dir_name(import_name)
    clean_installed_dir_name = installed_dir_name.lower().replace('-', '_')

    ### First, check if a dist-info directory exists.
    _found_versions = []
    if search_for_metadata:
        for filename in os.listdir(module_parent_dir):
            if not filename.endswith('.dist-info'):
                continue
            filename_lower = filename.lower()
            if not filename_lower.startswith(clean_installed_dir_name + '-'):
                continue
            _v = filename.replace('.dist-info', '').split("-")[-1]
            _found_versions.append(_v)

    if len(_found_versions) == 1:
        _version = _found_versions[0]
        with _locks['import_versions']:
            import_versions[venv][import_name] = _version
        if debug:
            print(f"Found version {_version} for {import_name}.")
        return _found_versions[0]

    if not _found_versions:
        try:
            import importlib.metadata as importlib_metadata
        except ImportError:
            importlib_metadata = attempt_import(
                'importlib_metadata',
                debug=debug, check_update=False, precheck=False,
                color=False, check_is_installed=False, lazy=False,
            )
        try:
            os.chdir(module_parent_dir)
            _version = importlib_metadata.metadata(import_name)['Version']
        except Exception as e:
            _version = None
        finally:
            os.chdir(old_cwd)

        if _version is not None:
            with _locks['import_versions']:
                import_versions[venv][import_name] = _version
            return _version

    if debug:
        print(f'Found multiple versions for {import_name}: {_found_versions}')

    module_parent_dir_str = module_parent_dir.as_posix()

    ### Not a pip package, so let's try importing the module directly (in a subprocess).
    _no_version_str = 'no-version'
    code = (
        f"import sys, importlib; sys.path.insert(0, '{module_parent_dir_str}');\n"
        + f"module = importlib.import_module('{import_name}');\n"
        + "try:\n"
        + "  print(module.__version__ , end='')\n"
        + "except:\n"
        + f"  print('{_no_version_str}', end='')"
    )
    exit_code, stdout_bytes, stderr_bytes = venv_exec(
        code, venv=venv, with_extras=True, debug=debug
    )
    stdout, stderr = stdout_bytes.decode('utf-8'), stderr_bytes.decode('utf-8')
    _version = stdout.split('\n')[-1] if exit_code == 0 else None
    _version = _version if _version != _no_version_str else None

    if _version is None:
        _version = _get_package_metadata(import_name, venv).get('version', None)
    if _version is None and warn:
        warn_function(
            f"Failed to determine a version for '{import_name}':\n{stderr}",
            stack = False
        )

    ### If `__version__` doesn't exist, return `None`.
    import_versions[venv][import_name] = _version
    return _version

def ensure_readline() ‑> 'ModuleType'

Make sure that the readline package is able to be imported.

Expand source code

def ensure_readline() -> 'ModuleType':
    """Make sure that the `readline` package is able to be imported."""
    import sys
    try:
        import readline
    except ImportError:
        readline = None

    if readline is None:
        import platform
        rl_name = "gnureadline" if platform.system() != 'Windows' else "pyreadline3"
        try:
            rl = attempt_import(
                rl_name,
                lazy = False,
                install = True,
                venv = None,
                warn = False,
            )
        except (ImportError, ModuleNotFoundError):
            if not pip_install(rl_name, args=['--upgrade', '--ignore-installed'], venv=None):
                print(f"Unable to import {rl_name}!", file=sys.stderr)
                sys.exit(1)

    sys.modules['readline'] = readline
    return readline

def get_install_no_version(install_name: str) ‑> str

Strip the version information from the install name.

Expand source code

def get_install_no_version(install_name: str) -> str:
    """
    Strip the version information from the install name.
    """
    import re
    return re.split('[\[=<>,! \]]', install_name)[0]

def get_module_path(import_name: str, venv: Optional[str] = 'mrsm', debug: bool = False) ‑> Optional[pathlib.Path]

Get a module's path without importing.

Expand source code

def get_module_path(
        import_name: str,
        venv: Optional[str] = 'mrsm',
        debug: bool = False,
        _try_install_name_on_fail: bool = True,
    ) -> Union[pathlib.Path, None]:
    """
    Get a module's path without importing.
    """
    if debug:
        from meerschaum.utils.debug import dprint
    if not _try_install_name_on_fail:
        install_name = _import_to_install_name(import_name, with_version=False)
        install_name_lower = install_name.lower().replace('-', '_')
        import_name_lower = install_name_lower
    else:
        import_name_lower = import_name.lower().replace('-', '_')
    vtp = venv_target_path(venv, allow_nonexistent=True, debug=debug)
    if not vtp.exists():
        if debug:
            dprint(f"Venv '{venv}' does not exist, cannot import '{import_name}'.", color=False)
        return None
    candidates = []
    for file_name in os.listdir(vtp):
        file_name_lower = file_name.lower().replace('-', '_')
        if not file_name_lower.startswith(import_name_lower):
            continue
        if file_name.endswith('dist_info'):
            continue
        file_path = vtp / file_name

        ### Most likely: Is a directory with __init__.py
        if file_name_lower == import_name_lower and file_path.is_dir():
            init_path = file_path / '__init__.py'
            if init_path.exists():
                candidates.append(init_path)

        ### May be a standalone .py file.
        elif file_name_lower == import_name_lower + '.py':
            candidates.append(file_path)

        ### Compiled wheels (e.g. pyodbc)
        elif file_name_lower.startswith(import_name_lower + '.'):
            candidates.append(file_path)

    if len(candidates) == 1:
        return candidates[0]

    if not candidates:
        if _try_install_name_on_fail:
            return get_module_path(
                import_name, venv=venv, debug=debug,
                _try_install_name_on_fail=False
            )
        if debug:
            dprint(f"No candidates found for '{import_name}' in venv '{venv}'.", color=False)
        return None

    specs_paths = []
    for candidate_path in candidates:
        spec = importlib.util.spec_from_file_location(import_name, str(candidate_path))
        if spec is not None:
            return candidate_path
    
    if debug:
        dprint(
            f"Was unable to find a file location for '{import_name}' in venv '{venv}'.",
            color = False,
        )
    return None

def get_modules_from_package(package: "'package'", names: bool = False, recursive: bool = False, lazy: bool = False, modules_venvs: bool = False, debug: bool = False)

Find and import all modules in a package.

Returns

Either list of modules or tuple of lists.

Expand source code

def get_modules_from_package(
        package: 'package',
        names: bool = False,
        recursive: bool = False,
        lazy: bool = False,
        modules_venvs: bool = False,
        debug: bool = False
    ):
    """
    Find and import all modules in a package.
    
    Returns
    -------
    Either list of modules or tuple of lists.

    """
    from os.path import dirname, join, isfile, isdir, basename
    import glob

    pattern = '*' if recursive else '*.py'
    module_names = glob.glob(join(dirname(package.__file__), pattern), recursive=recursive)
    _all = [
        basename(f)[:-3] if isfile(f) else basename(f)
        for f in module_names
            if ((isfile(f) and f.endswith('.py')) or isdir(f))
               and not f.endswith('__init__.py')
               and not f.endswith('__pycache__')
    ]

    if debug:
        from meerschaum.utils.debug import dprint
        dprint(str(_all))
    modules = []
    for module_name in [package.__name__ + "." + mod_name for mod_name in _all]:
        ### there's probably a better way than a try: catch but it'll do for now
        try:
            ### if specified, activate the module's virtual environment before importing.
            ### NOTE: this only considers the filename, so two modules from different packages
            ### may end up sharing virtual environments.
            if modules_venvs:
                activate_venv(module_name.split('.')[-1], debug=debug)
            m = lazy_import(module_name, debug=debug) if lazy else _import_module(module_name)
            modules.append(m)
        except Exception as e:
            if debug:
                dprint(e)
        finally:
            if modules_venvs:
                deactivate_venv(module_name.split('.')[-1], debug=debug)
    if names:
        return _all, modules

    return modules

def get_pip(venv: Optional[str] = 'mrsm', debug: bool = False) ‑> bool

Download and run the get-pip.py script.

Parameters

debug : bool, default False

Verbosity toggle.

Returns

A bool indicating success.

Expand source code

def get_pip(venv: Optional[str] = 'mrsm', debug: bool=False) -> bool:
    """
    Download and run the get-pip.py script.

    Parameters
    ----------
    debug: bool, default False
        Verbosity toggle.

    Returns
    -------
    A bool indicating success.

    """
    import sys, subprocess
    from meerschaum.utils.misc import wget
    from meerschaum.config._paths import CACHE_RESOURCES_PATH
    from meerschaum.config.static import STATIC_CONFIG
    url = STATIC_CONFIG['system']['urls']['get-pip.py']
    dest = CACHE_RESOURCES_PATH / 'get-pip.py'
    try:
        wget(url, dest, color=False, debug=debug)
    except Exception as e:
        print(f"Failed to fetch pip from '{url}'. Please install pip and restart Meerschaum.") 
        sys.exit(1)
    if venv is not None:
        init_venv(venv=venv, debug=debug)
    cmd_list = [venv_executable(venv=venv), dest.as_posix()] 
    return subprocess.call(cmd_list, env=_get_pip_os_env()) == 0

def import_children(package: "Optional['ModuleType']" = None, package_name: Optional[str] = None, types: Optional[List[str]] = None, lazy: bool = True, recursive: bool = False, debug: bool = False) ‑> List['ModuleType']

Import all functions in a package to its __init__.

Parameters

package : Optional[ModuleType], default None

Package to import its functions into. If None (default), use parent.

package_name : Optional[str], default None

Name of package to import its functions into If None (default), use parent.

types : Optional[List[str]], default None

Types of members to return. Defaults are ['method', 'builtin', 'class', 'function', 'package', 'module']

Returns

A list of modules.

Expand source code

def import_children(
        package: Optional['ModuleType'] = None,
        package_name: Optional[str] = None,
        types : Optional[List[str]] = None,
        lazy: bool = True,
        recursive: bool = False,
        debug: bool = False
    ) -> List['ModuleType']:
    """
    Import all functions in a package to its `__init__`.

    Parameters
    ----------
    package: Optional[ModuleType], default None
        Package to import its functions into.
        If `None` (default), use parent.

    package_name: Optional[str], default None
        Name of package to import its functions into
        If None (default), use parent.

    types: Optional[List[str]], default None
        Types of members to return.
        Defaults are `['method', 'builtin', 'class', 'function', 'package', 'module']`

    Returns
    -------
    A list of modules.
    """
    import sys, inspect

    if types is None:
        types = ['method', 'builtin', 'function', 'class', 'module']

    ### if package_name and package are None, use parent
    if package is None and package_name is None:
        package_name = inspect.stack()[1][0].f_globals['__name__']

    ### populate package or package_name from other other
    if package is None:
        package = sys.modules[package_name]
    elif package_name is None:
        package_name = package.__name__

    ### Set attributes in sys module version of package.
    ### Kinda like setting a dictionary
    ###   functions[name] = func
    modules = get_modules_from_package(package, recursive=recursive, lazy=lazy, debug=debug)
    _all, members = [], []
    objects = []
    for module in modules:
        _objects = []
        for ob in inspect.getmembers(module):
            for t in types:
                ### ob is a tuple of (name, object)
                if getattr(inspect, 'is' + t)(ob[1]):
                    _objects.append(ob)

        if 'module' in types:
            _objects.append((module.__name__.split('.')[0], module))
        objects += _objects
    for ob in objects:
        setattr(sys.modules[package_name], ob[0], ob[1])
        _all.append(ob[0])
        members.append(ob[1])

    if debug:
        from meerschaum.utils.debug import dprint
        dprint(str(_all))
    ### set __all__ for import *
    setattr(sys.modules[package_name], '__all__', _all)
    return members

def import_dcc(warn=False, **kw) ‑> 'ModuleType'

Import Dash Core Components (dcc).

Expand source code

def import_dcc(warn=False, **kw) -> 'ModuleType':
    """
    Import Dash Core Components (`dcc`).
    """
    return (
        attempt_import('dash_core_components', warn=warn, **kw)
        if _dash_less_than_2(warn=warn, **kw) else attempt_import('dash.dcc', warn=warn, **kw)
    )

def import_html(warn=False, **kw) ‑> 'ModuleType'

Import Dash HTML Components (html).

Expand source code

def import_html(warn=False, **kw) -> 'ModuleType':
    """
    Import Dash HTML Components (`html`).
    """
    return (
        attempt_import('dash_html_components', warn=warn, **kw)
        if _dash_less_than_2(warn=warn, **kw)
        else attempt_import('dash.html', warn=warn, **kw)
    )

def import_pandas(debug: bool = False, lazy: bool = False, **kw) ‑> 'ModuleType'

Quality-of-life function to attempt to import the configured version of pandas.

Expand source code

def import_pandas(
        debug: bool = False,
        lazy: bool = False,
        **kw
    ) -> 'ModuleType':
    """
    Quality-of-life function to attempt to import the configured version of `pandas`.
    """
    import sys
    from meerschaum.config import get_config
    pandas_module_name = get_config('system', 'connectors', 'all', 'pandas', patch=True)
    ### NOTE: modin does NOT currently work!
    if pandas_module_name == 'modin':
        pandas_module_name = 'modin.pandas'

    pytz = attempt_import('pytz', debug=debug, lazy=False, **kw)
    pd = attempt_import(pandas_module_name, debug=debug, lazy=lazy, **kw)
    return pd

def import_rich(lazy: bool = True, debug: bool = False, **kw: Any) ‑> 'ModuleType'

Quality of life function for importing rich.

Expand source code

def import_rich(
        lazy: bool = True,
        debug: bool = False,
        **kw : Any
    ) -> 'ModuleType':
    """
    Quality of life function for importing `rich`.
    """
    from meerschaum.utils.formatting import ANSI, UNICODE
    if not ANSI and not UNICODE:
        return None

    ## need typing_extensions for `from rich import box`
    typing_extensions = attempt_import(
        'typing_extensions', lazy=False, debug=debug
    )
    pygments = attempt_import(
        'pygments', lazy=False,
    )
    rich = attempt_import(
        'rich', lazy=lazy, **kw)
    return rich

def is_installed(import_name: str, venv: Optional[str] = 'mrsm', split: bool = True, debug: bool = False) ‑> bool

Check whether a package is installed.

Expand source code

def is_installed(
        import_name: str,
        venv: Optional[str] = 'mrsm',
        split: bool = True,
        debug: bool = False,
    ) -> bool:
    """
    Check whether a package is installed.
    """
    if debug:
        from meerschaum.utils.debug import dprint
    root_name = import_name.split('.')[0] if split else import_name
    import importlib.util
    with Venv(venv, debug=debug):
        try:
            spec_path = pathlib.Path(
                get_module_path(root_name, venv=venv, debug=debug)
                or
                importlib.util.find_spec(root_name).origin
            )
        except (ModuleNotFoundError, ValueError, AttributeError, TypeError) as e:
            spec_path = None

        if debug:
            if spec_path is not None:
                dprint(f"Found a path for '{root_name}':\n{spec_path}", color=False)
            else:
                dprint(f"Could not find a path for '{root_name}':\n{spec_path}", color=False)

        found = (
            not need_update(
                None, import_name = root_name,
                _run_determine_version = False,
                check_pypi = False,
                version = determine_version(
                    spec_path, venv=venv, debug=debug, import_name=root_name
                ),
                debug = debug,
            )
        ) if spec_path is not None else False

    return found

def lazy_import(name: str, local_name: str = None, **kw) ‑> LazyLoader

Lazily import a package.

Expand source code

def lazy_import(
        name: str,
        local_name: str = None,
        **kw
    ) -> meerschaum.utils.packages.lazy_loader.LazyLoader:
    """
    Lazily import a package.
    """
    from meerschaum.utils.packages.lazy_loader import LazyLoader
    if local_name is None:
        local_name = name
    return LazyLoader(
        local_name,
        globals(),
        name,
        **kw
    )

def manually_import_module(import_name: str, venv: Optional[str] = 'mrsm', check_update: bool = True, check_pypi: bool = False, install: bool = True, split: bool = True, warn: bool = True, color: bool = True, debug: bool = False, use_sys_modules: bool = True) ‑> Union['ModuleType', None]

Manually import a module from a virtual environment (or the base environment).

Parameters

import_name : str

The name of the module.

venv : Optional[str], default 'mrsm'

The virtual environment to read from.

check_update : bool, default True

If True, examine whether the available version of the package meets the required version.

check_pypi : bool, default False

If True, check PyPI for updates before importing.

install : bool, default True

If True, install the package if it's not installed or needs an update.

split : bool, default True

If True, split import_name on periods to get the package name.

warn : bool, default True

If True, raise a warning if the package cannot be imported.

color : bool, default True

If True, use color output for debug and warning text.

debug : bool, default False

Verbosity toggle.

use_sys_modules : bool, default True

If True, return the module in sys.modules if it exists. Otherwise continue with manually importing.

Returns

The specified module or None if it can't be imported.

Expand source code

def manually_import_module(
        import_name: str,
        venv: Optional[str] = 'mrsm',
        check_update: bool = True,
        check_pypi: bool = False,
        install: bool = True,
        split: bool = True,
        warn: bool = True,
        color: bool = True,
        debug: bool = False,
        use_sys_modules: bool = True,
    ) -> Union['ModuleType', None]:
    """
    Manually import a module from a virtual environment (or the base environment).

    Parameters
    ----------
    import_name: str
        The name of the module.
        
    venv: Optional[str], default 'mrsm'
        The virtual environment to read from.

    check_update: bool, default True
        If `True`, examine whether the available version of the package meets the required version.

    check_pypi: bool, default False
        If `True`, check PyPI for updates before importing.

    install: bool, default True
        If `True`, install the package if it's not installed or needs an update.

    split: bool, default True
        If `True`, split `import_name` on periods to get the package name.

    warn: bool, default True
        If `True`, raise a warning if the package cannot be imported.

    color: bool, default True
        If `True`, use color output for debug and warning text.

    debug: bool, default False
        Verbosity toggle.

    use_sys_modules: bool, default True
        If `True`, return the module in `sys.modules` if it exists.
        Otherwise continue with manually importing.

    Returns
    -------
    The specified module or `None` if it can't be imported.

    """
    import sys
    _previously_imported = import_name in sys.modules
    if _previously_imported and use_sys_modules:
        return sys.modules[import_name]
    if debug:
        from meerschaum.utils.debug import dprint
    from meerschaum.utils.warnings import warn as warn_function
    import warnings
    root_name = import_name.split('.')[0] if split else import_name
    install_name = _import_to_install_name(root_name)

    root_path = get_module_path(root_name, venv=venv)
    if root_path is None:
        return None
    mod_path = root_path
    if mod_path.is_dir():
        for _dir in import_name.split('.')[:-1]:
            mod_path = mod_path / _dir
            possible_end_module_filename = import_name.split('.')[-1] + '.py'
            try:
                mod_path = (
                    (mod_path / possible_end_module_filename)
                    if possible_end_module_filename in os.listdir(mod_path)
                    else (
                        mod_path / import_name.split('.')[-1] / '__init__.py'
                    )
                )
            except Exception as e:
                mod_path = None

    spec = (
        importlib.util.find_spec(import_name) if mod_path is None or not mod_path.exists()
        else importlib.util.spec_from_file_location(import_name, str(mod_path))
    )
    root_spec = (
        importlib.util.find_spec(root_name) if not root_path.exists()
        else importlib.util.spec_from_file_location(root_name, str(root_path))
    )

    ### Check for updates before importing.
    _version = (
        determine_version(
            pathlib.Path(root_spec.origin),
            import_name=root_name, venv=venv, debug=debug
        ) if root_spec is not None and root_spec.origin is not None else None
    )

    if debug:
        dprint(f'root_name: {root_name}', color=color)
    if _version is not None:
        if debug:
            dprint(f'_version: {_version}', color=color)
        if check_update:
            if need_update(
                None,
                import_name = root_name,
                version = _version,
                check_pypi = check_pypi,
                debug = debug,
            ):
                if install:
                    if not pip_install(
                        root_name,
                        venv = venv,
                        split = False,
                        check_update = check_update,
                        color = color,
                        debug = debug
                    ) and warn:
                        warn_function(
                            f"There's an update available for '{install_name}', "
                            + "but it failed to install. "
                            + "Try installig via Meerschaum with "
                            + "`install packages '{install_name}'`.",
                            ImportWarning,
                            stacklevel = 3,
                            color = False,
                        )
                elif warn:
                    warn_function(
                        f"There's an update available for '{root_name}'.",
                        stack = False,
                        color = False,
                    )
                spec = (
                    importlib.util.find_spec(import_name)
                    if mod_path is None or not mod_path.exists()
                    else importlib.util.spec_from_file_location(import_name, str(mod_path))
                )


    if spec is None:
        try:
            mod = _import_module(import_name)
        except Exception as e:
            mod = None
        return mod

    with Venv(venv, debug=debug):
        mod = importlib.util.module_from_spec(spec)
        old_sys_mod = sys.modules.get(import_name, None)
        sys.modules[import_name] = mod

        try:
            with warnings.catch_warnings():
                warnings.filterwarnings('ignore', 'The NumPy')
                spec.loader.exec_module(mod)
        except Exception as e:
            pass
        mod = _import_module(import_name)
        if old_sys_mod is not None:
            sys.modules[import_name] = old_sys_mod
        else:
            del sys.modules[import_name]
    
    return mod

def need_update(package: "Optional['ModuleType']" = None, install_name: Optional[str] = None, import_name: Optional[str] = None, version: Optional[str] = None, check_pypi: bool = False, split: bool = True, color: bool = True, debug: bool = False) ‑> bool

Check if a Meerschaum dependency needs an update. Returns a bool for whether or not a package needs to be updated.

Parameters

package : 'ModuleType'

The module of the package to be updated.

install_name : Optional[str], default None

If provided, use this string to determine the required version. Otherwise use the install name defined in meerschaum.utils.packages._packages.

import_name: If provided, override the package's __name__ string.

version : Optional[str], default None

If specified, override the package's __version__ string.

check_pypi : bool, default False

If True, check pypi.org for updates. Defaults to False.

split : bool, default True

If True, split the module's name on periods to detrive the root name. Defaults to True.

color : bool, default True

If True, format debug output. Defaults to True.

debug : bool, default True

Verbosity toggle.

Returns

A bool indicating whether the package requires an update.

Expand source code

def need_update(
        package: Optional['ModuleType'] = None,
        install_name: Optional[str] = None,
        import_name: Optional[str] = None,
        version: Optional[str] = None,
        check_pypi: bool = False,
        split: bool = True,
        color: bool = True,
        debug: bool = False,
        _run_determine_version: bool = True,
    ) -> bool:
    """
    Check if a Meerschaum dependency needs an update.
    Returns a bool for whether or not a package needs to be updated.

    Parameters
    ----------
    package: 'ModuleType'
        The module of the package to be updated.

    install_name: Optional[str], default None
        If provided, use this string to determine the required version.
        Otherwise use the install name defined in `meerschaum.utils.packages._packages`.

    import_name:
        If provided, override the package's `__name__` string.

    version: Optional[str], default None
        If specified, override the package's `__version__` string.

    check_pypi: bool, default False
        If `True`, check pypi.org for updates.
        Defaults to `False`.

    split: bool, default True
        If `True`, split the module's name on periods to detrive the root name.
        Defaults to `True`.

    color: bool, default True
        If `True`, format debug output.
        Defaults to `True`.

    debug: bool, default True
        Verbosity toggle.

    Returns
    -------
    A bool indicating whether the package requires an update.

    """
    if debug:
        from meerschaum.utils.debug import dprint
    from meerschaum.utils.warnings import warn as warn_function
    import re
    root_name = (
        package.__name__.split('.')[0] if split else package.__name__
    ) if import_name is None else (
        import_name.split('.')[0] if split else import_name
    )
    install_name = install_name or _import_to_install_name(root_name)
    with _locks['_checked_for_updates']:
        if install_name in _checked_for_updates:
            return False
        _checked_for_updates.add(install_name)

    _install_no_version = get_install_no_version(install_name)
    if debug:
        dprint(f"_install_no_version: {_install_no_version}", color=color)
    required_version = install_name.replace(_install_no_version, '')
    if ']' in required_version:
        required_version = required_version.split(']')[1]

    ### No minimum version was specified, and we're not going to check PyPI.
    if not required_version and not check_pypi:
        return False

    if debug:
        dprint(f"required_version: {required_version}", color=color)

    try:
        if not version:
            if not _run_determine_version:
                version = determine_version(
                    pathlib.Path(package.__file__),
                    import_name=root_name, warn=False, debug=debug
                )
        if debug:
            dprint(f"version: {version}", color=color)
        if version is None:
            return False
    except Exception as e:
        if debug:
            dprint(str(e), color=color)
            dprint("No version could be determined from the installed package.", color=color)
        return False
    split_version = version.split('.')
    last_part = split_version[-1]
    if len(split_version) == 2:
        version = '.'.join(split_version) + '.0'
    elif 'dev' in last_part or 'rc' in last_part:
        tag = 'dev' if 'dev' in last_part else 'rc'
        last_sep = '-'
        if not last_part.startswith(tag):
            last_part = f'-{tag}'.join(last_part.split(tag))
            last_sep = '.'
        version = '.'.join(split_version[:-1]) + last_sep + last_part
    elif len(split_version) > 3:
        version = '.'.join(split_version[:3])

    packaging_version = attempt_import(
        'packaging.version', check_update=False, lazy=False, debug=debug,
    )

    ### Get semver if necessary
    if required_version:
        semver_path = get_module_path('semver', debug=debug)
        if semver_path is None:
            pip_install(_import_to_install_name('semver'), debug=debug)
        semver = attempt_import('semver', check_update=False, lazy=False, debug=debug)
    if check_pypi:
        ### Check PyPI for updates
        update_checker = attempt_import(
            'update_checker', lazy=False, check_update=False, debug=debug
        )
        checker = update_checker.UpdateChecker()
        result = checker.check(_install_no_version, version)
    else:
        ### Skip PyPI and assume we can't be sure.
        result = None

    ### Compare PyPI's version with our own.
    if result is not None:
        if debug:
            dprint(f"Available version: {result.available_version}", color=color)
            dprint(f"Required version: {required_version}", color=color)

        ### We have a result from PyPI and a stated required version.
        if required_version:
            try:
                return semver.Version.parse(result.available_version).match(required_version)
            except AttributeError as e:
                pip_install(_import_to_install_name('semver'), venv='mrsm', debug=debug)
                semver = manually_import_module('semver', venv='mrsm')
                return semver.Version.parse(version).match(required_version)
            except Exception as e:
                if debug:
                    dprint(f"Failed to match versions with exception:\n{e}", color=color)
                return False

        ### If `check_pypi` and we don't have a required version, check if PyPI's version
        ### is newer than the installed version.
        else:
            return (
                packaging_version.parse(result.available_version) > 
                packaging_version.parse(version)
            )

    ### We might be depending on a prerelease.
    ### Sanity check that the required version is not greater than the installed version.
    try:
        return (
            (not semver.Version.parse(version).match(required_version))
            if required_version else False
        )
    except AttributeError as e:
        pip_install(_import_to_install_name('semver'), venv='mrsm', debug=debug)
        semver = manually_import_module('semver', venv='mrsm', debug=debug)
        return (
            (not semver.Version.parse(version).match(required_version))
            if required_version else False
        )
    except Exception as e:
        print(f"Unable to parse version ({version}) for package '{import_name}'.")
        print(e)
        #  import traceback
        #  traceback.print_exc()
        if debug:
            dprint(e)
        return False
    try:
        return (
            packaging_version.parse(version) > 
            packaging_version.parse(required_version)
        )
    except Exception as e:
        if debug:
            dprint(e)
        return False
    return False

def package_venv(package: "'ModuleType'") ‑> Union[str, None]

Inspect a package and return the virtual environment in which it presides.

Expand source code

def package_venv(package: 'ModuleType') -> Union[str, None]:
    """
    Inspect a package and return the virtual environment in which it presides.
    """
    import os
    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
    if str(VIRTENV_RESOURCES_PATH) not in package.__file__:
        return None
    return package.__file__.split(str(VIRTENV_RESOURCES_PATH))[1].split(os.path.sep)[1]

def pip_install(*install_names: str, args: Optional[List[str]] = None, requirements_file_path: Union[pathlib.Path, str, None] = None, venv: Optional[str] = 'mrsm', split: bool = False, check_update: bool = True, check_pypi: bool = True, check_wheel: bool = True, color: bool = True, silent: bool = False, debug: bool = False) ‑> bool

Install packages from PyPI with pip.

Parameters

*install_names : str

The installation names of packages to be installed. This includes version restrictions. Use _import_to_install_name() to get the predefined install_name for a package from its import name.

args : Optional[List[str]], default None

A list of command line arguments to pass to pip. If not provided, default to ['--upgrade'] if _uninstall is False, else [].

requirements_file_path : Optional[pathlib.Path, str], default None

If provided, append ['-r', '/path/to/requirements.txt'] to args.

venv : str, default 'mrsm'

The virtual environment to install into.

split : bool, default False

If True, split on periods and only install the root package name.

check_update : bool, default True

If True, check if the package requires an update.

check_pypi : bool, default True

If True and check_update is True, check PyPI for the latest version.

check_wheel : bool, default True

If True, check if wheel is available.

_uninstall : bool, default False

If True, uninstall packages instead.

color : bool, default True

If True, include color in debug text.

silent : bool, default False

If True, skip printing messages.

debug : bool, default False

Verbosity toggle.

Returns

A bool indicating success.

Expand source code

def pip_install(
        *install_names: str,
        args: Optional[List[str]] = None,
        requirements_file_path: Union[pathlib.Path, str, None] = None,
        venv: Optional[str] = 'mrsm',
        split: bool = False,
        check_update: bool = True,
        check_pypi: bool = True,
        check_wheel: bool = True,
        _uninstall: bool = False,
        color: bool = True,
        silent: bool = False,
        debug: bool = False,
    ) -> bool:
    """
    Install packages from PyPI with `pip`.

    Parameters
    ----------
    *install_names: str
        The installation names of packages to be installed.
        This includes version restrictions.
        Use `_import_to_install_name()` to get the predefined `install_name` for a package
        from its import name.
        
    args: Optional[List[str]], default None
        A list of command line arguments to pass to `pip`.
        If not provided, default to `['--upgrade']` if `_uninstall` is `False`, else `[]`.

    requirements_file_path: Optional[pathlib.Path, str], default None
        If provided, append `['-r', '/path/to/requirements.txt']` to `args`.

    venv: str, default 'mrsm'
        The virtual environment to install into.

    split: bool, default False
        If `True`, split on periods and only install the root package name.

    check_update: bool, default True
        If `True`, check if the package requires an update.

    check_pypi: bool, default True
        If `True` and `check_update` is `True`, check PyPI for the latest version.

    check_wheel: bool, default True
        If `True`, check if `wheel` is available.

    _uninstall: bool, default False
        If `True`, uninstall packages instead.

    color: bool, default True
        If `True`, include color in debug text.

    silent: bool, default False
        If `True`, skip printing messages.

    debug: bool, default False
        Verbosity toggle.

    Returns
    -------
    A bool indicating success.

    """
    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
    from meerschaum.utils.warnings import warn
    if args is None:
        args = ['--upgrade'] if not _uninstall else []
    if color:
        ANSI, UNICODE = True, True
    else:
        ANSI, UNICODE = False, False
    if check_wheel:
        have_wheel = venv_contains_package('wheel', venv=venv, debug=debug)

    _args = list(args)
    have_pip = venv_contains_package('pip', venv=venv, debug=debug)
    import sys
    if not have_pip:
        if not get_pip(venv=venv, debug=debug):
            import sys
            minor = sys.version_info.minor
            print(
                "\nFailed to import `pip` and `ensurepip`.\n"
                + "If you are running Ubuntu/Debian, "
                + "you might need to install `python3.{minor}-distutils`:\n\n"
                + f"    sudo apt install python3.{minor}-pip python3.{minor}-venv\n\n"
                + "Please install pip and restart Meerschaum.\n\n"
                + "You can find instructions on installing `pip` here:\n"
                + "https://pip.pypa.io/en/stable/installing/"
            )
            sys.exit(1)
    
    with Venv(venv, debug=debug):
        if venv is not None:
            if '--ignore-installed' not in args and '-I' not in _args and not _uninstall:
                _args += ['--ignore-installed']
            if '--cache-dir' not in args and not _uninstall:
                cache_dir_path = VIRTENV_RESOURCES_PATH / venv / 'cache'
                _args += ['--cache-dir', str(cache_dir_path)]

        if 'pip' not in ' '.join(_args):
            if check_update and not _uninstall:
                pip = attempt_import('pip', venv=venv, install=False, debug=debug, lazy=False)
                if need_update(pip, check_pypi=check_pypi, debug=debug):
                    _args.append(all_packages['pip'])
        
        _args = (['install'] if not _uninstall else ['uninstall']) + _args

        if check_wheel and not _uninstall:
            if not have_wheel:
                if not pip_install(
                    'setuptools', 'wheel',
                    venv = venv,
                    check_update = False, check_pypi = False,
                    check_wheel = False, debug = debug,
                ):
                    warn(
                        f"Failed to install `setuptools` and `wheel` for virtual environment '{venv}'.",
                        color=False,
                    )

        if requirements_file_path is not None:
            _args.append('-r')
            _args.append(str(pathlib.Path(requirements_file_path).resolve()))

        if not ANSI and '--no-color' not in _args:
            _args.append('--no-color')

        if '--no-input' not in _args:
            _args.append('--no-input')

        if _uninstall and '-y' not in _args:
            _args.append('-y')

        if '--no-warn-conflicts' not in _args and not _uninstall:
            _args.append('--no-warn-conflicts')

        if '--disable-pip-version-check' not in _args:
            _args.append('--disable-pip-version-check')

        if '--target' not in _args and '-t' not in _args and not _uninstall:
            if venv is not None:
                _args += ['--target', venv_target_path(venv, debug=debug)]
        elif (
            '--target' not in _args
                and '-t' not in _args
                and not inside_venv()
                and not _uninstall
        ):
            _args += ['--user']

        if debug:
            if '-v' not in _args or '-vv' not in _args or '-vvv' not in _args:
                pass
        else:
            if '-q' not in _args or '-qq' not in _args or '-qqq' not in _args:
                pass

        _packages = [
            (install_name if not _uninstall else get_install_no_version(install_name))
            for install_name in install_names
        ]
        msg = "Installing packages:" if not _uninstall else "Uninstalling packages:"
        for p in _packages:
            msg += f'\n  - {p}'
        if not silent:
            print(msg)

        if not _uninstall:
            for install_name in _packages:
                _install_no_version = get_install_no_version(install_name)
                if _install_no_version in ('pip', 'wheel'):
                    continue
                if not completely_uninstall_package(
                    _install_no_version,
                    venv=venv, debug=debug,
                ):
                    warn(
                        f"Failed to clean up package '{_install_no_version}'.",
                    )

        success = run_python_package(
            'pip',
            _args + _packages,
            venv = venv,
            env = _get_pip_os_env(),
            debug = debug,
        ) == 0

    msg = (
        "Successfully " + ('un' if _uninstall else '') + "installed packages." if success 
        else "Failed to " + ('un' if _uninstall else '') + "install packages."
    )
    if not silent:
        print(msg)
    if debug:
        print('pip ' + ('un' if _uninstall else '') + 'install returned:', success)
    return success

def pip_uninstall(*args, **kw) ‑> bool

Uninstall Python packages. This function is a wrapper around pip_install() but with _uninstall enforced as True.

Expand source code

def pip_uninstall(
        *args, **kw
    ) -> bool:
    """
    Uninstall Python packages.
    This function is a wrapper around `pip_install()` but with `_uninstall` enforced as `True`.
    """
    return pip_install(*args, _uninstall=True, **{k: v for k, v in kw.items() if k != '_uninstall'})

def reload_package(package: str, lazy: bool = False, debug: bool = False, **kw: Any) ‑> 'ModuleType'

Recursively load a package's subpackages, even if they were not previously loaded.

Expand source code

def reload_package(
        package: str,
        lazy: bool = False,
        debug: bool = False,
        **kw: Any
    ) -> 'ModuleType':
    """
    Recursively load a package's subpackages, even if they were not previously loaded.
    """
    import pydoc
    if isinstance(package, str):
        package_name = package
    else:
        try:
            package_name = package.__name__
        except Exception as e:
            package_name = str(package)
    return pydoc.safeimport(package_name, forceload=1)

def run_python_package(package_name: str, args: Optional[List[str]] = None, venv: Optional[str] = 'mrsm', cwd: Optional[str] = None, foreground: bool = False, as_proc: bool = False, capture_output: bool = False, debug: bool = False, **kw: Any) ‑> Union[int, subprocess.Popen]

Runs an installed python package. E.g. Translates to /usr/bin/python -m [package]

Parameters

package_name : str

The Python module to be executed.

args : Optional[List[str]], default None

Additional command line arguments to be appended after -m [package].

venv : Optional[str], default 'mrsm'

If specified, execute the Python interpreter from a virtual environment.

cwd : Optional[str], default None

If specified, change directories before starting the process. Defaults to None.

as_proc : bool, default False

If True, return a subprocess.Popen object.

capture_output : bool, default False

If as_proc is True, capture stdout and stderr.

foreground : bool, default False

If True, start the subprocess as a foreground process. Defaults to False.

kw : Any

Additional keyword arguments to pass to run_process() and by extension subprocess.Popen().

Returns

Either a return code integer or a subprocess.Popen object

Expand source code

def run_python_package(
        package_name: str,
        args: Optional[List[str]] = None,
        venv: Optional[str] = 'mrsm',
        cwd: Optional[str] = None,
        foreground: bool = False,
        as_proc: bool = False,
        capture_output: bool = False,
        debug: bool = False,
        **kw: Any,
    ) -> Union[int, subprocess.Popen]:
    """
    Runs an installed python package.
    E.g. Translates to `/usr/bin/python -m [package]`

    Parameters
    ----------
    package_name: str
        The Python module to be executed.

    args: Optional[List[str]], default None
        Additional command line arguments to be appended after `-m [package]`.

    venv: Optional[str], default 'mrsm'
        If specified, execute the Python interpreter from a virtual environment.

    cwd: Optional[str], default None
        If specified, change directories before starting the process.
        Defaults to `None`.

    as_proc: bool, default False
        If `True`, return a `subprocess.Popen` object.

    capture_output: bool, default False
        If `as_proc` is `True`, capture stdout and stderr.

    foreground: bool, default False
        If `True`, start the subprocess as a foreground process.
        Defaults to `False`.

    kw: Any
        Additional keyword arguments to pass to `meerschaum.utils.process.run_process()`
        and by extension `subprocess.Popen()`.

    Returns
    -------
    Either a return code integer or a `subprocess.Popen` object

    """
    import sys, platform
    import subprocess
    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
    from meerschaum.utils.process import run_process
    from meerschaum.utils.warnings import warn
    if args is None:
        args = []
    old_cwd = os.getcwd()
    if cwd is not None:
        os.chdir(cwd)
    executable = venv_executable(venv=venv)
    command = [executable, '-m', str(package_name)] + [str(a) for a in args]
    import traceback
    if debug:
        print(command, file=sys.stderr)
    try:
        to_return = run_process(
            command,
            foreground = foreground,
            as_proc = as_proc,
            capture_output = capture_output,
            **kw
        )
    except Exception as e:
        msg = f"Failed to execute {command}, will try again:\n{traceback.format_exc()}"
        warn(msg, color=False)
        stdout, stderr = (
            (None, None)
            if not capture_output
            else (subprocess.PIPE, subprocess.PIPE)
        )
        proc = subprocess.Popen(
            command,
            stdout = stdout,
            stderr = stderr,
            env = kw.get('env', os.environ),
        )
        to_return = proc if as_proc else proc.wait()
    except KeyboardInterrupt:
        to_return = 1 if not as_proc else None
    os.chdir(old_cwd)
    return to_return

def venv_contains_package(import_name: str, venv: Optional[str] = 'mrsm', split: bool = True, debug: bool = False) ‑> bool

Search the contents of a virtual environment for a package.

Expand source code

def venv_contains_package(
        import_name: str,
        venv: Optional[str] = 'mrsm',
        split: bool = True,
        debug: bool = False,
    ) -> bool:
    """
    Search the contents of a virtual environment for a package.
    """
    root_name = import_name.split('.')[0] if split else import_name
    return get_module_path(root_name, venv=venv, debug=debug) is not None