Coverage for src/prisma/cli/prisma.py: 79%

1from __future__ import annotations 

2 

3import os 

4import sys 

5import json 

6import logging 

7import subprocess 

8from typing import Any, Dict, List, Optional, NamedTuple 

9from pathlib import Path 

10 

11import click 

12 

13from .. import config 

14from ._node import npm, node 

15from ..errors import PrismaError 

16 

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

18 

19 

20def run( 

21 args: List[str], 

22 check: bool = False, 

23 env: Optional[Dict[str, str]] = None, 

24) -> int: 

25 log.debug('Running prisma command with args: %s', args) 

26 

27 default_env = { 

28 **os.environ, 

29 'PRISMA_HIDE_UPDATE_MESSAGE': 'true', 

30 'PRISMA_CLI_QUERY_ENGINE_TYPE': 'binary', 

31 } 

32 env = {**default_env, **env} if env is not None else default_env 

33 

34 # TODO: ensure graceful termination 

35 entrypoint = ensure_cached().entrypoint 

36 process = node.run( 

37 str(entrypoint), 

38 *args, 

39 env=env, 

40 check=check, 

41 stdout=sys.stdout, 

42 stderr=sys.stderr, 

43 ) 

44 

45 if args and args[0] in {'--help', '-h'}: 

46 click.echo(click.style('Python Commands\n', bold=True)) 

47 click.echo(' ' + 'For Prisma Client Python commands run ' + click.style('prisma py --help', bold=True)) 

48 

49 return process.returncode 

50 

51 

52class CLICache(NamedTuple): 

53 cache_dir: Path 

54 entrypoint: Path 

55 

56 

57DEFAULT_PACKAGE_JSON: dict[str, Any] = { 

58 'name': 'prisma-binaries', 

59 'version': '1.0.0', 

60 'private': True, 

61 'description': 'Cache directory created by Prisma Client Python to store Prisma Engines', 

62 'main': 'node_modules/prisma/build/index.js', 

63 'author': 'RobertCraigie', 

64 'license': 'Apache-2.0', 

65} 

66 

67 

68def ensure_cached() -> CLICache: 

69 cache_dir = config.binary_cache_dir 

70 entrypoint = cache_dir / 'node_modules' / 'prisma' / 'build' / 'index.js' 

71 

72 if not cache_dir.exists(): 

73 cache_dir.mkdir(parents=True) 

74 

75 # We need to create a dummy `package.json` file so that `npm` doesn't try 

76 # and search for it elsewhere. 

77 # 

78 # If it finds a different `package.json` file then the `prisma` package 

79 # will be installed there instead of our cache directory. 

80 package = cache_dir / 'package.json' 

81 if not package.exists(): 

82 package.write_text(json.dumps(DEFAULT_PACKAGE_JSON)) 

83 

84 if not entrypoint.exists(): 

85 click.echo('Installing Prisma CLI') 

86 

87 try: 

88 proc = npm.run( 

89 'install', 

90 f'prisma@{config.prisma_version}', 

91 cwd=config.binary_cache_dir, 

92 stdout=subprocess.PIPE, 

93 stderr=subprocess.STDOUT, 

94 ) 

95 if proc.returncode != 0: 95 ↛ 96line 95 didn't jump to line 96, because the condition on line 95 was never true

96 click.echo( 

97 f'An error ocurred while installing the Prisma CLI; npm install log: {proc.stdout.decode("utf-8")}' 

98 ) 

99 proc.check_returncode() 

100 except Exception: 

101 # as we use the entrypoint existing to check whether or not we should run `npm install` 

102 # we need to make sure it doesn't exist if running `npm install` fails as it will otherwise 

103 # lead to a broken state, https://github.com/RobertCraigie/prisma-client-py/issues/705 

104 if entrypoint.exists(): 

105 try: 

106 entrypoint.unlink() 

107 except Exception: 

108 pass 

109 raise 

110 

111 if not entrypoint.exists(): 111 ↛ 112line 111 didn't jump to line 112, because the condition on line 111 was never true

112 raise PrismaError( 

113 f'CLI installation appeared to complete but the expected entrypoint ({entrypoint}) could not be found.' 

114 ) 

115 

116 return CLICache(cache_dir=cache_dir, entrypoint=entrypoint)