Coverage for /usr/local/lib/python3.12/site-packages/prefect/_internal/installation.py: 17%

1import importlib 1 ctx1a

2import subprocess 1 ctx1a

3import sys 1 ctx1a

4 

5import prefect.utilities.processutils 1 ctx1a

6 

7 

8def install_packages( 1 ctx1a

9 packages: list[str], stream_output: bool = False, upgrade: bool = False 

10): 

11 """ 

12 Install packages using uv if available, otherwise use pip. 

13 """ 

14 base_command = ["pip", "install", *packages] 

15 if upgrade: 

16 base_command.append("--upgrade") 

17 if stream_output: 

18 stdout = sys.stdout 

19 stderr = sys.stderr 

20 else: 

21 stdout = subprocess.DEVNULL 

22 stderr = subprocess.DEVNULL 

23 try: 

24 uv = importlib.import_module("uv") 

25 

26 command = [uv.find_uv_bin(), *base_command] 

27 subprocess.check_call( 

28 command, 

29 stdout=stdout, 

30 stderr=stderr, 

31 ) 

32 except (ImportError, ModuleNotFoundError, FileNotFoundError): 

33 command = [sys.executable, "-m", *base_command] 

34 subprocess.check_call( 

35 command, 

36 stdout=stdout, 

37 stderr=stderr, 

38 ) 

39 

40 

41async def ainstall_packages( 1 ctx1a

42 packages: list[str], stream_output: bool = False, upgrade: bool = False 

43): 

44 """ 

45 Install packages using uv if available, otherwise use pip. 

46 """ 

47 base_command = ["pip", "install", *packages] 

48 if upgrade: 

49 base_command.append("--upgrade") 

50 try: 

51 uv = importlib.import_module("uv") 

52 

53 await prefect.utilities.processutils.run_process( 

54 [uv.find_uv_bin(), *base_command], stream_output=stream_output 

55 ) 

56 except (ImportError, ModuleNotFoundError, FileNotFoundError): 

57 await prefect.utilities.processutils.run_process( 

58 [sys.executable, "-m", *base_command], 

59 stream_output=stream_output, 

60 )