meerschaum.utils.venv

   1#! /usr/bin/env python3
   2# -*- coding: utf-8 -*-
   3# vim:fenc=utf-8
   4
   5"""
   6Manage virtual environments.
   7"""
   8
   9from __future__ import annotations
  10
  11import sys
  12import pathlib
  13
  14from meerschaum.utils.typing import Optional, Union, Dict, List, Tuple
  15from meerschaum.utils.threading import RLock, get_ident
  16
  17__all__ = sorted([
  18    'activate_venv', 'deactivate_venv', 'init_venv',
  19    'inside_venv', 'is_venv_active', 'venv_exec',
  20    'venv_executable', 'venv_exists', 'venv_target_path',
  21    'Venv', 'get_venvs', 'verify_venv', 'get_module_venv', 'ensure_default_venvs',
  22])
  23__pdoc__ = {'Venv': True}
  24
  25LOCKS = {
  26    'sys.path': RLock(),
  27    'active_venvs': RLock(),
  28    'venvs_active_counts': RLock(),
  29}
  30
  31active_venvs = set()
  32active_venvs_counts: Dict[str, int] = {}
  33active_venvs_order: List[Optional[str]] = []
  34threads_active_venvs: Dict[int, 'set[str]'] = {}
  35CREATE_NEW_PROCESS_GROUP = 0x00000200
  36
  37
  38def activate_venv(
  39    venv: Optional[str] = 'mrsm',
  40    color: bool = True,
  41    force: bool = False,
  42    init_if_not_exists: bool = True,
  43    debug: bool = False,
  44    **kw
  45) -> bool:
  46    """
  47    Create a virtual environment (if it doesn't exist) and add it to `sys.path` if necessary.
  48
  49    Parameters
  50    ----------
  51    venv: Optional[str], default 'mrsm'
  52        The virtual environment to activate.
  53
  54    color: bool, default True
  55        If `True`, include color in debug text.
  56
  57    init_if_not_exists: bool, default True
  58        If `True`, create the virtual environment if it does not exist.
  59
  60    force: bool, default False
  61        If `True`, do not exit early even if the venv is currently active.
  62
  63    debug: bool, default False
  64        Verbosity toggle.
  65
  66    Returns
  67    -------
  68    A bool indicating whether the virtual environment was successfully activated.
  69
  70    """
  71    thread_id = get_ident()
  72    import sys
  73    import os
  74    if venv is not None and init_if_not_exists:
  75        init_venv(venv=venv, debug=debug)
  76
  77    with LOCKS['active_venvs']:
  78        base_fallback = False
  79        try:
  80            target_path = venv_target_path(venv, debug=debug, allow_nonexistent=True)
  81        except OSError as error:
  82            import errno
  83            import site
  84            from meerschaum._internal.static import STATIC_CONFIG
  85            if (
  86                venv is not None
  87                or inside_venv()
  88                or os.environ.get(STATIC_CONFIG['environment']['runtime']) == 'portable'
  89                or error.errno not in (errno.EACCES, errno.EROFS)
  90            ):
  91                raise
  92            ### Activating the base interpreter does not install anything. If its user site
  93            ### cannot be created, reuse an existing global site already visible to it.
  94            ### Keep venv_target_path(None) strict for callers that need an install target.
  95            base_sites = set(site.getsitepackages())
  96            target_path = next((
  97                pathlib.Path(path)
  98                for path in sys.path
  99                if path in base_sites and pathlib.Path(path).is_dir()
 100            ), None)
 101            if target_path is None:
 102                raise
 103            base_fallback = True
 104
 105        if thread_id not in threads_active_venvs:
 106            threads_active_venvs[thread_id] = {}
 107        active_venvs.add(venv)
 108        if venv not in threads_active_venvs[thread_id]:
 109            threads_active_venvs[thread_id][venv] = 1
 110        else:
 111            threads_active_venvs[thread_id][venv] += 1
 112
 113        if not target_path.exists() and venv is not None and init_if_not_exists:
 114            init_venv(venv=venv, force=True, debug=debug)
 115
 116        if not target_path.exists():
 117            if init_if_not_exists:
 118                return False
 119            raise EnvironmentError(f"Could not activate virtual environment '{venv}'.")
 120
 121        target = target_path.as_posix()
 122        if (
 123            active_venvs_order
 124            and active_venvs_order[0] == venv
 125            and target in sys.path
 126            and not force
 127        ):
 128            return True
 129
 130        if venv in active_venvs_order or base_fallback:
 131            try:
 132                sys.path.remove(target)
 133            except Exception:
 134                pass
 135            try:
 136                active_venvs_order.remove(venv)
 137            except Exception:
 138                pass
 139
 140        if venv is not None:
 141            sys.path.insert(0, target)
 142        else:
 143            if sys.path and sys.path[0] in (os.getcwd(), ''):
 144                sys.path.insert(1, target)
 145            else:
 146                sys.path.insert(0, target)
 147        try:
 148            active_venvs_order.insert(0, venv)
 149        except Exception:
 150            pass
 151
 152    return True
 153
 154
 155def ensure_default_venvs(debug: bool = False) -> bool:
 156    """
 157    Put the shared `'mrsm'` virtual environment on `sys.path` (when it exists and is not already
 158    there), the way v4 left it for plugin code.
 159
 160    In v4 every process that ran plugin code had imported `meerschaum` and called
 161    `attempt_import(..., venv='mrsm')` long before the first plugin was imported, so the
 162    shared `mrsm` venv (where a dependency-free `pip install meerschaum` puts pandas and
 163    friends on first use) was always visible: a plugin with its own `required` venv could still
 164    `import pandas`. The order v4 produced is: the plugin's venv(s) first, then the base
 165    interpreter's site-packages, then `<root>/venvs/mrsm`, then the rest of `sys.path`. The
 166    plugin venv is activated later (on top), so this activates `'mrsm'` and then the base
 167    interpreter (`venv=None`), exactly as v4's `attempt_import` calls did.
 168
 169    This never creates a virtual environment and never installs anything
 170    (`MRSM_NO_AUTO_INSTALL` is unaffected): without a `mrsm` venv there is nothing to expose.
 171
 172    Returns
 173    -------
 174    A bool indicating whether the `mrsm` venv is on `sys.path` afterwards.
 175    """
 176    import sys
 177    try:
 178        target_path = venv_target_path('mrsm', allow_nonexistent=True, debug=debug)
 179        target = target_path.as_posix()
 180        if not target_path.exists():
 181            return False
 182        with LOCKS['sys.path']:
 183            if target not in sys.path:
 184                activate_venv('mrsm', init_if_not_exists=False, debug=debug)
 185            if target not in sys.path:
 186                return False
 187            try:
 188                base = venv_target_path(None, debug=debug).as_posix()
 189            except (Exception, SystemExit):
 190                return True
 191            ### The base interpreter's site-packages come before the `mrsm` venv (v4:
 192            ### `activate_venv(None)` ran after `activate_venv('mrsm')`).
 193            index = sys.path.index(target)
 194            if base != target and base not in sys.path[:index]:
 195                sys.path.insert(index, base)
 196                active_venvs.add(None)
 197            return True
 198    except (Exception, SystemExit):
 199        return False
 200
 201
 202def deactivate_venv(
 203    venv: str = 'mrsm',
 204    color: bool = True,
 205    debug: bool = False,
 206    previously_active_venvs: Union['set[str]', List[str], None] = None,
 207    force: bool = False,
 208    **kw
 209) -> bool:
 210    """
 211    Remove a virtual environment from `sys.path` (if it's been activated).
 212
 213    Parameters
 214    ----------
 215    venv: str, default 'mrsm'
 216        The virtual environment to deactivate.
 217
 218    color: bool, default True
 219        If `True`, include color in debug text.
 220
 221    debug: bool, default False
 222        Verbosity toggle.
 223
 224    previously_active_venvs: Union[Set[str], List[str], None]
 225        If provided, skip deactivating if a virtual environment is in this iterable.
 226
 227    force: bool, default False
 228        If `True`, forcibly deactivate the virtual environment.
 229        This may cause issues with other threads, so be careful!
 230
 231    Returns
 232    -------
 233    Return a bool indicating whether the virtual environment was successfully deactivated.
 234
 235    """
 236    import sys
 237    thread_id = get_ident()
 238    if venv is None:
 239        if venv in active_venvs:
 240            active_venvs.remove(venv)
 241        return True
 242
 243    if previously_active_venvs and venv in previously_active_venvs and not force:
 244        return True
 245
 246    with LOCKS['active_venvs']:
 247        if venv in threads_active_venvs.get(thread_id, {}):
 248            new_count = threads_active_venvs[thread_id][venv] - 1
 249            if new_count > 0 and not force:
 250                threads_active_venvs[thread_id][venv] = new_count
 251                return True
 252            else:
 253                del threads_active_venvs[thread_id][venv]
 254
 255        if not force:
 256            for other_thread_id, other_venvs in threads_active_venvs.items():
 257                if other_thread_id == thread_id:
 258                    continue
 259                if venv in other_venvs:
 260                    return True
 261        else:
 262            to_delete = [other_thread_id for other_thread_id in threads_active_venvs]
 263            for other_thread_id in to_delete:
 264                del threads_active_venvs[other_thread_id]
 265
 266        if venv in active_venvs:
 267            active_venvs.remove(venv)
 268
 269    if sys.path is None:
 270        return False
 271
 272    target = venv_target_path(venv, allow_nonexistent=True, debug=debug).as_posix()
 273    with LOCKS['sys.path']:
 274        if target in sys.path:
 275            try:
 276                sys.path.remove(target)
 277            except Exception:
 278                pass
 279            try:
 280                active_venvs_order.remove(venv)
 281            except Exception:
 282                pass
 283
 284    return True
 285
 286
 287def is_venv_active(
 288    venv: str = 'mrsm',
 289    color : bool = True,
 290    debug: bool = False
 291) -> bool:
 292    """
 293    Check if a virtual environment is active.
 294
 295    Parameters
 296    ----------
 297    venv: str, default 'mrsm'
 298        The virtual environment to check.
 299
 300    color: bool, default True
 301        If `True`, include color in debug text.
 302
 303    debug: bool, default False
 304        Verbosity toggle.
 305
 306    Returns
 307    -------
 308    A bool indicating whether the virtual environment `venv` is active.
 309
 310    """
 311    return venv in active_venvs
 312
 313
 314verified_venvs = set()
 315_venv_mutation_locks = {}
 316_venv_mutation_locks_guard = RLock()
 317
 318
 319def _locked_venv_mutation(func):
 320    """Serialize creation and verification, including recursive repair, for each environment."""
 321    from functools import wraps
 322    from inspect import signature
 323    call_signature = signature(func)
 324
 325    @wraps(func)
 326    def locked(*args, **kwargs):
 327        import meerschaum.config.paths as paths
 328        from meerschaum.utils.locks import InterProcessLock
 329
 330        arguments = call_signature.bind(*args, **kwargs)
 331        arguments.apply_defaults()
 332        venv = arguments.arguments['venv']
 333        # Already-verified activation performs no mutations and must keep working even
 334        # when its cache directory has become read-only.
 335        if (
 336            func.__name__ == 'init_venv'
 337            and not arguments.arguments['force']
 338            and venv in verified_venvs
 339        ):
 340            return func(*args, **kwargs)
 341
 342        # Keep this distinct from the legacy existence-marker `<venv>.lock`: file locks
 343        # persist after release. Share the same object for recursive init / verify calls.
 344        lock_path = paths.VENVS_CACHE_RESOURCES_PATH / (venv + '.mutation.lock')
 345        lock_key = str(lock_path.absolute())
 346        with _venv_mutation_locks_guard:
 347            thread_lock, process_lock = _venv_mutation_locks.setdefault(
 348                lock_key, (RLock(), InterProcessLock(lock_path)),
 349            )
 350        with thread_lock:
 351            with process_lock:
 352                return func(*args, **kwargs)
 353
 354    return locked
 355
 356
 357@_locked_venv_mutation
 358def verify_venv(
 359    venv: str,
 360    debug: bool = False,
 361) -> None:
 362    """
 363    Verify that the virtual environment matches the expected state.
 364    """
 365    import platform
 366    import os
 367    import sys
 368
 369    import meerschaum.config.paths as paths
 370    from meerschaum.utils.process import run_process
 371    from meerschaum.utils.misc import make_symlink, is_symlink
 372    from meerschaum.utils.warnings import warn
 373
 374    venv_path = paths.VIRTENV_RESOURCES_PATH / venv
 375    bin_path = venv_path / (
 376        'bin' if platform.system() != 'Windows' else "Scripts"
 377    )
 378    current_python_versioned_name = (
 379        'python' + str(sys.version_info.major) + '.' + str(sys.version_info.minor)
 380        + ('' if platform.system() != 'Windows' else '.exe')
 381    )
 382
 383    if not (bin_path / current_python_versioned_name).exists():
 384        init_venv(venv, verify=False, force=True, debug=debug)
 385        current_python_in_venv_path = pathlib.Path(venv_executable(venv=venv))
 386        current_python_in_sys_path = pathlib.Path(venv_executable(venv=None))
 387        if not current_python_in_venv_path.exists():
 388            if is_symlink(current_python_in_venv_path):
 389                try:
 390                    current_python_in_venv_path.unlink()
 391                except Exception as e:
 392                    print(f"Unable to remove symlink {current_python_in_venv_path}:\n{e}")
 393            try:
 394                make_symlink(current_python_in_sys_path, current_python_in_venv_path)
 395            except Exception:
 396                print(
 397                    f"Unable to create symlink {current_python_in_venv_path} "
 398                    + f"to {current_python_in_sys_path}."
 399                )
 400        files_to_inspect = sorted(os.listdir(bin_path), reverse=True)
 401    else:
 402        files_to_inspect = [current_python_versioned_name]
 403
 404    def get_python_version(python_path: pathlib.Path) -> Union[str, None]:
 405        """
 406        Return the version for the python binary at the given path.
 407        """
 408        proc = None
 409        try:
 410            ### It might be a broken symlink, so skip on errors.
 411            if debug:
 412                print(f"Getting python version for {python_path}")
 413            proc = run_process(
 414                [str(python_path), '-V'],
 415                as_proc=True,
 416                capture_output=True,
 417            )
 418            stdout, stderr = proc.communicate(timeout=1.0)
 419        except Exception as e:
 420            # A loaded host may exceed the probe timeout with a valid interpreter. Reap
 421            # that child, and only remove aliases whose targets are actually missing.
 422            if proc is not None and proc.poll() is None:
 423                proc.kill()
 424                proc.communicate()
 425            if is_symlink(python_path) and not python_path.exists():
 426                try:
 427                    python_path.unlink()
 428                except Exception as _e:
 429                    print(f"Unable to remove broken symlink {python_path}:\n{e}\n{_e}")
 430            return None
 431        return stdout.decode('utf-8').strip().replace('Python ', '')
 432
 433    ### Ensure the versions are symlinked correctly.
 434    for filename in files_to_inspect:
 435        if not filename.startswith('python'):
 436            continue
 437        python_path = bin_path / filename
 438        version = get_python_version(python_path)
 439        if version is None:
 440            continue
 441        try:
 442            major_version = version.split('.', maxsplit=1)[0]
 443            minor_version = version.split('.', maxsplit=2)[1]
 444        except IndexError:
 445            return
 446        python_versioned_name = (
 447            'python' + major_version + '.' + minor_version
 448            + ('' if platform.system() != 'Windows' else '.exe')
 449        )
 450
 451        ### E.g. python3.10 actually links to Python 3.10.
 452        if filename == python_versioned_name:
 453            try:
 454                real_path = pathlib.Path(os.path.realpath(python_path))
 455                real_path_exists = real_path.exists()
 456            except Exception:
 457                real_path_exists = False
 458
 459            if not real_path_exists:
 460                try:
 461                    python_path.unlink()
 462                except Exception:
 463                    pass
 464                init_venv(venv, verify=False, force=True, debug=debug)
 465                if not python_path.exists():
 466                    raise FileNotFoundError(f"Unable to verify Python symlink:\n{python_path}")
 467
 468            if python_path == real_path:
 469                continue
 470
 471            # Already-normalized aliases need no mutation. For indirect aliases, replace
 472            # atomically so another process may keep executing this interpreter throughout.
 473            if python_path.is_symlink() and python_path.readlink() == real_path:
 474                continue
 475            import uuid
 476            temporary_path = python_path.with_name('.' + filename + '.' + uuid.uuid4().hex)
 477            try:
 478                success, msg = make_symlink(real_path, temporary_path)
 479                if success:
 480                    os.replace(temporary_path, python_path)
 481                else:
 482                    warn(msg, color=False)
 483            except OSError as e:
 484                warn(str(e), color=False)
 485            finally:
 486                temporary_path.unlink(missing_ok=True)
 487            continue
 488
 489        python_versioned_path = bin_path / python_versioned_name
 490        if python_versioned_path.exists():
 491            ### Avoid circular symlinks.
 492            if get_python_version(python_versioned_path) == version:
 493                continue
 494        os.replace(python_path, python_versioned_path)
 495
 496
 497tried_virtualenv = False
 498@_locked_venv_mutation
 499def init_venv(
 500    venv: str = 'mrsm',
 501    verify: bool = True,
 502    force: bool = False,
 503    debug: bool = False,
 504) -> bool:
 505    """
 506    Initialize the virtual environment.
 507
 508    Parameters
 509    ----------
 510    venv: str, default 'mrsm'
 511        The name of the virtual environment to create.
 512
 513    verify: bool, default True
 514        If `True`, verify that the virtual environment is in the expected state.
 515
 516    force: bool, default False
 517        If `True`, recreate the virtual environment, even if already initalized.
 518
 519    Returns
 520    -------
 521    A `bool` indicating success.
 522    """
 523    if not force and venv in verified_venvs:
 524        return True
 525    if not force and venv_exists(venv, debug=debug):
 526        if verify:
 527            verify_venv(venv, debug=debug)
 528            verified_venvs.add(venv)
 529        return True
 530
 531    import io
 532    from contextlib import redirect_stdout
 533    import sys
 534    import platform
 535    import os
 536    import shutil
 537    import time
 538
 539    import meerschaum.config.paths as paths
 540    from meerschaum._internal.static import STATIC_CONFIG
 541    from meerschaum.utils.packages import is_uv_enabled
 542
 543    venv_path = paths.VIRTENV_RESOURCES_PATH / venv
 544    vtp = venv_target_path(venv=venv, allow_nonexistent=True, debug=debug)
 545    docker_home_venv_path = pathlib.Path('/home/meerschaum/venvs/mrsm')
 546    lock_path = paths.VENVS_CACHE_RESOURCES_PATH / (venv + '.lock')
 547    work_dir_env_var = STATIC_CONFIG['environment']['work_dir']
 548
 549    def update_lock(active: bool):
 550        try:
 551            if not active:
 552                if debug:
 553                    print(f"Releasing lock: '{lock_path}'")
 554                lock_path.unlink()
 555            else:
 556                if debug:
 557                    print(f"Acquiring lock: '{lock_path}'")
 558                lock_path.touch()
 559        except Exception:
 560            pass
 561
 562    def wait_for_lock():
 563        if platform.system() == 'Windows':
 564            return
 565        max_lock_seconds = 30.0
 566        sleep_message_seconds = 5.0
 567        step_sleep_seconds = 0.1
 568        init_venv_check_start = time.perf_counter()
 569        last_print = init_venv_check_start
 570        while ((time.perf_counter() - init_venv_check_start) < max_lock_seconds):
 571            if not lock_path.exists():
 572                break
 573
 574            now = time.perf_counter()
 575            if debug or (now - last_print) > sleep_message_seconds:
 576                print(f"Lock exists for venv '{venv}', sleeping...")
 577                last_print = now
 578            time.sleep(step_sleep_seconds)
 579        update_lock(False)
 580
 581    if (
 582        not force
 583        and venv == 'mrsm'
 584        and os.environ.get(work_dir_env_var, None) is not None
 585        and docker_home_venv_path.exists()
 586    ):
 587        wait_for_lock()
 588        shutil.move(docker_home_venv_path, venv_path)
 589        if verify:
 590            verify_venv(venv, debug=debug)
 591            verified_venvs.add(venv)
 592        return True
 593
 594    from meerschaum.utils.packages import (
 595        run_python_package, attempt_import, _get_pip_os_env, _get_uv_bin, _auto_install_allowed,
 596    )
 597    global tried_virtualenv
 598    try:
 599        import venv as _venv
 600        uv = _get_uv_bin() if is_uv_enabled() else None
 601        virtualenv = None
 602    except ImportError:
 603        _venv = None
 604        uv = None
 605        virtualenv = None
 606
 607    _venv_success = False
 608    temp_vtp = paths.VENVS_CACHE_RESOURCES_PATH / str(venv)
 609    ### NOTE: Disable site-packages movement for now.
 610    rename_vtp = False and vtp.exists() and not temp_vtp.exists()
 611
 612    wait_for_lock()
 613    update_lock(True)
 614    try:
 615        if rename_vtp:
 616            if debug:
 617                print(f"Moving '{vtp}' to '{temp_vtp}'...")
 618            shutil.move(vtp, temp_vtp)
 619
 620        if uv is not None:
 621            # Plugin imports run in this interpreter. Letting uv choose another version
 622            # makes verification recreate the venv while its creation lock is still held.
 623            _venv_success = run_python_package(
 624                'uv',
 625                [
 626                    'venv', venv_path.as_posix(), '--python', sys.executable,
 627                    '-q', '--no-project', '--allow-existing',
 628                ] + (
 629                    ### `--seed` downloads `pip`, `setuptools` and `wheel` into the new venv.
 630                    ['--seed'] if _auto_install_allowed() else []
 631                ),
 632                venv=None,
 633                env=_get_pip_os_env(),
 634                debug=debug,
 635            ) == 0
 636
 637        if _venv is not None and not _venv_success:
 638            f = io.StringIO()
 639            with redirect_stdout(f):
 640                _venv_success = run_python_package(
 641                    'venv',
 642                    [venv_path.as_posix()] + (
 643                        ['--symlinks']
 644                        if platform.system() != 'Windows'
 645                        else []
 646                    ),
 647                    venv=None, debug=debug
 648                ) == 0
 649            if not _venv_success:
 650                print(f"Please install python3-venv.\n{f.getvalue()}\nFalling back to virtualenv...")
 651            if not venv_exists(venv, debug=debug):
 652                _venv = None
 653        if not _venv_success:
 654            virtualenv = attempt_import(
 655                'virtualenv',
 656                venv=None,
 657                lazy=False,
 658                install=(not tried_virtualenv),
 659                warn=False,
 660                check_update=False,
 661                color=False,
 662                debug=debug,
 663            )
 664            if virtualenv is None:
 665                print(
 666                    "Failed to import `venv` or `virtualenv`! "
 667                    + "Please install `virtualenv` via pip then restart Meerschaum."
 668                )
 669                if rename_vtp and temp_vtp.exists():
 670                    if debug:
 671                        print(f"Moving '{temp_vtp}' back to '{vtp}'...")
 672                    shutil.move(temp_vtp, vtp)
 673                return False
 674
 675            tried_virtualenv = True
 676            try:
 677                python_folder = (
 678                    'python' + str(sys.version_info.major) + '.' + str(sys.version_info.minor)
 679                )
 680                dist_packages_path = (
 681                    paths.VIRTENV_RESOURCES_PATH /
 682                    venv / 'local' / 'lib' / python_folder / 'dist-packages'
 683                )
 684                local_bin_path = paths.VIRTENV_RESOURCES_PATH / venv / 'local' / 'bin'
 685                bin_path = paths.VIRTENV_RESOURCES_PATH / venv / 'bin'
 686                vtp = venv_target_path(venv=venv, allow_nonexistent=True, debug=debug)
 687                if bin_path.exists():
 688                    try:
 689                        shutil.rmtree(bin_path)
 690                    except Exception:
 691                        import traceback
 692                        traceback.print_exc()
 693                virtualenv.cli_run([venv_path.as_posix()])
 694                if dist_packages_path.exists():
 695                    vtp.mkdir(exist_ok=True, parents=True)
 696                    for file_path in dist_packages_path.glob('*'):
 697                        shutil.move(file_path, vtp)
 698                    shutil.rmtree(dist_packages_path)
 699                    #  shutil.move(dist_packages_path, vtp)
 700                    bin_path.mkdir(exist_ok=True, parents=True)
 701                    for file_path in local_bin_path.glob('*'):
 702                        shutil.move(file_path, bin_path)
 703                    #  shutil.move(local_bin_path, bin_path)
 704                    shutil.rmtree(local_bin_path)
 705
 706            except Exception:
 707                import traceback
 708                traceback.print_exc()
 709                if rename_vtp and temp_vtp.exists():
 710                    shutil.move(temp_vtp, vtp)
 711                return False
 712        if rename_vtp and temp_vtp.exists():
 713            if debug:
 714                print(f"Cleanup: move '{temp_vtp}' back to '{vtp}'.")
 715            shutil.move(temp_vtp, vtp)
 716    finally:
 717        update_lock(False)
 718
 719    # The mutation lock is reentrant: verification may recreate an incompatible environment.
 720    if verify:
 721        verify_venv(venv, debug=debug)
 722        verified_venvs.add(venv)
 723    return True
 724
 725
 726def venv_executable(venv: Optional[str] = 'mrsm') -> str:
 727    """
 728    The Python interpreter executable for a given virtual environment.
 729    """
 730    import sys
 731    import platform
 732    import meerschaum.config.paths as paths
 733    return (
 734        sys.executable if venv is None
 735        else str(
 736            paths.VIRTENV_RESOURCES_PATH
 737            / venv
 738            / (
 739                'bin' if platform.system() != 'Windows'
 740                else 'Scripts'
 741            ) / (
 742                'python'
 743                + str(sys.version_info.major)
 744                + '.'
 745                + str(sys.version_info.minor)
 746            )
 747        )
 748    )
 749
 750
 751def venv_exec(
 752    code: str,
 753    venv: Optional[str] = 'mrsm',
 754    env: Optional[Dict[str, str]] = None,
 755    with_extras: bool = False,
 756    as_proc: bool = False,
 757    capture_output: bool = True,
 758    debug: bool = False,
 759) -> Union[bool, Tuple[int, bytes, bytes], 'subprocess.Popen']:
 760    """
 761    Execute Python code in a subprocess via a virtual environment's interpeter.
 762    Return `True` if the code successfully executes, `False` on failure.
 763
 764    Parameters
 765    ----------
 766    code: str
 767        The Python code to excecute.
 768
 769    venv: str, default 'mrsm'
 770        The virtual environment to use to get the path for the Python executable.
 771        If `venv` is `None`, use the default `sys.executable` path.
 772
 773    env: Optional[Dict[str, str]], default None
 774        Optionally specify the environment variables for the subprocess.
 775        Defaults to `os.environ`.
 776
 777    with_extras: bool, default False
 778        If `True`, return a tuple of the exit code, stdout bytes, and stderr bytes.
 779
 780    as_proc: bool, default False
 781        If `True`, return the `subprocess.Popen` object instead of executing.
 782
 783    Returns
 784    -------
 785    By default, return a bool indicating success.
 786    If `as_proc` is `True`, return a `subprocess.Popen` object.
 787    If `with_extras` is `True`, return a tuple of the exit code, stdout bytes, and stderr bytes.
 788
 789    """
 790    import os
 791    import subprocess
 792    import platform
 793    from meerschaum.utils.debug import dprint
 794    from meerschaum.utils.process import _child_processes
 795
 796    executable = venv_executable(venv=venv)
 797    cmd_list = [executable, '-c', code]
 798    if env is None:
 799        env = os.environ
 800    if debug:
 801        dprint(str(cmd_list))
 802    if not with_extras and not as_proc:
 803        return subprocess.call(cmd_list, env=env) == 0
 804
 805    stdout, stderr = (None, None) if not capture_output else (subprocess.PIPE, subprocess.PIPE)
 806    group_kwargs = (
 807        {
 808            'preexec_fn': os.setsid,
 809        } if platform.system() != 'Windows'
 810        else {
 811            'creationflags': CREATE_NEW_PROCESS_GROUP,
 812        }
 813    )
 814    process = subprocess.Popen(
 815        cmd_list,
 816        stdout=stdout,
 817        stderr=stderr,
 818        stdin=sys.stdin,
 819        env=env,
 820        **group_kwargs
 821    )
 822    if as_proc:
 823        _child_processes.append(process)
 824        return process
 825    stdout, stderr = process.communicate()
 826    exit_code = process.returncode
 827    return exit_code, stdout, stderr
 828
 829
 830def venv_exists(venv: Union[str, None], debug: bool = False) -> bool:
 831    """
 832    Determine whether a virtual environment has been created.
 833    """
 834    target_path = venv_target_path(venv, allow_nonexistent=True, debug=debug)
 835    return target_path.exists()
 836
 837
 838def venv_target_path(
 839    venv: Union[str, None],
 840    allow_nonexistent: bool = False,
 841    debug: bool = False,
 842) -> pathlib.Path:
 843    """
 844    Return a virtual environment's site-package path.
 845
 846    Parameters
 847    ----------
 848    venv: Union[str, None]
 849        The virtual environment for which a path should be returned.
 850
 851    allow_nonexistent: bool, default False
 852        If `True`, return a path even if it does not exist.
 853
 854    Returns
 855    -------
 856    The `pathlib.Path` object for the virtual environment's path.
 857
 858    """
 859    import os
 860    import sys
 861    import platform
 862    import site
 863
 864    import meerschaum.config.paths as paths
 865    from meerschaum._internal.static import STATIC_CONFIG
 866
 867    ### Check sys.path for a user-writable site-packages directory.
 868    if venv is None:
 869
 870        ### Return the known value for the portable environment.
 871        environment_runtime = STATIC_CONFIG['environment']['runtime']
 872        if os.environ.get(environment_runtime, None) == 'portable':
 873            python_version_folder = (
 874                'python' + str(sys.version_info.major) + '.' + str(sys.version_info.minor)
 875            )
 876            executable_path = pathlib.Path(sys.executable)
 877            site_packages_path = (
 878                (
 879                    executable_path.parent.parent / 'lib' / python_version_folder / 'site-packages'
 880                ) if platform.system() != 'Windows' else (
 881                    executable_path.parent / 'Lib' / 'site-packages'
 882                )
 883            )
 884            if not site_packages_path.exists():
 885                raise EnvironmentError(f"Could not find '{site_packages_path}'. Does it exist?")
 886            return site_packages_path
 887
 888        if not inside_venv():
 889            user_site_packages = site.getusersitepackages()
 890            if user_site_packages is None:
 891                raise EnvironmentError("Could not determine user site packages.")
 892
 893            site_path = pathlib.Path(user_site_packages)
 894            if not site_path.exists():
 895
 896                ### Windows does not have `os.geteuid()`.
 897                if platform.system() == 'Windows' or os.geteuid() != 0:
 898                    site_path.mkdir(parents=True, exist_ok=True)
 899                    return site_path
 900
 901                ### Allow for dist-level paths (running as root).
 902                for possible_dist in site.getsitepackages():
 903                    dist_path = pathlib.Path(possible_dist)
 904                    if not dist_path.exists():
 905                        continue
 906                    return dist_path
 907                
 908                raise EnvironmentError("Could not determine the dist-packages directory.")
 909
 910            return site_path
 911
 912    venv_root_path = (
 913        (paths.VIRTENV_RESOURCES_PATH / venv)
 914        if venv is not None
 915        else pathlib.Path(sys.prefix)
 916    )
 917    target_path = venv_root_path
 918
 919    ### Ensure 'lib' or 'Lib' exists.
 920    lib = 'lib' if platform.system() != 'Windows' else 'Lib'
 921    if not allow_nonexistent:
 922        if not venv_root_path.exists() or lib not in os.listdir(venv_root_path):
 923            print(f"Failed to find lib directory for virtual environment '{venv}'.")
 924            import traceback
 925            traceback.print_stack()
 926            sys.exit(1)
 927    target_path = target_path / lib
 928
 929    ### Check if a 'python3.x' folder exists.
 930    python_folder = 'python' + str(sys.version_info.major) + '.' + str(sys.version_info.minor)
 931    if target_path.exists():
 932        target_path = (
 933            (target_path / python_folder)
 934            if python_folder in os.listdir(target_path)
 935            else target_path
 936        )
 937    else:
 938        target_path = (
 939            (target_path / python_folder)
 940            if platform.system() != 'Windows'
 941            else target_path
 942        )
 943
 944    ### Ensure 'site-packages' exists.
 945    if allow_nonexistent or 'site-packages' in os.listdir(target_path): ### Windows
 946        target_path = target_path / 'site-packages'
 947    else:
 948        import traceback
 949        traceback.print_stack()
 950        print(f"Failed to find site-packages directory for virtual environment '{venv}'.")
 951        print("This may be because you are using a different Python version.")
 952        print("Try deleting the following directory and restarting Meerschaum:")
 953        print(paths.VIRTENV_RESOURCES_PATH)
 954        sys.exit(1)
 955
 956    return target_path
 957
 958
 959def inside_venv() -> bool:
 960    """
 961    Determine whether current Python interpreter is running inside a virtual environment.
 962    """
 963    import sys
 964    return (
 965        hasattr(sys, 'real_prefix') or (
 966            hasattr(sys, 'base_prefix')
 967                and sys.base_prefix != sys.prefix
 968        )
 969    )
 970
 971
 972def get_venvs() -> List[str]:
 973    """
 974    Return a list of all the virtual environments.
 975    """
 976    import os
 977    import meerschaum.config.paths as paths
 978    venvs = []
 979    for filename in os.listdir(paths.VIRTENV_RESOURCES_PATH):
 980        path = paths.VIRTENV_RESOURCES_PATH / filename
 981        if not path.is_dir():
 982            continue
 983        if not venv_exists(filename):
 984            continue
 985        venvs.append(filename)
 986    return venvs
 987
 988
 989def get_module_venv(module) -> Union[str, None]:
 990    """
 991    Return the virtual environment where an imported module is installed.
 992
 993    Parameters
 994    ----------
 995    module: ModuleType
 996        The imported module to inspect.
 997
 998    Returns
 999    -------
1000    The name of a venv or `None`.
1001    """
1002    import meerschaum.config.paths as paths
1003    module_path = pathlib.Path(module.__file__).resolve()
1004    try:
1005        rel_path = module_path.relative_to(paths.VIRTENV_RESOURCES_PATH)
1006    except ValueError:
1007        return None
1008
1009    return rel_path.as_posix().split('/', maxsplit=1)[0]
1010
1011
1012from meerschaum.utils.venv._Venv import Venv