Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/anyio/_lazyimport.py: 44%

1from __future__ import annotations 

2 

3__all__ = ( 

4 "fix_package_names", 

5 "install_lazy_importer", 

6 "set_deprecated_aliases", 

7) 

8 

9import ast 

10import inspect 

11import sys 

12import warnings 

13from importlib import import_module 

14from types import ModuleType 

15from typing import Any 

16 

17 

18def install_lazy_importer() -> bool: 

19 module_globals = sys._getframe(1).f_globals 

20 module_name = module_globals["__name__"] 

21 module_prefix = module_name + "." 

22 module = sys.modules[module_name] 

23 lazy_map, deprecated_aliases, submodule_names = _build_lazy_map(module) 

24 names = sorted(lazy_map) 

25 

26 # Delete symbols that are not part of the API 

27 del module_globals["TYPE_CHECKING"] 

28 del module_globals["install_lazy_importer"] 

29 

30 if not lazy_map and not deprecated_aliases: 

31 return False 

32 

33 def __getattr__(name: str) -> Any: 

34 if new_name := deprecated_aliases.get(name): 

35 emit_deprecation_warning(module_name, name, new_name) 

36 target_mod, target_attr = new_name.rsplit(".", 1) 

37 elif name in submodule_names: 

38 target_mod, target_attr = "." + name, "" 

39 else: 

40 try: 

41 target_mod, target_attr = lazy_map[name] 

42 except KeyError: 

43 raise AttributeError( 

44 f"module {module_name!r} has no attribute {name!r}" 

45 ) from None 

46 

47 imported = import_module(target_mod, module_name) 

48 value = getattr(imported, target_attr) if target_attr else imported 

49 

50 # patch the module name to match 

51 if ( 

52 getattr(value, "__module__", "").startswith(module_prefix) 

53 and name not in deprecated_aliases 

54 ): 

55 value.__module__ = module_name 

56 

57 module_globals[name] = value 

58 return value 

59 

60 def __dir__() -> list[str]: 

61 return names 

62 

63 module_globals["__dir__"] = __dir__ 

64 module_globals["__getattr__"] = __getattr__ 

65 module_globals.pop("fix_package_names", None) 

66 module_globals.pop("set_deprecated_aliases", None) 

67 return True 

68 

69 

70def fix_package_names() -> None: 

71 module_globals = sys._getframe(1).f_globals 

72 module_prefix = module_globals["__name__"] + "." 

73 del module_globals[fix_package_names.__name__] 

74 for value in module_globals.values(): 

75 if modname := getattr(value, "__module__", ""): 

76 if modname.startswith(module_prefix): 

77 parts = modname.split(".") 

78 value.__module__ = ".".join( 

79 part for part in parts if not part.startswith("_") 

80 ) 

81 

82 

83def emit_deprecation_warning(module_name: str, name: str, target: str) -> None: 

84 warnings.warn( 

85 f"The {module_name}.{name} alias is deprecated, use {target} instead.", 

86 DeprecationWarning, 

87 stacklevel=3, 

88 ) 

89 

90 

91def set_deprecated_aliases(aliases: dict[str, str]) -> None: 

92 module_globals = sys._getframe(1).f_globals 

93 module_name = module_globals["__name__"] 

94 del module_globals[set_deprecated_aliases.__name__] 

95 

96 def __getattr__(name: str) -> Any: 

97 try: 

98 target = aliases[name] 

99 except KeyError: 

100 raise AttributeError( 

101 f"module {module_name!r} has no attribute {name!r}" 

102 ) from None 

103 

104 emit_deprecation_warning(module_name, name, target) 

105 target_modname, attrname = target.rsplit(".", 1) 

106 module = import_module(target_modname) 

107 return getattr(module, attrname) 

108 

109 sys.modules[module_name].__dict__["__getattr__"] = __getattr__ 

110 

111 

112def _build_lazy_map( 

113 module: ModuleType, 

114) -> tuple[dict[str, tuple[str, str]], dict[str, str], list[str]]: 

115 try: 

116 source = inspect.getsource(module) 

117 except OSError: 

118 return {}, {}, [] 

119 

120 tree = compile(source, module.__file__ or "", "exec", ast.PyCF_ONLY_AST) 

121 assert isinstance(tree, ast.Module) 

122 out: dict[str, tuple[str, str]] = {} 

123 deprecated_aliases: dict[str, str] = {} 

124 submodule_names: list[str] = [] 

125 

126 for node in tree.body: 

127 if not isinstance(node, ast.If) or not _is_type_checking_block(node.test): 

128 continue 

129 

130 for stmt in node.body: 

131 match stmt: 

132 case ast.ImportFrom(): 

133 if stmt.module is None: 

134 submodule_names.extend(alias.name for alias in stmt.names) 

135 else: 

136 base = "." * stmt.level + (stmt.module or "") 

137 for alias in stmt.names: 

138 if alias.name == "*": 

139 raise RuntimeError("star imports not supported") 

140 

141 exported = alias.asname or alias.name 

142 out[exported] = (base, alias.name) 

143 case ast.Expr() if isinstance(stmt.value, ast.Call): 

144 call = stmt.value 

145 if ( 

146 isinstance(call.func, ast.Name) 

147 and call.func.id == "set_deprecated_aliases" 

148 ): 

149 arg0 = call.args[0] 

150 assert isinstance(arg0, ast.Dict) 

151 for key, value in zip(arg0.keys, arg0.values, strict=True): 

152 assert isinstance(key, ast.Constant) 

153 assert isinstance(key.value, str) 

154 assert isinstance(value, ast.Constant) 

155 assert isinstance(value.value, str) 

156 deprecated_aliases[key.value] = value.value 

157 

158 return out, deprecated_aliases, submodule_names 

159 

160 

161def _is_type_checking_block(test: ast.AST) -> bool: 

162 if not isinstance(test, ast.BoolOp): 

163 return False 

164 

165 subtest = test.values[0] 

166 match subtest: 

167 case ast.Name(): 

168 return subtest.id == "TYPE_CHECKING" 

169 case ast.Attribute(): 

170 return ( 

171 isinstance(subtest.value, ast.Name) 

172 and subtest.value.id == "typing" 

173 and subtest.attr == "TYPE_CHECKING" 

174 ) 

175 case _: 

176 return False