|
8 | 8 | import logging |
9 | 9 | import platform |
10 | 10 | import subprocess |
11 | | -from typing import Any, Dict, Tuple, Union, Mapping, Optional, cast |
| 11 | +import importlib.util |
| 12 | +from typing import Any, Dict, Tuple, Union, Mapping, Optional, NamedTuple, cast |
12 | 13 | from pathlib import Path |
13 | 14 | from functools import lru_cache |
| 15 | +from typing_extensions import Literal, assert_never |
14 | 16 |
|
15 | 17 | from . import errors |
16 | | -from .types import Binary, Target, Strategy, check_target |
| 18 | +from .types import Target, check_target |
17 | 19 | from .utils import env_to_bool, get_bin_dir, get_env_dir, maybe_decode |
18 | 20 |
|
19 | 21 | log: logging.Logger = logging.getLogger(__name__) |
20 | 22 |
|
21 | 23 | ENV_DIR: Path = get_env_dir() |
22 | 24 | BINARIES_DIR: Path = get_bin_dir(env_dir=ENV_DIR) |
23 | 25 | USE_GLOBAL_NODE = env_to_bool('PYRIGHT_PYTHON_GLOBAL_NODE', default=True) |
| 26 | +USE_NODEJS_WHEEL = env_to_bool('PYRIGHT_PYTHON_NODEJS_WHEEL', default=True) |
24 | 27 | NODE_VERSION = os.environ.get('PYRIGHT_PYTHON_NODE_VERSION', default=None) |
25 | 28 | VERSION_RE = re.compile(r'\d+\.\d+\.\d+') |
26 | 29 |
|
27 | 30 |
|
28 | | -def _ensure_available(target: Target) -> Binary: |
29 | | - """Ensure the target node executable is available""" |
30 | | - path = None |
31 | | - if USE_GLOBAL_NODE: |
32 | | - path = _get_global_binary(target) |
33 | | - |
34 | | - if path is not None: |
35 | | - return Binary(path=path, strategy=Strategy.GLOBAL) |
36 | | - |
37 | | - return Binary(path=_ensure_node_env(target), strategy=Strategy.NODEENV) |
38 | | - |
39 | | - |
40 | 31 | def _is_windows() -> bool: |
41 | 32 | return platform.system().lower() == 'windows' |
42 | 33 |
|
@@ -97,31 +88,83 @@ def _install_node_env() -> None: |
97 | 88 | subprocess.run(args, check=True) |
98 | 89 |
|
99 | 90 |
|
| 91 | +class GlobalStrategy(NamedTuple): |
| 92 | + type: Literal['global'] |
| 93 | + path: Path |
| 94 | + |
| 95 | + |
| 96 | +class NodeJSWheelStrategy(NamedTuple): |
| 97 | + type: Literal['nodejs_wheel'] |
| 98 | + |
| 99 | + |
| 100 | +class NodeenvStrategy(NamedTuple): |
| 101 | + type: Literal['nodeenv'] |
| 102 | + path: Path |
| 103 | + |
| 104 | + |
| 105 | +Strategy = Union[GlobalStrategy, NodeJSWheelStrategy, NodeenvStrategy] |
| 106 | + |
| 107 | + |
| 108 | +def _resolve_strategy(target: Target) -> Strategy: |
| 109 | + if USE_NODEJS_WHEEL: |
| 110 | + if importlib.util.find_spec('nodejs_wheel') is not None: |
| 111 | + log.debug('Using nodejs_wheel package for resolving binaries') |
| 112 | + return NodeJSWheelStrategy(type='nodejs_wheel') |
| 113 | + |
| 114 | + if USE_GLOBAL_NODE: |
| 115 | + path = _get_global_binary(target) |
| 116 | + if path is not None: |
| 117 | + log.debug('Using global %s binary', target) |
| 118 | + return GlobalStrategy(type='global', path=path) |
| 119 | + |
| 120 | + log.debug('Installing binaries using nodeenv') |
| 121 | + return NodeenvStrategy(type='nodeenv', path=_ensure_node_env(target)) |
| 122 | + |
| 123 | + |
100 | 124 | def run( |
101 | 125 | target: Target, *args: str, **kwargs: Any |
102 | 126 | ) -> Union['subprocess.CompletedProcess[bytes]', 'subprocess.CompletedProcess[str]']: |
103 | 127 | check_target(target) |
104 | | - binary = _ensure_available(target) |
105 | | - env = kwargs.pop('env', None) or os.environ.copy() |
106 | 128 |
|
107 | | - if binary.strategy == Strategy.NODEENV: |
| 129 | + strategy = _resolve_strategy(target) |
| 130 | + if strategy.type == 'global': |
| 131 | + node_args = [str(strategy.path), *args] |
| 132 | + log.debug('Running global node command with args: %s', node_args) |
| 133 | + return cast( |
| 134 | + 'subprocess.CompletedProcess[str] | subprocess.CompletedProcess[bytes]', |
| 135 | + subprocess.run(node_args, **kwargs), |
| 136 | + ) |
| 137 | + elif strategy.type == 'nodejs_wheel': |
| 138 | + import nodejs_wheel |
| 139 | + |
| 140 | + if target == 'node': |
| 141 | + return cast( |
| 142 | + 'subprocess.CompletedProcess[str] | subprocess.CompletedProcess[bytes]', |
| 143 | + nodejs_wheel.node(args, return_completed_process=True, **kwargs), |
| 144 | + ) |
| 145 | + elif target == 'npm': |
| 146 | + return cast( |
| 147 | + 'subprocess.CompletedProcess[str] | subprocess.CompletedProcess[bytes]', |
| 148 | + nodejs_wheel.npm(args, return_completed_process=True, **kwargs), |
| 149 | + ) |
| 150 | + else: |
| 151 | + assert_never(target) |
| 152 | + elif strategy.type == 'nodeenv': |
| 153 | + env = kwargs.pop('env', None) or os.environ.copy() |
108 | 154 | env.update(get_env_variables()) |
109 | 155 |
|
110 | 156 | # If we're using `nodeenv` to resolve the node binary then we also need |
111 | 157 | # to ensure that `node` is in the PATH so that any install scripts that |
112 | 158 | # assume it is present will work. |
113 | | - env.update(PATH=_update_path_env(env=env, target_bin=binary.path.parent)) |
114 | | - node_args = [str(binary.path), *args] |
115 | | - elif binary.strategy == Strategy.GLOBAL: |
116 | | - node_args = [str(binary.path), *args] |
| 159 | + env.update(PATH=_update_path_env(env=env, target_bin=strategy.path.parent)) |
| 160 | + node_args = [str(strategy.path), *args] |
| 161 | + log.debug('Running nodeenv command with args: %s', node_args) |
| 162 | + return cast( |
| 163 | + 'subprocess.CompletedProcess[str] | subprocess.CompletedProcess[bytes]', |
| 164 | + subprocess.run(node_args, **kwargs), |
| 165 | + ) |
117 | 166 | else: |
118 | | - raise RuntimeError(f'Unknown strategy: {binary.strategy}') |
119 | | - |
120 | | - log.debug('Running node command with args: %s', node_args) |
121 | | - return cast( |
122 | | - 'subprocess.CompletedProcess[str] | subprocess.CompletedProcess[bytes]', |
123 | | - subprocess.run(node_args, env=env, **kwargs), |
124 | | - ) |
| 167 | + assert_never(strategy) |
125 | 168 |
|
126 | 169 |
|
127 | 170 | def version(target: Target) -> Tuple[int, ...]: |
|
0 commit comments