Coverage for src/prisma/cli/_node.py: 89%

1from __future__ import annotations 

2 

3import os 

4import re 

5import sys 

6import shutil 

7import logging 

8import subprocess 

9from abc import ABC, abstractmethod 

10from typing import IO, Any, Union, Mapping, cast 

11from pathlib import Path 

12from typing_extensions import Literal, override 

13 

14from .. import config 

15from .._proxy import LazyProxy 

16from ..errors import PrismaError 

17from .._compat import nodejs, get_args 

18from ..binaries import platform 

19 

20log: logging.Logger = logging.getLogger(__name__) 

21File = Union[int, IO[Any]] 

22Target = Literal['node', 'npm'] 

23 

24# taken from https://github.com/prisma/prisma/blob/main/package.json 

25MIN_NODE_VERSION = (16, 13) 

26 

27# mapped the node version above from https://nodejs.org/en/download/releases/ 

28MIN_NPM_VERSION = (6, 14) 

29 

30# we only care about the first two entries in the version number 

31VERSION_RE = re.compile(r'v?(\d+)(?:\.?(\d+))') 

32 

33 

34# TODO: remove the possibility to get mismatched paths for `node` and `npm` 

35 

36 

37class UnknownTargetError(PrismaError): 

38 def __init__(self, *, target: str) -> None: 

39 super().__init__(f'Unknown target: {target}; Valid choices are: {", ".join(get_args(cast(type, Target)))}') 

40 

41 

42# TODO: add tests for this error 

43class MissingNodejsBinError(PrismaError): 

44 def __init__(self) -> None: 

45 super().__init__( 

46 'Attempted to access a function that requires the `nodejs-bin` package to be installed but it is not.' 

47 ) 

48 

49 

50class Strategy(ABC): 

51 # TODO: support more options 

52 def run( 

53 self, 

54 *args: str, 

55 check: bool = False, 

56 cwd: Path | None = None, 

57 stdout: File | None = None, 

58 stderr: File | None = None, 

59 env: Mapping[str, str] | None = None, 

60 ) -> subprocess.CompletedProcess[bytes]: 

61 """Call the underlying Node.js binary. 

62 

63 The interface for this function is very similar to `subprocess.run()`. 

64 """ 

65 return self.__run__( 

66 *args, 

67 check=check, 

68 cwd=cwd, 

69 stdout=stdout, 

70 stderr=stderr, 

71 env=_update_path_env( 

72 env=env, 

73 target_bin=self.target_bin, 

74 ), 

75 ) 

76 

77 @abstractmethod 

78 def __run__( 

79 self, 

80 *args: str, 

81 check: bool = False, 

82 cwd: Path | None = None, 

83 stdout: File | None = None, 

84 stderr: File | None = None, 

85 env: Mapping[str, str] | None = None, 

86 ) -> subprocess.CompletedProcess[bytes]: 

87 """Call the underlying Node.js binary. 

88 

89 This should not be directly accessed, the `run()` function should be used instead. 

90 """ 

91 

92 @property 

93 @abstractmethod 

94 def target_bin(self) -> Path: 

95 """Property containing the location of the `bin` directory for the resolved node installation. 

96 

97 This is used to dynamically alter the `PATH` environment variable to give the appearance that Node 

98 is installed globally on the machine as this is a requirement of Prisma's installation step, see this 

99 comment for more context: https://github.com/RobertCraigie/prisma-client-py/pull/454#issuecomment-1280059779 

100 """ 

101 ... 

102 

103 

104class NodeBinaryStrategy(Strategy): 

105 target: Target 

106 resolver: Literal['global', 'nodeenv'] 

107 

108 def __init__( 

109 self, 

110 *, 

111 path: Path, 

112 target: Target, 

113 resolver: Literal['global', 'nodeenv'], 

114 ) -> None: 

115 self.path = path 

116 self.target = target 

117 self.resolver = resolver 

118 

119 @property 

120 @override 

121 def target_bin(self) -> Path: 

122 return self.path.parent 

123 

124 @override 

125 def __run__( 

126 self, 

127 *args: str, 

128 check: bool = False, 

129 cwd: Path | None = None, 

130 stdout: File | None = None, 

131 stderr: File | None = None, 

132 env: Mapping[str, str] | None = None, 

133 ) -> subprocess.CompletedProcess[bytes]: 

134 path = str(self.path.absolute()) 

135 log.debug('Executing binary at %s with args: %s', path, args) 

136 return subprocess.run( 

137 [path, *args], 

138 check=check, 

139 cwd=cwd, 

140 env=env, 

141 stdout=stdout, 

142 stderr=stderr, 

143 ) 

144 

145 @classmethod 

146 def resolve(cls, target: Target) -> NodeBinaryStrategy: 

147 path = None 

148 if config.use_global_node: 

149 path = _get_global_binary(target) 

150 

151 if path is not None: 

152 return NodeBinaryStrategy( 

153 path=path, 

154 target=target, 

155 resolver='global', 

156 ) 

157 

158 return NodeBinaryStrategy.from_nodeenv(target) 

159 

160 @classmethod 

161 def from_nodeenv(cls, target: Target) -> NodeBinaryStrategy: 

162 cache_dir = config.nodeenv_cache_dir.absolute() 

163 if cache_dir.exists(): 

164 log.debug( 

165 'Skipping nodeenv installation as it already exists at %s', 

166 cache_dir, 

167 ) 

168 else: 

169 log.debug('Installing nodeenv to %s', cache_dir) 

170 try: 

171 subprocess.run( 

172 [ 

173 sys.executable, 

174 '-m', 

175 'nodeenv', 

176 str(cache_dir), 

177 *config.nodeenv_extra_args, 

178 ], 

179 check=True, 

180 stdout=sys.stdout, 

181 stderr=sys.stderr, 

182 ) 

183 except Exception as exc: 

184 print( # noqa: T201 

185 'nodeenv installation failed; You may want to try installing `nodejs-bin` as it is more reliable.', 

186 file=sys.stderr, 

187 ) 

188 raise exc 

189 

190 if not cache_dir.exists(): 190 ↛ 191line 190 didn't jump to line 191, because the condition on line 190 was never true

191 raise RuntimeError( 

192 'Could not install nodeenv to the expected directory; See the output above for more details.' 

193 ) 

194 

195 # TODO: what hapens on cygwin? 

196 if platform.name() == 'windows': 

197 bin_dir = cache_dir / 'Scripts' 

198 if target == 'node': 

199 path = bin_dir / 'node.exe' 

200 else: 

201 path = bin_dir / f'{target}.cmd' 

202 else: 

203 path = cache_dir / 'bin' / target 

204 

205 if target == 'npm': 

206 return cls(path=path, resolver='nodeenv', target=target) 

207 elif target == 'node': 207 ↛ 210line 207 didn't jump to line 210, because the condition on line 207 was never false

208 return cls(path=path, resolver='nodeenv', target=target) 

209 else: 

210 raise UnknownTargetError(target=target) 

211 

212 

213class NodeJSPythonStrategy(Strategy): 

214 target: Target 

215 resolver: Literal['nodejs-bin'] 

216 

217 def __init__(self, *, target: Target) -> None: 

218 self.target = target 

219 self.resolver = 'nodejs-bin' 

220 

221 @override 

222 def __run__( 

223 self, 

224 *args: str, 

225 check: bool = False, 

226 cwd: Path | None = None, 

227 stdout: File | None = None, 

228 stderr: File | None = None, 

229 env: Mapping[str, str] | None = None, 

230 ) -> subprocess.CompletedProcess[bytes]: 

231 if nodejs is None: 231 ↛ 232line 231 didn't jump to line 232, because the condition on line 231 was never true

232 raise MissingNodejsBinError() 

233 

234 func = None 

235 if self.target == 'node': 

236 func = nodejs.node.run 

237 elif self.target == 'npm': 237 ↛ 240line 237 didn't jump to line 240, because the condition on line 237 was never false

238 func = nodejs.npm.run 

239 else: 

240 raise UnknownTargetError(target=self.target) 

241 

242 return cast( 

243 'subprocess.CompletedProcess[bytes]', 

244 func( 

245 args, 

246 check=check, 

247 cwd=cwd, 

248 env=env, 

249 stdout=stdout, 

250 stderr=stderr, 

251 ), 

252 ) 

253 

254 @property 

255 def node_path(self) -> Path: 

256 """Returns the path to the `node` binary""" 

257 if nodejs is None: 257 ↛ 258line 257 didn't jump to line 258, because the condition on line 257 was never true

258 raise MissingNodejsBinError() 

259 

260 return Path(nodejs.node.path) 

261 

262 @property 

263 @override 

264 def target_bin(self) -> Path: 

265 return Path(self.node_path).parent 

266 

267 

268Node = Union[NodeJSPythonStrategy, NodeBinaryStrategy] 

269 

270 

271def resolve(target: Target) -> Node: 

272 if target not in {'node', 'npm'}: 

273 raise UnknownTargetError(target=target) 

274 

275 if config.use_nodejs_bin: 

276 log.debug('Checking if nodejs-bin is installed') 

277 if nodejs is not None: 277 ↛ 281line 277 didn't jump to line 281, because the condition on line 277 was never false

278 log.debug('Using nodejs-bin with version: %s', nodejs.node_version) 

279 return NodeJSPythonStrategy(target=target) 

280 

281 return NodeBinaryStrategy.resolve(target) 

282 

283 

284def _update_path_env( 

285 *, 

286 env: Mapping[str, str] | None, 

287 target_bin: Path, 

288 sep: str = os.pathsep, 

289) -> dict[str, str]: 

290 """Returns a modified version of `os.environ` with the `PATH` environment variable updated 

291 to include the location of the downloaded Node binaries. 

292 """ 

293 if env is None: 

294 env = dict(os.environ) 

295 

296 log.debug('Attempting to preprend %s to the PATH', target_bin) 

297 assert target_bin.exists(), 'Target `bin` directory does not exist' 

298 

299 path = env.get('PATH', '') or os.environ.get('PATH', '') 

300 if path: 300 ↛ 308line 300 didn't jump to line 308, because the condition on line 300 was never false

301 # handle the case where the PATH already starts with the separator (this probably shouldn't happen) 

302 if path.startswith(sep): 

303 path = f'{target_bin.absolute()}{path}' 

304 else: 

305 path = f'{target_bin.absolute()}{sep}{path}' 

306 else: 

307 # handle the case where there is no PATH set (unlikely / impossible to actually happen?) 

308 path = str(target_bin.absolute()) 

309 

310 log.debug('Using PATH environment variable: %s', path) 

311 return {**env, 'PATH': path} 

312 

313 

314def _get_global_binary(target: Target) -> Path | None: 

315 """Returns the path to a globally installed binary. 

316 

317 This also ensures that the binary is of the right version. 

318 """ 

319 log.debug('Checking for global target binary: %s', target) 

320 

321 which = shutil.which(target) 

322 if which is None: 322 ↛ 323line 322 didn't jump to line 323, because the condition on line 322 was never true

323 log.debug('Global target binary: %s not found', target) 

324 return None 

325 

326 log.debug('Found global binary at: %s', which) 

327 

328 path = Path(which) 

329 if not path.exists(): 329 ↛ 330line 329 didn't jump to line 330, because the condition on line 329 was never true

330 log.debug('Global binary does not exist at: %s', which) 

331 return None 

332 

333 if not _should_use_binary(target=target, path=path): 333 ↛ 334line 333 didn't jump to line 334, because the condition on line 333 was never true

334 return None 

335 

336 log.debug('Using global %s binary at %s', target, path) 

337 return path 

338 

339 

340def _should_use_binary(target: Target, path: Path) -> bool: 

341 """Call the binary at `path` with a `--version` flag to check if it matches our minimum version requirements. 

342 

343 This only applies to the global node installation as: 

344 

345 - the minimum version of `nodejs-bin` is higher than our requirement 

346 - `nodeenv` defaults to the latest stable version of node 

347 """ 

348 if target == 'node': 

349 min_version = MIN_NODE_VERSION 

350 elif target == 'npm': 

351 min_version = MIN_NPM_VERSION 

352 else: 

353 raise UnknownTargetError(target=target) 

354 

355 version = _get_binary_version(target, path) 

356 if version is None: 356 ↛ 357line 356 didn't jump to line 357, because the condition on line 356 was never true

357 log.debug( 

358 'Could not resolve %s version, ignoring global %s installation', 

359 target, 

360 target, 

361 ) 

362 return False 

363 

364 if version < min_version: 

365 log.debug( 

366 'Global %s version (%s) is lower than the minimum required version (%s), ignoring', 

367 target, 

368 version, 

369 min_version, 

370 ) 

371 return False 

372 

373 return True 

374 

375 

376def _get_binary_version(target: Target, path: Path) -> tuple[int, ...] | None: 

377 proc = subprocess.run( 

378 [str(path), '--version'], 

379 stdout=subprocess.PIPE, 

380 stderr=subprocess.STDOUT, 

381 check=False, 

382 ) 

383 log.debug('%s version check exited with code %s', target, proc.returncode) 

384 

385 output = proc.stdout.decode('utf-8').rstrip('\n') 

386 log.debug('%s version check output: %s', target, output) 

387 

388 match = VERSION_RE.search(output) 

389 if not match: 

390 return None 

391 

392 version = tuple(int(value) for value in match.groups()) 

393 log.debug('%s version check returning %s', target, version) 

394 return version 

395 

396 

397class LazyBinaryProxy(LazyProxy[Node]): 

398 target: Target 

399 

400 def __init__(self, target: Target) -> None: 

401 super().__init__() 

402 self.target = target 

403 

404 @override 

405 def __load__(self) -> Node: 

406 return resolve(self.target) 

407 

408 

409npm = LazyBinaryProxy('npm').__as_proxied__() 

410node = LazyBinaryProxy('node').__as_proxied__()