Coverage for packages / dqm-ml / src / dqm_ml / dependency.py: 80%

1"""Dependency management utilities for DQM-ML v2. 

2 

3This module provides functions for handling optional dependencies, 

4displaying version information, and discovering available commands. 

5""" 

6 

7from collections.abc import Generator 

8from contextlib import contextmanager 

9import logging 

10from typing import Any 

11 

12from dqm_ml_core import PluginLoadedRegistry 

13from dqm_ml_core._version_ import version as core_version 

14 

15logger = logging.getLogger(__name__) 

16 

17 

18@contextmanager 

19def optional_dependencies(error: str = "ignore") -> Generator[None, None, None]: 

20 """Context manager for handling optional dependencies. 

21 

22 Args: 

23 error: How to handle missing optional dependencies: 

24 - "ignore": Silently continue (default) 

25 - "warn": Print a warning message 

26 - "raise": Raise the ImportError 

27 

28 Yields: 

29 None - this is a context manager that doesn't provide any value. 

30 """ 

31 assert error in {"raise", "warn", "ignore"} 

32 try: 

33 yield None 

34 except ImportError as e: 

35 if error == "raise": 

36 raise e 

37 if error == "warn": 

38 msg = f'Missing optional dependency "{e.name}". Use pip to install.' 

39 print(f"Warning: {msg}") 

40 

41 

42def display_version(arg_list: list[str] | None = None) -> None: 

43 print(f"DQM-ML version : {core_version}") 

44 

45 

46def display_list_of(arg_list: list[str] | None = None) -> None: 

47 # TODO : we display all but we can filter / use extra parameters 

48 print("Available data metrics_registry") 

49 for key, value in PluginLoadedRegistry.get_metrics_registry().items(): 

50 print(f"- {key} - {value}") 

51 

52 print("Available data loaders") 

53 for key, value in PluginLoadedRegistry.get_dataloaders_registry().items(): 

54 print(f"- {key} - {value}") 

55 

56 print("Available outputs writers") 

57 for key, value in PluginLoadedRegistry.get_outputwriter_registry().items(): 

58 print(f"- {key} - {value}") 

59 

60 

61def get_available_command() -> dict[str, Any]: 

62 command_list = {"version": display_version, "list": display_list_of} 

63 optional_dep_mode = "warn" 

64 

65 # We import available command for dqml cli 

66 with optional_dependencies(optional_dep_mode): 

67 from dqm_ml_job._version_ import version as pipeline_version 

68 from dqm_ml_job.cli import execute 

69 

70 logger.debug(f"Different dqm-ml-job version {pipeline_version}") 

71 command_list["process"] = execute 

72 

73 return command_list