Coverage for graphqler / utils / plugins_handler.py: 68%

1import importlib 

2import importlib.util 

3import inspect 

4import pathlib 

5 

6from graphqler import config 

7from graphqler.utils import request_utils 

8from graphqler.utils.protocols.request_utils_protocol import RequestUtilsProtocol 

9 

10# The possible plugins and their map to the original module in GraphQLer 

11POSSIBLE_PLUGINS = { 

12 "request_utils.py": request_utils 

13} 

14 

15 

16def get_plugin_path(plugin_name: str) -> pathlib.Path: 

17 """Gets the plugin path 

18 

19 Args: 

20 plugin_name (str): The plugin name 

21 

22 Returns: 

23 pathlib.Path: The plugin path 

24 """ 

25 return config.PLUGINS_PATH / pathlib.Path(plugin_name) 

26 

27 

28def does_plugin_exist(plugin_name: str) -> bool: 

29 """Checks if the plugin exists 

30 

31 Args: 

32 plugin_name (str): The name of the plugin 

33 

34 Returns: 

35 bool: Whether the plugin exists 

36 """ 

37 plugins_path = get_plugin_path(plugin_name) 

38 if not plugins_path.exists(): 

39 return False 

40 

41 plugin_file_path = get_plugin_path(plugin_name) 

42 return plugin_file_path.is_file() 

43 

44 

45def get_plugin(plugin_name: str): 

46 if does_plugin_exist(plugin_name): 

47 plugin_path = get_plugin_path(plugin_name) 

48 spec = importlib.util.spec_from_file_location(plugin_name.split('.')[0], plugin_path) 

49 if spec: 

50 module = importlib.util.module_from_spec(spec) 

51 if module and spec.loader: 

52 spec.loader.exec_module(module) 

53 return module 

54 else: 

55 return POSSIBLE_PLUGINS[plugin_name] 

56 else: 

57 return POSSIBLE_PLUGINS[plugin_name] 

58 else: 

59 return POSSIBLE_PLUGINS[plugin_name] 

60 

61 

62def get_request_utils() -> RequestUtilsProtocol: 

63 """Gets the request utils plugin (if it exists) and ensures it conforms to the protocol 

64 

65 Returns: 

66 RequestUtilsProtocol: The request utils protocol 

67 """ 

68 plugin_name = "request_utils.py" 

69 original_module = POSSIBLE_PLUGINS[plugin_name] 

70 new_module = get_plugin(plugin_name) 

71 

72 original_functions = inspect.getmembers(original_module, inspect.isfunction) 

73 

74 for name, func in original_functions: 

75 if not hasattr(new_module, name): 

76 setattr(new_module, name, func) 

77 

78 # Ensure new_module conforms to the protocol 

79 assert isinstance(new_module, RequestUtilsProtocol), f"Module {plugin_name} does not match expected interface" 

80 

81 return new_module