Coverage for faststream / _internal / cli / utils / imports.py: 69%

1import importlib (empty)

2from importlib.util import module_from_spec, spec_from_file_location (empty)

3from pathlib import Path (empty)

4 

5import typer (empty)

6 

7from faststream.exceptions import SetupError (empty)

8 

9 

10def import_from_string( (empty)

11 import_str: str, 

12 *, 

13 is_factory: bool = False, 

14) -> tuple[Path, object]: 

15 module_path, instance = _import_object_or_factory(import_str) (empty)

16 

17 if is_factory: 17 ↛ 18line 17 didn't jump to line 18 because the condition on line 17 was never true(empty)

18 if callable(instance): 

19 instance = instance() 

20 else: 

21 msg = f'"{instance}" is not a factory.' 

22 raise typer.BadParameter(msg) 

23 

24 return module_path, instance (empty)

25 

26 

27def _is_missing_requested_module( (empty)

28 module_str: str, 

29 exc: ModuleNotFoundError, 

30) -> bool: 

31 if exc.name is None: 31 ↛ 32line 31 didn't jump to line 32 because the condition on line 31 was never true(empty)

32 return False 

33 

34 requested_module = module_str.split(".") (empty)

35 missing_module = exc.name.split(".") (empty)

36 

37 return missing_module == requested_module[: len(missing_module)] (empty)

38 

39 

40def _import_object_or_factory(import_str: str) -> tuple[Path, object]: (empty)

41 """Import FastStream application from module specified by a string.""" 

42 if not isinstance(import_str, str): 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true(empty)

43 msg = "Given value is not of type string" 

44 raise typer.BadParameter(msg) 

45 

46 module_str, _, attrs_str = import_str.partition(":") (empty)

47 if not module_str or not attrs_str: (empty)

48 msg = f'Import string "{import_str}" must be in format "<module>:<attribute>"' (empty)

49 raise typer.BadParameter( (empty)

50 msg, 

51 ) 

52 

53 try: (empty)

54 module = importlib.import_module( # nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import (empty)

55 module_str, 

56 ) 

57 

58 except ModuleNotFoundError as e: (empty)

59 if not _is_missing_requested_module(module_str, e): (empty)

60 raise (empty)

61 

62 module_path, import_obj_name = _get_obj_path(import_str) 

63 instance = _try_import_app(module_path, import_obj_name) 

64 

65 else: 

66 attr = module (empty)

67 try: (empty)

68 for attr_str in attrs_str.split("."): (empty)

69 attr = getattr(attr, attr_str) (empty)

70 instance = attr (empty)

71 

72 except AttributeError as e: (empty)

73 typer.echo(e, err=True) (empty)

74 msg = f'Attribute "{attrs_str}" not found in module "{module_str}".' (empty)

75 raise typer.BadParameter( (empty)

76 msg, 

77 ) from e 

78 

79 if module.__file__: 79 ↛ 82line 79 didn't jump to line 82 because the condition on line 79 was always true(empty)

80 module_path = Path(module.__file__).resolve().parent (empty)

81 else: 

82 module_path = Path.cwd() 

83 

84 return module_path, instance (empty)

85 

86 

87def _try_import_app(module: Path, app: str) -> object: (empty)

88 """Tries to import a FastStream app from a module.""" 

89 try: 

90 app_object = _import_object(module, app) 

91 

92 except FileNotFoundError as e: 

93 typer.echo(e, err=True) 

94 msg = "Please, input module like [python_file:docs_object] or [module:attribute]" 

95 raise typer.BadParameter( 

96 msg, 

97 ) from e 

98 

99 else: 

100 return app_object 

101 

102 

103def _import_object(module: Path, app: str) -> object: (empty)

104 """Import an object from a module.""" 

105 spec = spec_from_file_location( (empty)

106 "mode", 

107 f"{module}.py", 

108 submodule_search_locations=[str(module.parent.absolute())], 

109 ) 

110 

111 if spec is None: # pragma: no cover (empty)

112 raise FileNotFoundError(module) 

113 

114 mod = module_from_spec(spec) (empty)

115 loader = spec.loader (empty)

116 

117 if loader is None: # pragma: no cover (empty)

118 msg = f"{spec} has no loader" 

119 raise SetupError(msg) 

120 

121 loader.exec_module(mod) (empty)

122 

123 try: 

124 obj = getattr(mod, app) 

125 except AttributeError as e: 

126 raise FileNotFoundError(module) from e 

127 

128 return obj 

129 

130 

131def _get_obj_path(obj_path: str) -> tuple[Path, str]: (empty)

132 """Get the application path.""" 

133 if ":" not in obj_path: (empty)

134 msg = f"`{obj_path}` is not a path to object" (empty)

135 raise SetupError(msg) (empty)

136 

137 module, app_name = obj_path.split(":", 2) (empty)

138 

139 mod_path = Path.cwd() (empty)

140 for i in module.split("."): (empty)

141 mod_path /= i (empty)

142 

143 return mod_path, app_name (empty)