meerschaum.utils.venv

Manage virtual environments.

  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
 11from meerschaum.utils.typing import Optional, Union, Dict, List, Tuple
 12from meerschaum.utils.threading import RLock, get_ident
 13
 14__all__ = sorted([
 15    'activate_venv', 'deactivate_venv', 'init_venv',
 16    'inside_venv', 'is_venv_active', 'venv_exec',
 17    'venv_executable', 'venv_exists', 'venv_target_path',
 18    'Venv', 'get_venvs', 'verify_venv', 'get_module_venv',
 19])
 20__pdoc__ = {'Venv': True}
 21
 22LOCKS = {
 23    'sys.path': RLock(),
 24    'active_venvs': RLock(),
 25    'venvs_active_counts': RLock(),
 26}
 27
 28active_venvs = set()
 29active_venvs_counts: Dict[str, int] = {}
 30active_venvs_order: List[Optional[str]] = []
 31threads_active_venvs: Dict[int, 'set[str]'] = {}
 32CREATE_NEW_PROCESS_GROUP = 0x00000200
 33
 34
 35def activate_venv(
 36    venv: Optional[str] = 'mrsm',
 37    color: bool = True,
 38    force: bool = False,
 39    debug: bool = False,
 40    **kw
 41) -> bool:
 42    """
 43    Create a virtual environment (if it doesn't exist) and add it to `sys.path` if necessary.
 44
 45    Parameters
 46    ----------
 47    venv: Optional[str], default 'mrsm'
 48        The virtual environment to activate.
 49
 50    color: bool, default True
 51        If `True`, include color in debug text.
 52
 53    force: bool, default False
 54        If `True`, do not exit early even if the venv is currently active.
 55
 56    debug: bool, default False
 57        Verbosity toggle.
 58
 59    Returns
 60    -------
 61    A bool indicating whether the virtual environment was successfully activated.
 62
 63    """
 64    thread_id = get_ident()
 65    if active_venvs_order and active_venvs_order[0] == venv:
 66        if not force:
 67            return True
 68    import sys
 69    import os
 70    if venv is not None:
 71        init_venv(venv=venv, debug=debug)
 72    with LOCKS['active_venvs']:
 73        if thread_id not in threads_active_venvs:
 74            threads_active_venvs[thread_id] = {}
 75        active_venvs.add(venv)
 76        if venv not in threads_active_venvs[thread_id]:
 77            threads_active_venvs[thread_id][venv] = 1
 78        else:
 79            threads_active_venvs[thread_id][venv] += 1
 80
 81        target_path = venv_target_path(venv, debug=debug, allow_nonexistent=True)
 82        if not target_path.exists():
 83            init_venv(venv=venv, force=True, debug=debug)
 84        if not target_path.exists():
 85            raise EnvironmentError(f"Could not activate virtual environment '{venv}'.")
 86        target = target_path.as_posix()
 87
 88        if venv in active_venvs_order:
 89            sys.path.remove(target)
 90            try:
 91                active_venvs_order.remove(venv)
 92            except Exception:
 93                pass
 94        if venv is not None:
 95            sys.path.insert(0, target)
 96        else:
 97            if sys.path and sys.path[0] in (os.getcwd(), ''):
 98                sys.path.insert(1, target)
 99            else:
100                sys.path.insert(0, target)
101        try:
102            active_venvs_order.insert(0, venv)
103        except Exception:
104            pass
105
106    return True
107
108
109def deactivate_venv(
110    venv: str = 'mrsm',
111    color: bool = True,
112    debug: bool = False,
113    previously_active_venvs: Union['set[str]', List[str], None] = None,
114    force: bool = False,
115    **kw
116) -> bool:
117    """
118    Remove a virtual environment from `sys.path` (if it's been activated).
119
120    Parameters
121    ----------
122    venv: str, default 'mrsm'
123        The virtual environment to deactivate.
124
125    color: bool, default True
126        If `True`, include color in debug text.
127
128    debug: bool, default False
129        Verbosity toggle.
130
131    previously_active_venvs: Union[Set[str], List[str], None]
132        If provided, skip deactivating if a virtual environment is in this iterable.
133
134    force: bool, default False
135        If `True`, forcibly deactivate the virtual environment.
136        This may cause issues with other threads, so be careful!
137
138    Returns
139    -------
140    Return a bool indicating whether the virtual environment was successfully deactivated.
141
142    """
143    import sys
144    thread_id = get_ident()
145    if venv is None:
146        if venv in active_venvs:
147            active_venvs.remove(venv)
148        return True
149
150    if previously_active_venvs and venv in previously_active_venvs and not force:
151        return True
152
153    with LOCKS['active_venvs']:
154        if venv in threads_active_venvs.get(thread_id, {}):
155            new_count = threads_active_venvs[thread_id][venv] - 1
156            if new_count > 0 and not force:
157                threads_active_venvs[thread_id][venv] = new_count
158                return True
159            else:
160                del threads_active_venvs[thread_id][venv]
161
162        if not force:
163            for other_thread_id, other_venvs in threads_active_venvs.items():
164                if other_thread_id == thread_id:
165                    continue
166                if venv in other_venvs:
167                    return True
168        else:
169            to_delete = [other_thread_id for other_thread_id in threads_active_venvs]
170            for other_thread_id in to_delete:
171                del threads_active_venvs[other_thread_id]
172
173        if venv in active_venvs:
174            active_venvs.remove(venv)
175
176    if sys.path is None:
177        return False
178
179    target = venv_target_path(venv, allow_nonexistent=True, debug=debug).as_posix()
180    with LOCKS['sys.path']:
181        if target in sys.path:
182            try:
183                sys.path.remove(target)
184            except Exception:
185                pass
186            try:
187                active_venvs_order.remove(venv)
188            except Exception:
189                pass
190
191    return True
192
193
194def is_venv_active(
195    venv: str = 'mrsm',
196    color : bool = True,
197    debug: bool = False
198) -> bool:
199    """
200    Check if a virtual environment is active.
201
202    Parameters
203    ----------
204    venv: str, default 'mrsm'
205        The virtual environment to check.
206
207    color: bool, default True
208        If `True`, include color in debug text.
209
210    debug: bool, default False
211        Verbosity toggle.
212
213    Returns
214    -------
215    A bool indicating whether the virtual environment `venv` is active.
216
217    """
218    return venv in active_venvs
219
220
221verified_venvs = set()
222def verify_venv(
223    venv: str,
224    debug: bool = False,
225) -> None:
226    """
227    Verify that the virtual environment matches the expected state.
228    """
229    import pathlib
230    import platform
231    import os
232    import shutil
233    import sys
234    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
235    from meerschaum.utils.process import run_process
236    from meerschaum.utils.misc import make_symlink, is_symlink
237    from meerschaum.utils.warnings import warn
238
239    venv_path = VIRTENV_RESOURCES_PATH / venv
240    bin_path = venv_path / (
241        'bin' if platform.system() != 'Windows' else "Scripts"
242    )
243    current_python_versioned_name = (
244        'python' + str(sys.version_info.major) + '.' + str(sys.version_info.minor)
245        + ('' if platform.system() != 'Windows' else '.exe')
246    )
247
248    if not (bin_path / current_python_versioned_name).exists():
249        init_venv(venv, verify=False, force=True, debug=debug)
250        current_python_in_venv_path = pathlib.Path(venv_executable(venv=venv))
251        current_python_in_sys_path = pathlib.Path(venv_executable(venv=None))
252        if not current_python_in_venv_path.exists():
253            if is_symlink(current_python_in_venv_path):
254                try:
255                    current_python_in_venv_path.unlink()
256                except Exception as e:
257                    print(f"Unable to remove symlink {current_python_in_venv_path}:\n{e}")
258            try:
259                make_symlink(current_python_in_sys_path, current_python_in_venv_path)
260            except Exception:
261                print(
262                    f"Unable to create symlink {current_python_in_venv_path} "
263                    + f"to {current_python_in_sys_path}."
264                )
265        files_to_inspect = sorted(os.listdir(bin_path), reverse=True)
266    else:
267        files_to_inspect = [current_python_versioned_name]
268
269    def get_python_version(python_path: pathlib.Path) -> Union[str, None]:
270        """
271        Return the version for the python binary at the given path.
272        """
273        try:
274            ### It might be a broken symlink, so skip on errors.
275            if debug:
276                print(f"Getting python version for {python_path}")
277            proc = run_process(
278                [str(python_path), '-V'],
279                as_proc=True,
280                capture_output=True,
281            )
282            stdout, stderr = proc.communicate(timeout=1.0)
283        except Exception as e:
284            ### E.g. the symlink may be broken.
285            if is_symlink(python_path):
286                try:
287                    python_path.unlink()
288                except Exception as _e:
289                    print(f"Unable to remove broken symlink {python_path}:\n{e}\n{_e}")
290            return None
291        return stdout.decode('utf-8').strip().replace('Python ', '')
292
293    ### Ensure the versions are symlinked correctly.
294    for filename in files_to_inspect:
295        if not filename.startswith('python'):
296            continue
297        python_path = bin_path / filename
298        version = get_python_version(python_path)
299        if version is None:
300            continue
301        try:
302            major_version = version.split('.', maxsplit=1)[0]
303            minor_version = version.split('.', maxsplit=2)[1]
304        except IndexError:
305            return
306        python_versioned_name = (
307            'python' + major_version + '.' + minor_version
308            + ('' if platform.system() != 'Windows' else '.exe')
309        )
310
311        ### E.g. python3.10 actually links to Python 3.10.
312        if filename == python_versioned_name:
313            real_path = pathlib.Path(os.path.realpath(python_path))
314            if not real_path.exists():
315                try:
316                    python_path.unlink()
317                except Exception:
318                    pass
319                init_venv(venv, verify=False, force=True, debug=debug)
320                if not python_path.exists():
321                    raise FileNotFoundError(f"Unable to verify Python symlink:\n{python_path}")
322
323            if python_path == real_path:
324                continue
325
326            try:
327                python_path.unlink()
328            except Exception:
329                pass
330            success, msg = make_symlink(real_path, python_path)
331            if not success:
332                warn(msg, color=False)
333            continue
334
335        python_versioned_path = bin_path / python_versioned_name
336        if python_versioned_path.exists():
337            ### Avoid circular symlinks.
338            if get_python_version(python_versioned_path) == version:
339                continue
340            python_versioned_path.unlink()
341        shutil.move(python_path, python_versioned_path)
342
343
344tried_virtualenv = False
345def init_venv(
346    venv: str = 'mrsm',
347    verify: bool = True,
348    force: bool = False,
349    debug: bool = False,
350) -> bool:
351    """
352    Initialize the virtual environment.
353
354    Parameters
355    ----------
356    venv: str, default 'mrsm'
357        The name of the virtual environment to create.
358
359    verify: bool, default True
360        If `True`, verify that the virtual environment is in the expected state.
361
362    force: bool, default False
363        If `True`, recreate the virtual environment, even if already initalized.
364
365    Returns
366    -------
367    A `bool` indicating success.
368    """
369    if not force and venv in verified_venvs:
370        return True
371    if not force and venv_exists(venv, debug=debug):
372        if verify:
373            verify_venv(venv, debug=debug)
374            verified_venvs.add(venv)
375        return True
376
377    import io
378    from contextlib import redirect_stdout
379    import sys
380    import platform
381    import os
382    import pathlib
383    import shutil
384    import time
385
386    from meerschaum.config.static import STATIC_CONFIG
387    from meerschaum.config._paths import (
388        VIRTENV_RESOURCES_PATH,
389        VENVS_CACHE_RESOURCES_PATH,
390    )
391    from meerschaum.utils.packages import is_uv_enabled
392
393    venv_path = VIRTENV_RESOURCES_PATH / venv
394    vtp = venv_target_path(venv=venv, allow_nonexistent=True, debug=debug)
395    docker_home_venv_path = pathlib.Path('/home/meerschaum/venvs/mrsm')
396    lock_path = VENVS_CACHE_RESOURCES_PATH / (venv + '.lock')
397    work_dir_env_var = STATIC_CONFIG['environment']['work_dir']
398
399    def update_lock(active: bool):
400        try:
401            if not active:
402                if debug:
403                    print(f"Releasing lock: '{lock_path}'")
404                lock_path.unlink()
405            else:
406                if debug:
407                    print(f"Acquiring lock: '{lock_path}'")
408                lock_path.touch()
409        except Exception:
410            pass
411
412    def wait_for_lock():
413        max_lock_seconds = 30.0
414        sleep_message_seconds = 5.0
415        step_sleep_seconds = 0.1
416        init_venv_check_start = time.perf_counter()
417        last_print = init_venv_check_start
418        while ((time.perf_counter() - init_venv_check_start) < max_lock_seconds):
419            if not lock_path.exists():
420                break
421
422            now = time.perf_counter()
423            if debug or (now - last_print) > sleep_message_seconds:
424                print(f"Lock exists for venv '{venv}', sleeping...")
425                last_print = now
426            time.sleep(step_sleep_seconds)
427        update_lock(False)
428
429    if (
430        not force
431        and venv == 'mrsm'
432        and os.environ.get(work_dir_env_var, None) is not None
433        and docker_home_venv_path.exists()
434    ):
435        wait_for_lock()
436        shutil.move(docker_home_venv_path, venv_path)
437        if verify:
438            verify_venv(venv, debug=debug)
439            verified_venvs.add(venv)
440        return True
441
442    from meerschaum.utils.packages import run_python_package, attempt_import, _get_pip_os_env
443    global tried_virtualenv
444    try:
445        import venv as _venv
446        uv = attempt_import('uv', venv=None, debug=debug) if is_uv_enabled() else None
447        virtualenv = None
448    except ImportError:
449        _venv = None
450        uv = None
451        virtualenv = None
452
453    _venv_success = False
454    temp_vtp = VENVS_CACHE_RESOURCES_PATH / str(venv)
455    ### NOTE: Disable site-packages movement for now.
456    rename_vtp = False and vtp.exists() and not temp_vtp.exists()
457
458    wait_for_lock()
459    update_lock(True)
460
461    if rename_vtp:
462        if debug:
463            print(f"Moving '{vtp}' to '{temp_vtp}'...")
464        shutil.move(vtp, temp_vtp)
465
466    if uv is not None:
467        _venv_success = run_python_package(
468            'uv',
469            ['venv', venv_path.as_posix(), '-q', '--no-project', '--allow-existing', '--seed'],
470            venv=None,
471            env=_get_pip_os_env(),
472            debug=debug,
473        ) == 0
474
475    if _venv is not None and not _venv_success:
476        f = io.StringIO()
477        with redirect_stdout(f):
478            _venv_success = run_python_package(
479                'venv',
480                [venv_path.as_posix()] + (
481                    ['--symlinks']
482                    if platform.system() != 'Windows'
483                    else []
484                ),
485                venv=None, debug=debug
486            ) == 0
487        if not _venv_success:
488            print(f"Please install python3-venv.\n{f.getvalue()}\nFalling back to virtualenv...")
489        if not venv_exists(venv, debug=debug):
490            _venv = None
491    if not _venv_success:
492        virtualenv = attempt_import(
493            'virtualenv',
494            venv=None,
495            lazy=False,
496            install=(not tried_virtualenv),
497            warn=False,
498            check_update=False,
499            color=False,
500            debug=debug,
501        )
502        if virtualenv is None:
503            print(
504                "Failed to import `venv` or `virtualenv`! "
505                + "Please install `virtualenv` via pip then restart Meerschaum."
506            )
507            if rename_vtp and temp_vtp.exists():
508                if debug:
509                    print(f"Moving '{temp_vtp}' back to '{vtp}'...")
510                shutil.move(temp_vtp, vtp)
511            update_lock(False)
512            return False
513
514        tried_virtualenv = True
515        try:
516            python_folder = (
517                'python' + str(sys.version_info.major) + '.' + str(sys.version_info.minor)
518            )
519            dist_packages_path = (
520                VIRTENV_RESOURCES_PATH /
521                venv / 'local' / 'lib' / python_folder / 'dist-packages'
522            )
523            local_bin_path = VIRTENV_RESOURCES_PATH / venv / 'local' / 'bin'
524            bin_path = VIRTENV_RESOURCES_PATH / venv / 'bin'
525            vtp = venv_target_path(venv=venv, allow_nonexistent=True, debug=debug)
526            if bin_path.exists():
527                try:
528                    shutil.rmtree(bin_path)
529                except Exception:
530                    import traceback
531                    traceback.print_exc()
532            virtualenv.cli_run([venv_path.as_posix()])
533            if dist_packages_path.exists():
534                vtp.mkdir(exist_ok=True, parents=True)
535                for file_path in dist_packages_path.glob('*'):
536                    shutil.move(file_path, vtp)
537                shutil.rmtree(dist_packages_path)
538                #  shutil.move(dist_packages_path, vtp)
539                bin_path.mkdir(exist_ok=True, parents=True)
540                for file_path in local_bin_path.glob('*'):
541                    shutil.move(file_path, bin_path)
542                #  shutil.move(local_bin_path, bin_path)
543                shutil.rmtree(local_bin_path)
544
545        except Exception:
546            import traceback
547            traceback.print_exc()
548            if rename_vtp and temp_vtp.exists():
549                shutil.move(temp_vtp, vtp)
550            update_lock(False)
551            return False
552    if verify:
553        verify_venv(venv, debug=debug)
554        verified_venvs.add(venv)
555
556    if rename_vtp and temp_vtp.exists():
557        if debug:
558            print(f"Cleanup: move '{temp_vtp}' back to '{vtp}'.")
559        shutil.move(temp_vtp, vtp)
560
561    update_lock(False)
562    return True
563
564
565def venv_executable(venv: Optional[str] = 'mrsm') -> str:
566    """
567    The Python interpreter executable for a given virtual environment.
568    """
569    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
570    import sys
571    import platform
572    return (
573        sys.executable if venv is None
574        else str(
575            VIRTENV_RESOURCES_PATH
576            / venv
577            / (
578                'bin' if platform.system() != 'Windows'
579                else 'Scripts'
580            ) / (
581                'python'
582                + str(sys.version_info.major)
583                + '.'
584                + str(sys.version_info.minor)
585            )
586        )
587    )
588
589
590def venv_exec(
591    code: str,
592    venv: Optional[str] = 'mrsm',
593    env: Optional[Dict[str, str]] = None,
594    with_extras: bool = False,
595    as_proc: bool = False,
596    capture_output: bool = True,
597    debug: bool = False,
598) -> Union[bool, Tuple[int, bytes, bytes], 'subprocess.Popen']:
599    """
600    Execute Python code in a subprocess via a virtual environment's interpeter.
601    Return `True` if the code successfully executes, `False` on failure.
602
603    Parameters
604    ----------
605    code: str
606        The Python code to excecute.
607
608    venv: str, default 'mrsm'
609        The virtual environment to use to get the path for the Python executable.
610        If `venv` is `None`, use the default `sys.executable` path.
611
612    env: Optional[Dict[str, str]], default None
613        Optionally specify the environment variables for the subprocess.
614        Defaults to `os.environ`.
615
616    with_extras: bool, default False
617        If `True`, return a tuple of the exit code, stdout bytes, and stderr bytes.
618
619    as_proc: bool, default False
620        If `True`, return the `subprocess.Popen` object instead of executing.
621
622    Returns
623    -------
624    By default, return a bool indicating success.
625    If `as_proc` is `True`, return a `subprocess.Popen` object.
626    If `with_extras` is `True`, return a tuple of the exit code, stdout bytes, and stderr bytes.
627
628    """
629    import os
630    import subprocess
631    import platform
632    from meerschaum.utils.debug import dprint
633    from meerschaum.utils.process import _child_processes
634
635    executable = venv_executable(venv=venv)
636    cmd_list = [executable, '-c', code]
637    if env is None:
638        env = os.environ
639    if debug:
640        dprint(str(cmd_list))
641    if not with_extras and not as_proc:
642        return subprocess.call(cmd_list, env=env) == 0
643
644    stdout, stderr = (None, None) if not capture_output else (subprocess.PIPE, subprocess.PIPE)
645    group_kwargs = (
646        {
647            'preexec_fn': os.setsid,
648        } if platform.system() != 'Windows'
649        else {
650            'creationflags': CREATE_NEW_PROCESS_GROUP,
651        }
652    )
653    process = subprocess.Popen(
654        cmd_list,
655        stdout=stdout,
656        stderr=stderr,
657        env=env,
658        **group_kwargs
659    )
660    if as_proc:
661        _child_processes.append(process)
662        return process
663    stdout, stderr = process.communicate()
664    exit_code = process.returncode
665    return exit_code, stdout, stderr
666
667
668def venv_exists(venv: Union[str, None], debug: bool = False) -> bool:
669    """
670    Determine whether a virtual environment has been created.
671    """
672    target_path = venv_target_path(venv, allow_nonexistent=True, debug=debug)
673    return target_path.exists()
674
675
676def venv_target_path(
677    venv: Union[str, None],
678    allow_nonexistent: bool = False,
679    debug: bool = False,
680) -> 'pathlib.Path':
681    """
682    Return a virtual environment's site-package path.
683
684    Parameters
685    ----------
686    venv: Union[str, None]
687        The virtual environment for which a path should be returned.
688
689    allow_nonexistent: bool, default False
690        If `True`, return a path even if it does not exist.
691
692    Returns
693    -------
694    The `pathlib.Path` object for the virtual environment's path.
695
696    """
697    import os
698    import sys
699    import platform
700    import pathlib
701    import site
702    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
703    from meerschaum.config.static import STATIC_CONFIG
704
705    ### Check sys.path for a user-writable site-packages directory.
706    if venv is None:
707
708        ### Return the known value for the portable environment.
709        environment_runtime = STATIC_CONFIG['environment']['runtime']
710        if os.environ.get(environment_runtime, None) == 'portable':
711            python_version_folder = (
712                'python' + str(sys.version_info.major) + '.' + str(sys.version_info.minor)
713            )
714            executable_path = pathlib.Path(sys.executable)
715            site_packages_path = (
716                (
717                    executable_path.parent.parent / 'lib' / python_version_folder / 'site-packages'
718                ) if platform.system() != 'Windows' else (
719                    executable_path.parent / 'Lib' / 'site-packages'
720                )
721            )
722            if not site_packages_path.exists():
723                raise EnvironmentError(f"Could not find '{site_packages_path}'. Does it exist?")
724            return site_packages_path
725
726        if not inside_venv():
727            user_site_packages = site.getusersitepackages()
728            if user_site_packages is None:
729                raise EnvironmentError("Could not determine user site packages.")
730
731            site_path = pathlib.Path(user_site_packages)
732            if not site_path.exists():
733
734                ### Windows does not have `os.geteuid()`.
735                if platform.system() == 'Windows' or os.geteuid() != 0:
736                    site_path.mkdir(parents=True, exist_ok=True)
737                    return site_path
738
739                ### Allow for dist-level paths (running as root).
740                for possible_dist in site.getsitepackages():
741                    dist_path = pathlib.Path(possible_dist)
742                    if not dist_path.exists():
743                        continue
744                    return dist_path
745                
746                raise EnvironmentError("Could not determine the dist-packages directory.")
747
748            return site_path
749
750    venv_root_path = (
751        (VIRTENV_RESOURCES_PATH / venv)
752        if venv is not None else pathlib.Path(sys.prefix)
753    )
754    target_path = venv_root_path
755
756    ### Ensure 'lib' or 'Lib' exists.
757    lib = 'lib' if platform.system() != 'Windows' else 'Lib'
758    if not allow_nonexistent:
759        if not venv_root_path.exists() or lib not in os.listdir(venv_root_path):
760            print(f"Failed to find lib directory for virtual environment '{venv}'.")
761            import traceback
762            traceback.print_stack()
763            sys.exit(1)
764    target_path = target_path / lib
765
766    ### Check if a 'python3.x' folder exists.
767    python_folder = 'python' + str(sys.version_info.major) + '.' + str(sys.version_info.minor)
768    if target_path.exists():
769        target_path = (
770            (target_path / python_folder) if python_folder in os.listdir(target_path)
771            else target_path
772        )
773    else:
774        target_path = (
775            (target_path / python_folder) if platform.system() != 'Windows'
776            else target_path
777        )
778
779    ### Ensure 'site-packages' exists.
780    if allow_nonexistent or 'site-packages' in os.listdir(target_path): ### Windows
781        target_path = target_path / 'site-packages'
782    else:
783        import traceback
784        traceback.print_stack()
785        print(f"Failed to find site-packages directory for virtual environment '{venv}'.")
786        print("This may be because you are using a different Python version.")
787        print("Try deleting the following directory and restarting Meerschaum:")
788        print(VIRTENV_RESOURCES_PATH)
789        sys.exit(1)
790
791    return target_path
792
793
794def inside_venv() -> bool:
795    """
796    Determine whether current Python interpreter is running inside a virtual environment.
797    """
798    import sys
799    return (
800        hasattr(sys, 'real_prefix') or (
801            hasattr(sys, 'base_prefix')
802                and sys.base_prefix != sys.prefix
803        )
804    )
805
806
807def get_venvs() -> List[str]:
808    """
809    Return a list of all the virtual environments.
810    """
811    import os
812    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
813    venvs = []
814    for filename in os.listdir(VIRTENV_RESOURCES_PATH):
815        path = VIRTENV_RESOURCES_PATH / filename
816        if not path.is_dir():
817            continue
818        if not venv_exists(filename):
819            continue
820        venvs.append(filename)
821    return venvs
822
823
824def get_module_venv(module) -> Union[str, None]:
825    """
826    Return the virtual environment where an imported module is installed.
827
828    Parameters
829    ----------
830    module: ModuleType
831        The imported module to inspect.
832
833    Returns
834    -------
835    The name of a venv or `None`.
836    """
837    import pathlib
838    from meerschaum.config.paths import VIRTENV_RESOURCES_PATH
839    module_path = pathlib.Path(module.__file__).resolve()
840    try:
841        rel_path = module_path.relative_to(VIRTENV_RESOURCES_PATH)
842    except ValueError:
843        return None
844
845    return rel_path.as_posix().split('/', maxsplit=1)[0]
846
847
848from meerschaum.utils.venv._Venv import Venv
class Venv:
 18class Venv:
 19    """
 20    Manage a virtual enviroment's activation status.
 21
 22    Examples
 23    --------
 24    >>> from meerschaum.plugins import Plugin
 25    >>> with Venv('mrsm') as venv:
 26    ...     import pandas
 27    >>> with Venv(Plugin('noaa')) as venv:
 28    ...     import requests
 29    >>> venv = Venv('mrsm')
 30    >>> venv.activate()
 31    True
 32    >>> venv.deactivate()
 33    True
 34    >>> 
 35    """
 36
 37    def __init__(
 38        self,
 39        venv: Union[str, 'meerschaum.plugins.Plugin', None] = 'mrsm',
 40        debug: bool = False,
 41    ) -> None:
 42        from meerschaum.utils.venv import activate_venv, deactivate_venv, active_venvs
 43        ### For some weird threading issue,
 44        ### we can't use `isinstance` here.
 45        if 'meerschaum.plugins._Plugin' in str(type(venv)):
 46            self._venv = venv.name
 47            self._activate = venv.activate_venv
 48            self._deactivate = venv.deactivate_venv
 49            self._kwargs = {}
 50        else:
 51            self._venv = venv
 52            self._activate = activate_venv
 53            self._deactivate = deactivate_venv
 54            self._kwargs = {'venv': venv}
 55        self._debug = debug
 56        ### In case someone calls `deactivate()` before `activate()`.
 57        self._kwargs['previously_active_venvs'] = copy.deepcopy(active_venvs)
 58
 59
 60    def activate(self, debug: bool = False) -> bool:
 61        """
 62        Activate this virtual environment.
 63        If a `meerschaum.plugins.Plugin` was provided, its dependent virtual environments
 64        will also be activated.
 65        """
 66        from meerschaum.utils.venv import active_venvs, init_venv
 67        self._kwargs['previously_active_venvs'] = copy.deepcopy(active_venvs)
 68        try:
 69            return self._activate(debug=(debug or self._debug), **self._kwargs)
 70        except OSError as e:
 71            if not init_venv(self._venv, force=True):
 72                raise e
 73        return self._activate(debug=(debug or self._debug), **self._kwargs)
 74
 75
 76    def deactivate(self, debug: bool = False) -> bool:
 77        """
 78        Deactivate this virtual environment.
 79        If a `meerschaum.plugins.Plugin` was provided, its dependent virtual environments
 80        will also be deactivated.
 81        """
 82        return self._deactivate(debug=(debug or self._debug), **self._kwargs)
 83
 84
 85    @property
 86    def target_path(self) -> pathlib.Path:
 87        """
 88        Return the target site-packages path for this virtual environment.
 89        A `meerschaum.utils.venv.Venv` may have one virtual environment per minor Python version
 90        (e.g. Python 3.10 and Python 3.7).
 91        """
 92        from meerschaum.utils.venv import venv_target_path
 93        return venv_target_path(venv=self._venv, allow_nonexistent=True, debug=self._debug)
 94
 95
 96    @property
 97    def root_path(self) -> pathlib.Path:
 98        """
 99        Return the top-level path for this virtual environment.
100        """
101        from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
102        if self._venv is None:
103            return self.target_path.parent
104        return VIRTENV_RESOURCES_PATH / self._venv
105
106
107    def __enter__(self) -> None:
108        self.activate(debug=self._debug)
109
110
111    def __exit__(self, exc_type, exc_value, exc_traceback) -> None:
112        self.deactivate(debug=self._debug)
113
114
115    def __str__(self) -> str:
116        quote = "'" if self._venv is not None else ""
117        return "Venv(" + quote + str(self._venv) + quote + ")"
118
119
120    def __repr__(self) -> str:
121        return self.__str__()

Manage a virtual enviroment's activation status.

Examples
>>> from meerschaum.plugins import Plugin
>>> with Venv('mrsm') as venv:
...     import pandas
>>> with Venv(Plugin('noaa')) as venv:
...     import requests
>>> venv = Venv('mrsm')
>>> venv.activate()
True
>>> venv.deactivate()
True
>>>
Venv( venv: Union[str, meerschaum.Plugin, NoneType] = 'mrsm', debug: bool = False)
37    def __init__(
38        self,
39        venv: Union[str, 'meerschaum.plugins.Plugin', None] = 'mrsm',
40        debug: bool = False,
41    ) -> None:
42        from meerschaum.utils.venv import activate_venv, deactivate_venv, active_venvs
43        ### For some weird threading issue,
44        ### we can't use `isinstance` here.
45        if 'meerschaum.plugins._Plugin' in str(type(venv)):
46            self._venv = venv.name
47            self._activate = venv.activate_venv
48            self._deactivate = venv.deactivate_venv
49            self._kwargs = {}
50        else:
51            self._venv = venv
52            self._activate = activate_venv
53            self._deactivate = deactivate_venv
54            self._kwargs = {'venv': venv}
55        self._debug = debug
56        ### In case someone calls `deactivate()` before `activate()`.
57        self._kwargs['previously_active_venvs'] = copy.deepcopy(active_venvs)
def activate(self, debug: bool = False) -> bool:
60    def activate(self, debug: bool = False) -> bool:
61        """
62        Activate this virtual environment.
63        If a `meerschaum.plugins.Plugin` was provided, its dependent virtual environments
64        will also be activated.
65        """
66        from meerschaum.utils.venv import active_venvs, init_venv
67        self._kwargs['previously_active_venvs'] = copy.deepcopy(active_venvs)
68        try:
69            return self._activate(debug=(debug or self._debug), **self._kwargs)
70        except OSError as e:
71            if not init_venv(self._venv, force=True):
72                raise e
73        return self._activate(debug=(debug or self._debug), **self._kwargs)

Activate this virtual environment. If a meerschaum.plugins.Plugin was provided, its dependent virtual environments will also be activated.

def deactivate(self, debug: bool = False) -> bool:
76    def deactivate(self, debug: bool = False) -> bool:
77        """
78        Deactivate this virtual environment.
79        If a `meerschaum.plugins.Plugin` was provided, its dependent virtual environments
80        will also be deactivated.
81        """
82        return self._deactivate(debug=(debug or self._debug), **self._kwargs)

Deactivate this virtual environment. If a meerschaum.plugins.Plugin was provided, its dependent virtual environments will also be deactivated.

target_path: pathlib.Path
85    @property
86    def target_path(self) -> pathlib.Path:
87        """
88        Return the target site-packages path for this virtual environment.
89        A `meerschaum.utils.venv.Venv` may have one virtual environment per minor Python version
90        (e.g. Python 3.10 and Python 3.7).
91        """
92        from meerschaum.utils.venv import venv_target_path
93        return venv_target_path(venv=self._venv, allow_nonexistent=True, debug=self._debug)

Return the target site-packages path for this virtual environment. A meerschaum.utils.venv.Venv may have one virtual environment per minor Python version (e.g. Python 3.10 and Python 3.7).

root_path: pathlib.Path
 96    @property
 97    def root_path(self) -> pathlib.Path:
 98        """
 99        Return the top-level path for this virtual environment.
100        """
101        from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
102        if self._venv is None:
103            return self.target_path.parent
104        return VIRTENV_RESOURCES_PATH / self._venv

Return the top-level path for this virtual environment.

def activate_venv( venv: Optional[str] = 'mrsm', color: bool = True, force: bool = False, debug: bool = False, **kw) -> bool:
 36def activate_venv(
 37    venv: Optional[str] = 'mrsm',
 38    color: bool = True,
 39    force: bool = False,
 40    debug: bool = False,
 41    **kw
 42) -> bool:
 43    """
 44    Create a virtual environment (if it doesn't exist) and add it to `sys.path` if necessary.
 45
 46    Parameters
 47    ----------
 48    venv: Optional[str], default 'mrsm'
 49        The virtual environment to activate.
 50
 51    color: bool, default True
 52        If `True`, include color in debug text.
 53
 54    force: bool, default False
 55        If `True`, do not exit early even if the venv is currently active.
 56
 57    debug: bool, default False
 58        Verbosity toggle.
 59
 60    Returns
 61    -------
 62    A bool indicating whether the virtual environment was successfully activated.
 63
 64    """
 65    thread_id = get_ident()
 66    if active_venvs_order and active_venvs_order[0] == venv:
 67        if not force:
 68            return True
 69    import sys
 70    import os
 71    if venv is not None:
 72        init_venv(venv=venv, debug=debug)
 73    with LOCKS['active_venvs']:
 74        if thread_id not in threads_active_venvs:
 75            threads_active_venvs[thread_id] = {}
 76        active_venvs.add(venv)
 77        if venv not in threads_active_venvs[thread_id]:
 78            threads_active_venvs[thread_id][venv] = 1
 79        else:
 80            threads_active_venvs[thread_id][venv] += 1
 81
 82        target_path = venv_target_path(venv, debug=debug, allow_nonexistent=True)
 83        if not target_path.exists():
 84            init_venv(venv=venv, force=True, debug=debug)
 85        if not target_path.exists():
 86            raise EnvironmentError(f"Could not activate virtual environment '{venv}'.")
 87        target = target_path.as_posix()
 88
 89        if venv in active_venvs_order:
 90            sys.path.remove(target)
 91            try:
 92                active_venvs_order.remove(venv)
 93            except Exception:
 94                pass
 95        if venv is not None:
 96            sys.path.insert(0, target)
 97        else:
 98            if sys.path and sys.path[0] in (os.getcwd(), ''):
 99                sys.path.insert(1, target)
100            else:
101                sys.path.insert(0, target)
102        try:
103            active_venvs_order.insert(0, venv)
104        except Exception:
105            pass
106
107    return True

Create a virtual environment (if it doesn't exist) and add it to sys.path if necessary.

Parameters
  • venv (Optional[str], default 'mrsm'): The virtual environment to activate.
  • color (bool, default True): If True, include color in debug text.
  • force (bool, default False): If True, do not exit early even if the venv is currently active.
  • debug (bool, default False): Verbosity toggle.
Returns
  • A bool indicating whether the virtual environment was successfully activated.
def deactivate_venv( venv: str = 'mrsm', color: bool = True, debug: bool = False, previously_active_venvs: Union[set[str], List[str], NoneType] = None, force: bool = False, **kw) -> bool:
110def deactivate_venv(
111    venv: str = 'mrsm',
112    color: bool = True,
113    debug: bool = False,
114    previously_active_venvs: Union['set[str]', List[str], None] = None,
115    force: bool = False,
116    **kw
117) -> bool:
118    """
119    Remove a virtual environment from `sys.path` (if it's been activated).
120
121    Parameters
122    ----------
123    venv: str, default 'mrsm'
124        The virtual environment to deactivate.
125
126    color: bool, default True
127        If `True`, include color in debug text.
128
129    debug: bool, default False
130        Verbosity toggle.
131
132    previously_active_venvs: Union[Set[str], List[str], None]
133        If provided, skip deactivating if a virtual environment is in this iterable.
134
135    force: bool, default False
136        If `True`, forcibly deactivate the virtual environment.
137        This may cause issues with other threads, so be careful!
138
139    Returns
140    -------
141    Return a bool indicating whether the virtual environment was successfully deactivated.
142
143    """
144    import sys
145    thread_id = get_ident()
146    if venv is None:
147        if venv in active_venvs:
148            active_venvs.remove(venv)
149        return True
150
151    if previously_active_venvs and venv in previously_active_venvs and not force:
152        return True
153
154    with LOCKS['active_venvs']:
155        if venv in threads_active_venvs.get(thread_id, {}):
156            new_count = threads_active_venvs[thread_id][venv] - 1
157            if new_count > 0 and not force:
158                threads_active_venvs[thread_id][venv] = new_count
159                return True
160            else:
161                del threads_active_venvs[thread_id][venv]
162
163        if not force:
164            for other_thread_id, other_venvs in threads_active_venvs.items():
165                if other_thread_id == thread_id:
166                    continue
167                if venv in other_venvs:
168                    return True
169        else:
170            to_delete = [other_thread_id for other_thread_id in threads_active_venvs]
171            for other_thread_id in to_delete:
172                del threads_active_venvs[other_thread_id]
173
174        if venv in active_venvs:
175            active_venvs.remove(venv)
176
177    if sys.path is None:
178        return False
179
180    target = venv_target_path(venv, allow_nonexistent=True, debug=debug).as_posix()
181    with LOCKS['sys.path']:
182        if target in sys.path:
183            try:
184                sys.path.remove(target)
185            except Exception:
186                pass
187            try:
188                active_venvs_order.remove(venv)
189            except Exception:
190                pass
191
192    return True

Remove a virtual environment from sys.path (if it's been activated).

Parameters
  • venv (str, default 'mrsm'): The virtual environment to deactivate.
  • color (bool, default True): If True, include color in debug text.
  • debug (bool, default False): Verbosity toggle.
  • previously_active_venvs (Union[Set[str], List[str], None]): If provided, skip deactivating if a virtual environment is in this iterable.
  • force (bool, default False): If True, forcibly deactivate the virtual environment. This may cause issues with other threads, so be careful!
Returns
  • Return a bool indicating whether the virtual environment was successfully deactivated.
def get_module_venv(module) -> Optional[str]:
825def get_module_venv(module) -> Union[str, None]:
826    """
827    Return the virtual environment where an imported module is installed.
828
829    Parameters
830    ----------
831    module: ModuleType
832        The imported module to inspect.
833
834    Returns
835    -------
836    The name of a venv or `None`.
837    """
838    import pathlib
839    from meerschaum.config.paths import VIRTENV_RESOURCES_PATH
840    module_path = pathlib.Path(module.__file__).resolve()
841    try:
842        rel_path = module_path.relative_to(VIRTENV_RESOURCES_PATH)
843    except ValueError:
844        return None
845
846    return rel_path.as_posix().split('/', maxsplit=1)[0]

Return the virtual environment where an imported module is installed.

Parameters
  • module (ModuleType): The imported module to inspect.
Returns
  • The name of a venv or None.
def get_venvs() -> List[str]:
808def get_venvs() -> List[str]:
809    """
810    Return a list of all the virtual environments.
811    """
812    import os
813    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
814    venvs = []
815    for filename in os.listdir(VIRTENV_RESOURCES_PATH):
816        path = VIRTENV_RESOURCES_PATH / filename
817        if not path.is_dir():
818            continue
819        if not venv_exists(filename):
820            continue
821        venvs.append(filename)
822    return venvs

Return a list of all the virtual environments.

def init_venv( venv: str = 'mrsm', verify: bool = True, force: bool = False, debug: bool = False) -> bool:
346def init_venv(
347    venv: str = 'mrsm',
348    verify: bool = True,
349    force: bool = False,
350    debug: bool = False,
351) -> bool:
352    """
353    Initialize the virtual environment.
354
355    Parameters
356    ----------
357    venv: str, default 'mrsm'
358        The name of the virtual environment to create.
359
360    verify: bool, default True
361        If `True`, verify that the virtual environment is in the expected state.
362
363    force: bool, default False
364        If `True`, recreate the virtual environment, even if already initalized.
365
366    Returns
367    -------
368    A `bool` indicating success.
369    """
370    if not force and venv in verified_venvs:
371        return True
372    if not force and venv_exists(venv, debug=debug):
373        if verify:
374            verify_venv(venv, debug=debug)
375            verified_venvs.add(venv)
376        return True
377
378    import io
379    from contextlib import redirect_stdout
380    import sys
381    import platform
382    import os
383    import pathlib
384    import shutil
385    import time
386
387    from meerschaum.config.static import STATIC_CONFIG
388    from meerschaum.config._paths import (
389        VIRTENV_RESOURCES_PATH,
390        VENVS_CACHE_RESOURCES_PATH,
391    )
392    from meerschaum.utils.packages import is_uv_enabled
393
394    venv_path = VIRTENV_RESOURCES_PATH / venv
395    vtp = venv_target_path(venv=venv, allow_nonexistent=True, debug=debug)
396    docker_home_venv_path = pathlib.Path('/home/meerschaum/venvs/mrsm')
397    lock_path = VENVS_CACHE_RESOURCES_PATH / (venv + '.lock')
398    work_dir_env_var = STATIC_CONFIG['environment']['work_dir']
399
400    def update_lock(active: bool):
401        try:
402            if not active:
403                if debug:
404                    print(f"Releasing lock: '{lock_path}'")
405                lock_path.unlink()
406            else:
407                if debug:
408                    print(f"Acquiring lock: '{lock_path}'")
409                lock_path.touch()
410        except Exception:
411            pass
412
413    def wait_for_lock():
414        max_lock_seconds = 30.0
415        sleep_message_seconds = 5.0
416        step_sleep_seconds = 0.1
417        init_venv_check_start = time.perf_counter()
418        last_print = init_venv_check_start
419        while ((time.perf_counter() - init_venv_check_start) < max_lock_seconds):
420            if not lock_path.exists():
421                break
422
423            now = time.perf_counter()
424            if debug or (now - last_print) > sleep_message_seconds:
425                print(f"Lock exists for venv '{venv}', sleeping...")
426                last_print = now
427            time.sleep(step_sleep_seconds)
428        update_lock(False)
429
430    if (
431        not force
432        and venv == 'mrsm'
433        and os.environ.get(work_dir_env_var, None) is not None
434        and docker_home_venv_path.exists()
435    ):
436        wait_for_lock()
437        shutil.move(docker_home_venv_path, venv_path)
438        if verify:
439            verify_venv(venv, debug=debug)
440            verified_venvs.add(venv)
441        return True
442
443    from meerschaum.utils.packages import run_python_package, attempt_import, _get_pip_os_env
444    global tried_virtualenv
445    try:
446        import venv as _venv
447        uv = attempt_import('uv', venv=None, debug=debug) if is_uv_enabled() else None
448        virtualenv = None
449    except ImportError:
450        _venv = None
451        uv = None
452        virtualenv = None
453
454    _venv_success = False
455    temp_vtp = VENVS_CACHE_RESOURCES_PATH / str(venv)
456    ### NOTE: Disable site-packages movement for now.
457    rename_vtp = False and vtp.exists() and not temp_vtp.exists()
458
459    wait_for_lock()
460    update_lock(True)
461
462    if rename_vtp:
463        if debug:
464            print(f"Moving '{vtp}' to '{temp_vtp}'...")
465        shutil.move(vtp, temp_vtp)
466
467    if uv is not None:
468        _venv_success = run_python_package(
469            'uv',
470            ['venv', venv_path.as_posix(), '-q', '--no-project', '--allow-existing', '--seed'],
471            venv=None,
472            env=_get_pip_os_env(),
473            debug=debug,
474        ) == 0
475
476    if _venv is not None and not _venv_success:
477        f = io.StringIO()
478        with redirect_stdout(f):
479            _venv_success = run_python_package(
480                'venv',
481                [venv_path.as_posix()] + (
482                    ['--symlinks']
483                    if platform.system() != 'Windows'
484                    else []
485                ),
486                venv=None, debug=debug
487            ) == 0
488        if not _venv_success:
489            print(f"Please install python3-venv.\n{f.getvalue()}\nFalling back to virtualenv...")
490        if not venv_exists(venv, debug=debug):
491            _venv = None
492    if not _venv_success:
493        virtualenv = attempt_import(
494            'virtualenv',
495            venv=None,
496            lazy=False,
497            install=(not tried_virtualenv),
498            warn=False,
499            check_update=False,
500            color=False,
501            debug=debug,
502        )
503        if virtualenv is None:
504            print(
505                "Failed to import `venv` or `virtualenv`! "
506                + "Please install `virtualenv` via pip then restart Meerschaum."
507            )
508            if rename_vtp and temp_vtp.exists():
509                if debug:
510                    print(f"Moving '{temp_vtp}' back to '{vtp}'...")
511                shutil.move(temp_vtp, vtp)
512            update_lock(False)
513            return False
514
515        tried_virtualenv = True
516        try:
517            python_folder = (
518                'python' + str(sys.version_info.major) + '.' + str(sys.version_info.minor)
519            )
520            dist_packages_path = (
521                VIRTENV_RESOURCES_PATH /
522                venv / 'local' / 'lib' / python_folder / 'dist-packages'
523            )
524            local_bin_path = VIRTENV_RESOURCES_PATH / venv / 'local' / 'bin'
525            bin_path = VIRTENV_RESOURCES_PATH / venv / 'bin'
526            vtp = venv_target_path(venv=venv, allow_nonexistent=True, debug=debug)
527            if bin_path.exists():
528                try:
529                    shutil.rmtree(bin_path)
530                except Exception:
531                    import traceback
532                    traceback.print_exc()
533            virtualenv.cli_run([venv_path.as_posix()])
534            if dist_packages_path.exists():
535                vtp.mkdir(exist_ok=True, parents=True)
536                for file_path in dist_packages_path.glob('*'):
537                    shutil.move(file_path, vtp)
538                shutil.rmtree(dist_packages_path)
539                #  shutil.move(dist_packages_path, vtp)
540                bin_path.mkdir(exist_ok=True, parents=True)
541                for file_path in local_bin_path.glob('*'):
542                    shutil.move(file_path, bin_path)
543                #  shutil.move(local_bin_path, bin_path)
544                shutil.rmtree(local_bin_path)
545
546        except Exception:
547            import traceback
548            traceback.print_exc()
549            if rename_vtp and temp_vtp.exists():
550                shutil.move(temp_vtp, vtp)
551            update_lock(False)
552            return False
553    if verify:
554        verify_venv(venv, debug=debug)
555        verified_venvs.add(venv)
556
557    if rename_vtp and temp_vtp.exists():
558        if debug:
559            print(f"Cleanup: move '{temp_vtp}' back to '{vtp}'.")
560        shutil.move(temp_vtp, vtp)
561
562    update_lock(False)
563    return True

Initialize the virtual environment.

Parameters
  • venv (str, default 'mrsm'): The name of the virtual environment to create.
  • verify (bool, default True): If True, verify that the virtual environment is in the expected state.
  • force (bool, default False): If True, recreate the virtual environment, even if already initalized.
Returns
  • A bool indicating success.
def inside_venv() -> bool:
795def inside_venv() -> bool:
796    """
797    Determine whether current Python interpreter is running inside a virtual environment.
798    """
799    import sys
800    return (
801        hasattr(sys, 'real_prefix') or (
802            hasattr(sys, 'base_prefix')
803                and sys.base_prefix != sys.prefix
804        )
805    )

Determine whether current Python interpreter is running inside a virtual environment.

def is_venv_active(venv: str = 'mrsm', color: bool = True, debug: bool = False) -> bool:
195def is_venv_active(
196    venv: str = 'mrsm',
197    color : bool = True,
198    debug: bool = False
199) -> bool:
200    """
201    Check if a virtual environment is active.
202
203    Parameters
204    ----------
205    venv: str, default 'mrsm'
206        The virtual environment to check.
207
208    color: bool, default True
209        If `True`, include color in debug text.
210
211    debug: bool, default False
212        Verbosity toggle.
213
214    Returns
215    -------
216    A bool indicating whether the virtual environment `venv` is active.
217
218    """
219    return venv in active_venvs

Check if a virtual environment is active.

Parameters
  • venv (str, default 'mrsm'): The virtual environment to check.
  • color (bool, default True): If True, include color in debug text.
  • debug (bool, default False): Verbosity toggle.
Returns
  • A bool indicating whether the virtual environment venv is active.
def venv_exec( code: str, venv: Optional[str] = 'mrsm', env: Optional[Dict[str, str]] = None, with_extras: bool = False, as_proc: bool = False, capture_output: bool = True, debug: bool = False) -> Union[bool, Tuple[int, bytes, bytes], subprocess.Popen]:
591def venv_exec(
592    code: str,
593    venv: Optional[str] = 'mrsm',
594    env: Optional[Dict[str, str]] = None,
595    with_extras: bool = False,
596    as_proc: bool = False,
597    capture_output: bool = True,
598    debug: bool = False,
599) -> Union[bool, Tuple[int, bytes, bytes], 'subprocess.Popen']:
600    """
601    Execute Python code in a subprocess via a virtual environment's interpeter.
602    Return `True` if the code successfully executes, `False` on failure.
603
604    Parameters
605    ----------
606    code: str
607        The Python code to excecute.
608
609    venv: str, default 'mrsm'
610        The virtual environment to use to get the path for the Python executable.
611        If `venv` is `None`, use the default `sys.executable` path.
612
613    env: Optional[Dict[str, str]], default None
614        Optionally specify the environment variables for the subprocess.
615        Defaults to `os.environ`.
616
617    with_extras: bool, default False
618        If `True`, return a tuple of the exit code, stdout bytes, and stderr bytes.
619
620    as_proc: bool, default False
621        If `True`, return the `subprocess.Popen` object instead of executing.
622
623    Returns
624    -------
625    By default, return a bool indicating success.
626    If `as_proc` is `True`, return a `subprocess.Popen` object.
627    If `with_extras` is `True`, return a tuple of the exit code, stdout bytes, and stderr bytes.
628
629    """
630    import os
631    import subprocess
632    import platform
633    from meerschaum.utils.debug import dprint
634    from meerschaum.utils.process import _child_processes
635
636    executable = venv_executable(venv=venv)
637    cmd_list = [executable, '-c', code]
638    if env is None:
639        env = os.environ
640    if debug:
641        dprint(str(cmd_list))
642    if not with_extras and not as_proc:
643        return subprocess.call(cmd_list, env=env) == 0
644
645    stdout, stderr = (None, None) if not capture_output else (subprocess.PIPE, subprocess.PIPE)
646    group_kwargs = (
647        {
648            'preexec_fn': os.setsid,
649        } if platform.system() != 'Windows'
650        else {
651            'creationflags': CREATE_NEW_PROCESS_GROUP,
652        }
653    )
654    process = subprocess.Popen(
655        cmd_list,
656        stdout=stdout,
657        stderr=stderr,
658        env=env,
659        **group_kwargs
660    )
661    if as_proc:
662        _child_processes.append(process)
663        return process
664    stdout, stderr = process.communicate()
665    exit_code = process.returncode
666    return exit_code, stdout, stderr

Execute Python code in a subprocess via a virtual environment's interpeter. Return True if the code successfully executes, False on failure.

Parameters
  • code (str): The Python code to excecute.
  • venv (str, default 'mrsm'): The virtual environment to use to get the path for the Python executable. If venv is None, use the default sys.executable path.
  • env (Optional[Dict[str, str]], default None): Optionally specify the environment variables for the subprocess. Defaults to os.environ.
  • with_extras (bool, default False): If True, return a tuple of the exit code, stdout bytes, and stderr bytes.
  • as_proc (bool, default False): If True, return the subprocess.Popen object instead of executing.
Returns
  • By default, return a bool indicating success.
  • If as_proc is True, return a subprocess.Popen object.
  • If with_extras is True, return a tuple of the exit code, stdout bytes, and stderr bytes.
def venv_executable(venv: Optional[str] = 'mrsm') -> str:
566def venv_executable(venv: Optional[str] = 'mrsm') -> str:
567    """
568    The Python interpreter executable for a given virtual environment.
569    """
570    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
571    import sys
572    import platform
573    return (
574        sys.executable if venv is None
575        else str(
576            VIRTENV_RESOURCES_PATH
577            / venv
578            / (
579                'bin' if platform.system() != 'Windows'
580                else 'Scripts'
581            ) / (
582                'python'
583                + str(sys.version_info.major)
584                + '.'
585                + str(sys.version_info.minor)
586            )
587        )
588    )

The Python interpreter executable for a given virtual environment.

def venv_exists(venv: Optional[str], debug: bool = False) -> bool:
669def venv_exists(venv: Union[str, None], debug: bool = False) -> bool:
670    """
671    Determine whether a virtual environment has been created.
672    """
673    target_path = venv_target_path(venv, allow_nonexistent=True, debug=debug)
674    return target_path.exists()

Determine whether a virtual environment has been created.

def venv_target_path( venv: Optional[str], allow_nonexistent: bool = False, debug: bool = False) -> pathlib.Path:
677def venv_target_path(
678    venv: Union[str, None],
679    allow_nonexistent: bool = False,
680    debug: bool = False,
681) -> 'pathlib.Path':
682    """
683    Return a virtual environment's site-package path.
684
685    Parameters
686    ----------
687    venv: Union[str, None]
688        The virtual environment for which a path should be returned.
689
690    allow_nonexistent: bool, default False
691        If `True`, return a path even if it does not exist.
692
693    Returns
694    -------
695    The `pathlib.Path` object for the virtual environment's path.
696
697    """
698    import os
699    import sys
700    import platform
701    import pathlib
702    import site
703    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
704    from meerschaum.config.static import STATIC_CONFIG
705
706    ### Check sys.path for a user-writable site-packages directory.
707    if venv is None:
708
709        ### Return the known value for the portable environment.
710        environment_runtime = STATIC_CONFIG['environment']['runtime']
711        if os.environ.get(environment_runtime, None) == 'portable':
712            python_version_folder = (
713                'python' + str(sys.version_info.major) + '.' + str(sys.version_info.minor)
714            )
715            executable_path = pathlib.Path(sys.executable)
716            site_packages_path = (
717                (
718                    executable_path.parent.parent / 'lib' / python_version_folder / 'site-packages'
719                ) if platform.system() != 'Windows' else (
720                    executable_path.parent / 'Lib' / 'site-packages'
721                )
722            )
723            if not site_packages_path.exists():
724                raise EnvironmentError(f"Could not find '{site_packages_path}'. Does it exist?")
725            return site_packages_path
726
727        if not inside_venv():
728            user_site_packages = site.getusersitepackages()
729            if user_site_packages is None:
730                raise EnvironmentError("Could not determine user site packages.")
731
732            site_path = pathlib.Path(user_site_packages)
733            if not site_path.exists():
734
735                ### Windows does not have `os.geteuid()`.
736                if platform.system() == 'Windows' or os.geteuid() != 0:
737                    site_path.mkdir(parents=True, exist_ok=True)
738                    return site_path
739
740                ### Allow for dist-level paths (running as root).
741                for possible_dist in site.getsitepackages():
742                    dist_path = pathlib.Path(possible_dist)
743                    if not dist_path.exists():
744                        continue
745                    return dist_path
746                
747                raise EnvironmentError("Could not determine the dist-packages directory.")
748
749            return site_path
750
751    venv_root_path = (
752        (VIRTENV_RESOURCES_PATH / venv)
753        if venv is not None else pathlib.Path(sys.prefix)
754    )
755    target_path = venv_root_path
756
757    ### Ensure 'lib' or 'Lib' exists.
758    lib = 'lib' if platform.system() != 'Windows' else 'Lib'
759    if not allow_nonexistent:
760        if not venv_root_path.exists() or lib not in os.listdir(venv_root_path):
761            print(f"Failed to find lib directory for virtual environment '{venv}'.")
762            import traceback
763            traceback.print_stack()
764            sys.exit(1)
765    target_path = target_path / lib
766
767    ### Check if a 'python3.x' folder exists.
768    python_folder = 'python' + str(sys.version_info.major) + '.' + str(sys.version_info.minor)
769    if target_path.exists():
770        target_path = (
771            (target_path / python_folder) if python_folder in os.listdir(target_path)
772            else target_path
773        )
774    else:
775        target_path = (
776            (target_path / python_folder) if platform.system() != 'Windows'
777            else target_path
778        )
779
780    ### Ensure 'site-packages' exists.
781    if allow_nonexistent or 'site-packages' in os.listdir(target_path): ### Windows
782        target_path = target_path / 'site-packages'
783    else:
784        import traceback
785        traceback.print_stack()
786        print(f"Failed to find site-packages directory for virtual environment '{venv}'.")
787        print("This may be because you are using a different Python version.")
788        print("Try deleting the following directory and restarting Meerschaum:")
789        print(VIRTENV_RESOURCES_PATH)
790        sys.exit(1)
791
792    return target_path

Return a virtual environment's site-package path.

Parameters
  • venv (Union[str, None]): The virtual environment for which a path should be returned.
  • allow_nonexistent (bool, default False): If True, return a path even if it does not exist.
Returns
  • The pathlib.Path object for the virtual environment's path.
def verify_venv(venv: str, debug: bool = False) -> None:
223def verify_venv(
224    venv: str,
225    debug: bool = False,
226) -> None:
227    """
228    Verify that the virtual environment matches the expected state.
229    """
230    import pathlib
231    import platform
232    import os
233    import shutil
234    import sys
235    from meerschaum.config._paths import VIRTENV_RESOURCES_PATH
236    from meerschaum.utils.process import run_process
237    from meerschaum.utils.misc import make_symlink, is_symlink
238    from meerschaum.utils.warnings import warn
239
240    venv_path = VIRTENV_RESOURCES_PATH / venv
241    bin_path = venv_path / (
242        'bin' if platform.system() != 'Windows' else "Scripts"
243    )
244    current_python_versioned_name = (
245        'python' + str(sys.version_info.major) + '.' + str(sys.version_info.minor)
246        + ('' if platform.system() != 'Windows' else '.exe')
247    )
248
249    if not (bin_path / current_python_versioned_name).exists():
250        init_venv(venv, verify=False, force=True, debug=debug)
251        current_python_in_venv_path = pathlib.Path(venv_executable(venv=venv))
252        current_python_in_sys_path = pathlib.Path(venv_executable(venv=None))
253        if not current_python_in_venv_path.exists():
254            if is_symlink(current_python_in_venv_path):
255                try:
256                    current_python_in_venv_path.unlink()
257                except Exception as e:
258                    print(f"Unable to remove symlink {current_python_in_venv_path}:\n{e}")
259            try:
260                make_symlink(current_python_in_sys_path, current_python_in_venv_path)
261            except Exception:
262                print(
263                    f"Unable to create symlink {current_python_in_venv_path} "
264                    + f"to {current_python_in_sys_path}."
265                )
266        files_to_inspect = sorted(os.listdir(bin_path), reverse=True)
267    else:
268        files_to_inspect = [current_python_versioned_name]
269
270    def get_python_version(python_path: pathlib.Path) -> Union[str, None]:
271        """
272        Return the version for the python binary at the given path.
273        """
274        try:
275            ### It might be a broken symlink, so skip on errors.
276            if debug:
277                print(f"Getting python version for {python_path}")
278            proc = run_process(
279                [str(python_path), '-V'],
280                as_proc=True,
281                capture_output=True,
282            )
283            stdout, stderr = proc.communicate(timeout=1.0)
284        except Exception as e:
285            ### E.g. the symlink may be broken.
286            if is_symlink(python_path):
287                try:
288                    python_path.unlink()
289                except Exception as _e:
290                    print(f"Unable to remove broken symlink {python_path}:\n{e}\n{_e}")
291            return None
292        return stdout.decode('utf-8').strip().replace('Python ', '')
293
294    ### Ensure the versions are symlinked correctly.
295    for filename in files_to_inspect:
296        if not filename.startswith('python'):
297            continue
298        python_path = bin_path / filename
299        version = get_python_version(python_path)
300        if version is None:
301            continue
302        try:
303            major_version = version.split('.', maxsplit=1)[0]
304            minor_version = version.split('.', maxsplit=2)[1]
305        except IndexError:
306            return
307        python_versioned_name = (
308            'python' + major_version + '.' + minor_version
309            + ('' if platform.system() != 'Windows' else '.exe')
310        )
311
312        ### E.g. python3.10 actually links to Python 3.10.
313        if filename == python_versioned_name:
314            real_path = pathlib.Path(os.path.realpath(python_path))
315            if not real_path.exists():
316                try:
317                    python_path.unlink()
318                except Exception:
319                    pass
320                init_venv(venv, verify=False, force=True, debug=debug)
321                if not python_path.exists():
322                    raise FileNotFoundError(f"Unable to verify Python symlink:\n{python_path}")
323
324            if python_path == real_path:
325                continue
326
327            try:
328                python_path.unlink()
329            except Exception:
330                pass
331            success, msg = make_symlink(real_path, python_path)
332            if not success:
333                warn(msg, color=False)
334            continue
335
336        python_versioned_path = bin_path / python_versioned_name
337        if python_versioned_path.exists():
338            ### Avoid circular symlinks.
339            if get_python_version(python_versioned_path) == version:
340                continue
341            python_versioned_path.unlink()
342        shutil.move(python_path, python_versioned_path)

Verify that the virtual environment matches the expected state.