|
| 1 | +""" |
| 2 | +Plugin to only import types when you call methods |
| 3 | +
|
| 4 | +This will massively reduce import times for larger projects since you only have |
| 5 | +to load the input types when loading the client. |
| 6 | +
|
| 7 | +All result types that's used to process the server response will only be |
| 8 | +imported when the method is called. |
| 9 | +""" |
| 10 | + |
| 11 | +import ast |
| 12 | +from typing import Dict |
| 13 | + |
| 14 | +from graphql import GraphQLSchema |
| 15 | + |
| 16 | +from ariadne_codegen import Plugin |
| 17 | + |
| 18 | + |
| 19 | +class NoGlobalImportsPlugin(Plugin): |
| 20 | + """Only import types when you call an endpoint needing it""" |
| 21 | + |
| 22 | + def __init__(self, schema: GraphQLSchema, config_dict: Dict) -> None: |
| 23 | + """Constructor""" |
| 24 | + # Types that should only be imported in a `TYPE_CHECKING` context. This |
| 25 | + # is all the types used as arguments to a method or as a return type, |
| 26 | + # i.e. for type checking. |
| 27 | + self.input_and_return_types: set[str] = set() |
| 28 | + |
| 29 | + # Imported classes are classes imported from local imports. We keep a |
| 30 | + # map between name and module so we know how to import them in each |
| 31 | + # method. |
| 32 | + self.imported_classes: dict[str, str] = {} |
| 33 | + |
| 34 | + # Imported classes in each method definition. |
| 35 | + self.imported_in_method: set[str] = set() |
| 36 | + |
| 37 | + super().__init__(schema, config_dict) |
| 38 | + |
| 39 | + def generate_client_module(self, module: ast.Module) -> ast.Module: |
| 40 | + """ |
| 41 | + Update the generated client. |
| 42 | +
|
| 43 | + This will parse all current imports to map them to a path. It will then |
| 44 | + traverse all methods and look for the actual return type. The return |
| 45 | + node will be converted to an `ast.Constant` if it's an `ast.Name` and |
| 46 | + the return type will be imported only under `if TYPE_CHECKING` |
| 47 | + conditions. |
| 48 | +
|
| 49 | + It will also move all imports of the types used to parse the response |
| 50 | + inside each method since that's the only place where they're used. The |
| 51 | + result will be that we end up with imports in the global scope only for |
| 52 | + types used as input types. |
| 53 | +
|
| 54 | + :param module: The ast for the module |
| 55 | + """ |
| 56 | + self._store_imported_classes(module.body) |
| 57 | + |
| 58 | + # Find the actual client class so we can grab all input and output |
| 59 | + # types. We also ensure to manipulate the ast while we do this. |
| 60 | + client_class_def = next( |
| 61 | + filter(lambda o: isinstance(o, ast.ClassDef), module.body), None |
| 62 | + ) |
| 63 | + if not client_class_def or not isinstance(client_class_def, ast.ClassDef): |
| 64 | + return super().generate_client_module(module) |
| 65 | + |
| 66 | + for method_def in [ |
| 67 | + m |
| 68 | + for m in client_class_def.body |
| 69 | + if isinstance(m, (ast.FunctionDef, ast.AsyncFunctionDef)) |
| 70 | + ]: |
| 71 | + method_def = self._rewrite_input_args_to_constants(method_def) |
| 72 | + |
| 73 | + # If the method returns anything, update whatever it returns. |
| 74 | + if method_def.returns: |
| 75 | + method_def.returns = self._update_name_to_constant(method_def.returns) |
| 76 | + |
| 77 | + self._insert_import_statement_in_method(method_def) |
| 78 | + |
| 79 | + self._update_imports(module) |
| 80 | + |
| 81 | + return super().generate_client_module(module) |
| 82 | + |
| 83 | + def _store_imported_classes(self, module_body: list[ast.stmt]): |
| 84 | + """Fetch and store imported classes. |
| 85 | +
|
| 86 | + Grab all imported classes with level 1 or starting with `.` because |
| 87 | + these are the ones generated by us. We store a map between the class and |
| 88 | + which module it was imported from so we can easily import it when |
| 89 | + needed. This can be in a `TYPE_CHECKING` condition or inside a method. |
| 90 | +
|
| 91 | + :param module_body: The body of an `ast.Module` |
| 92 | + """ |
| 93 | + for node in module_body: |
| 94 | + if not isinstance(node, ast.ImportFrom): |
| 95 | + continue |
| 96 | + |
| 97 | + if node.module is None: |
| 98 | + continue |
| 99 | + |
| 100 | + # We only care about local imports from our generated code. |
| 101 | + if node.level != 1 and not node.module.startswith("."): |
| 102 | + continue |
| 103 | + |
| 104 | + for name in node.names: |
| 105 | + from_ = "." * node.level + node.module |
| 106 | + if isinstance(name, ast.alias): |
| 107 | + self.imported_classes[name.name] = from_ |
| 108 | + |
| 109 | + def _rewrite_input_args_to_constants( |
| 110 | + self, method_def: ast.FunctionDef | ast.AsyncFunctionDef |
| 111 | + ) -> ast.FunctionDef | ast.AsyncFunctionDef: |
| 112 | + """Rewrite the arguments to a method. |
| 113 | +
|
| 114 | + For any `ast.Name` that requires an import convert it to an |
| 115 | + `ast.Constant` instead. The actual class will be noted and imported |
| 116 | + in a `TYPE_CHECKING` context. |
| 117 | +
|
| 118 | + :param method_def: Method definition |
| 119 | + :returns: The same definition but updated |
| 120 | + """ |
| 121 | + if not isinstance(method_def, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| 122 | + return method_def |
| 123 | + |
| 124 | + for i, input_arg in enumerate(method_def.args.args): |
| 125 | + annotation = input_arg.annotation |
| 126 | + if isinstance(annotation, (ast.Name, ast.Subscript, ast.Tuple)): |
| 127 | + method_def.args.args[i].annotation = self._update_name_to_constant( |
| 128 | + annotation |
| 129 | + ) |
| 130 | + |
| 131 | + return method_def |
| 132 | + |
| 133 | + def _insert_import_statement_in_method( |
| 134 | + self, method_def: ast.FunctionDef | ast.AsyncFunctionDef |
| 135 | + ): |
| 136 | + """Insert import statement in method. |
| 137 | +
|
| 138 | + Each method will eventually pass the returned value to a class we've |
| 139 | + generated. Since we only need it in the scope of the method ensure we |
| 140 | + add it at the top of the method only. It will be removed from the global |
| 141 | + scope. |
| 142 | +
|
| 143 | + :param method_def: The method definition to updated |
| 144 | + """ |
| 145 | + # Find the last statement in the body, the call to this class is |
| 146 | + # what we need to import first. |
| 147 | + return_stmt = method_def.body[-1] |
| 148 | + if isinstance(return_stmt, ast.Return): |
| 149 | + call = self._get_call_arg_from_return(return_stmt) |
| 150 | + elif isinstance(return_stmt, ast.AsyncFor): |
| 151 | + call = self._get_call_arg_from_async_for(return_stmt) |
| 152 | + else: |
| 153 | + return |
| 154 | + |
| 155 | + if call is None: |
| 156 | + return |
| 157 | + |
| 158 | + import_class = self._get_class_from_call(call) |
| 159 | + if import_class is None: |
| 160 | + return |
| 161 | + |
| 162 | + import_class_id = import_class.id |
| 163 | + |
| 164 | + # We add the class to our set of imported in methods - these classes |
| 165 | + # don't need to be imported at all in the global scope. |
| 166 | + self.imported_in_method.add(import_class.id) |
| 167 | + method_def.body.insert( |
| 168 | + 0, |
| 169 | + ast.ImportFrom( |
| 170 | + module=self.imported_classes[import_class_id], |
| 171 | + names=[import_class], |
| 172 | + ), |
| 173 | + ) |
| 174 | + |
| 175 | + def _get_call_arg_from_return(self, return_stmt: ast.Return) -> ast.Call | None: |
| 176 | + """Get the class used in the return statement. |
| 177 | +
|
| 178 | + :param return_stmt: The statement used for return |
| 179 | + """ |
| 180 | + # If it's a call of the class like produced by |
| 181 | + # `ShorterResultsPlugin` we have an attribute. |
| 182 | + if isinstance(return_stmt.value, ast.Attribute) and isinstance( |
| 183 | + return_stmt.value.value, ast.Call |
| 184 | + ): |
| 185 | + return return_stmt.value.value |
| 186 | + # If not it's just a call statement to the generated class. |
| 187 | + elif isinstance(return_stmt.value, ast.Call): |
| 188 | + return return_stmt.value |
| 189 | + |
| 190 | + return None |
| 191 | + |
| 192 | + def _get_call_arg_from_async_for(self, last_stmt: ast.AsyncFor) -> ast.Call | None: |
| 193 | + """Get the class used in the yield expression. |
| 194 | +
|
| 195 | + :param last_stmt: The statement used in `ast.AsyncFor` |
| 196 | + """ |
| 197 | + if isinstance(last_stmt.body, list) and isinstance(last_stmt.body[0], ast.Expr): |
| 198 | + body = last_stmt.body[0] |
| 199 | + elif isinstance(last_stmt.body, ast.Expr): |
| 200 | + body = last_stmt.body |
| 201 | + else: |
| 202 | + return None |
| 203 | + |
| 204 | + if not isinstance(body, ast.Expr): |
| 205 | + return None |
| 206 | + |
| 207 | + if not isinstance(body.value, ast.Yield): |
| 208 | + return None |
| 209 | + |
| 210 | + # If it's a call of the class like produced by |
| 211 | + # `ShorterResultsPlugin` we have an attribute. |
| 212 | + if isinstance(body.value.value, ast.Attribute) and isinstance( |
| 213 | + body.value.value.value, ast.Call |
| 214 | + ): |
| 215 | + return body.value.value.value |
| 216 | + # If not it's just a call statement to the generated class. |
| 217 | + elif isinstance(body.value.value, ast.Call): |
| 218 | + return body.value.value |
| 219 | + |
| 220 | + return None |
| 221 | + |
| 222 | + def _get_class_from_call(self, call: ast.Call) -> ast.Name | None: |
| 223 | + """Get the class from an `ast.Call`. |
| 224 | +
|
| 225 | + :param call: The `ast.Call` arg |
| 226 | + :returns: `ast.Name` or `None` |
| 227 | + """ |
| 228 | + if not isinstance(call.func, ast.Attribute): |
| 229 | + return None |
| 230 | + |
| 231 | + if not isinstance(call.func.value, ast.Name): |
| 232 | + return None |
| 233 | + |
| 234 | + return call.func.value |
| 235 | + |
| 236 | + def _update_imports(self, module: ast.Module) -> ast.Name | None: |
| 237 | + """Update all imports. |
| 238 | +
|
| 239 | + Iterate over all imports and remove the aliases that we use as input or |
| 240 | + return value. These will be moved and added to an `if TYPE_CHECKING` |
| 241 | + block. |
| 242 | +
|
| 243 | + **NOTE** If an `ast.ImportFrom` ends up without any names we must remove |
| 244 | + it completely otherwise formatting will not work (it would remove the |
| 245 | + empty `import from` but not format the rest of the code without running |
| 246 | + it twice). |
| 247 | +
|
| 248 | + We do this by storing all imports that we want to keep in an array, we |
| 249 | + then drop all from the body and re-insert the ones to keep. Lastly we |
| 250 | + import `TYPE_CHECKING` and add all our imports in the `if TYPE_CHECKING` |
| 251 | + block. |
| 252 | +
|
| 253 | + :param module: The ast for the whole module. |
| 254 | + """ |
| 255 | + # We now know all our input types and all our return types. The return |
| 256 | + # types that are _not_ used as import types should be in an `if |
| 257 | + # TYPE_CHECKING` import block. |
| 258 | + return_types_not_used_as_input = {k for k in self.input_and_return_types} |
| 259 | + |
| 260 | + # The ones we import in the method don't need to be imported at all - |
| 261 | + # unless that's the type we return. This behaviour can differ if you use |
| 262 | + # a plugin such as `ShorterResultsPlugin` that will import a type that |
| 263 | + # is different from the type returned. |
| 264 | + return_types_not_used_as_input.update( |
| 265 | + {k for k in self.imported_in_method if k not in self.input_and_return_types} |
| 266 | + ) |
| 267 | + |
| 268 | + if len(return_types_not_used_as_input) == 0: |
| 269 | + return None |
| 270 | + |
| 271 | + # We sadly have to iterate over all imports again and remove the imports |
| 272 | + # we will do conditionally. |
| 273 | + # It's very important that we get this right, if we keep any |
| 274 | + # `ImportFrom` that ends up without any names, the formatting will not |
| 275 | + # work! It will only remove the empty `import from` but not other unused |
| 276 | + # imports. |
| 277 | + non_empty_imports: list[ast.Import | ast.ImportFrom] = [] |
| 278 | + last_import_at = 0 |
| 279 | + for i, node in enumerate(module.body): |
| 280 | + if isinstance(node, ast.Import): |
| 281 | + last_import_at = i |
| 282 | + non_empty_imports.append(node) |
| 283 | + |
| 284 | + if not isinstance(node, ast.ImportFrom): |
| 285 | + continue |
| 286 | + |
| 287 | + last_import_at = i |
| 288 | + reduced_names = [] |
| 289 | + for name in node.names: |
| 290 | + if name.name not in return_types_not_used_as_input: |
| 291 | + reduced_names.append(name) |
| 292 | + |
| 293 | + node.names = reduced_names |
| 294 | + |
| 295 | + if len(reduced_names) > 0: |
| 296 | + non_empty_imports.append(node) |
| 297 | + |
| 298 | + # We can now remove all imports and re-insert the ones that's not empty. |
| 299 | + module.body = non_empty_imports + module.body[last_import_at + 1 :] |
| 300 | + |
| 301 | + # Create import to use for type checking. These will be put in an `if |
| 302 | + # TYPE_CHECKING` block. |
| 303 | + type_checking_imports = {} |
| 304 | + for cls in self.input_and_return_types: |
| 305 | + module_name = self.imported_classes[cls] |
| 306 | + if module_name not in type_checking_imports: |
| 307 | + type_checking_imports[module_name] = ast.ImportFrom( |
| 308 | + module=module_name, names=[] |
| 309 | + ) |
| 310 | + |
| 311 | + type_checking_imports[module_name].names.append(ast.alias(cls)) |
| 312 | + |
| 313 | + import_if_type_checking = ast.If( |
| 314 | + test=ast.Name(id="TYPE_CHECKING"), |
| 315 | + body=list(type_checking_imports.values()), |
| 316 | + orelse=[], |
| 317 | + ) |
| 318 | + |
| 319 | + module.body.insert(len(non_empty_imports), import_if_type_checking) |
| 320 | + |
| 321 | + # Import `TYPE_CHECKING`. |
| 322 | + module.body.insert( |
| 323 | + len(non_empty_imports), |
| 324 | + ast.ImportFrom( |
| 325 | + module="typing", |
| 326 | + names=[ast.Name("TYPE_CHECKING")], |
| 327 | + ), |
| 328 | + ) |
| 329 | + |
| 330 | + return None |
| 331 | + |
| 332 | + def _update_name_to_constant(self, node: ast.expr) -> ast.expr: |
| 333 | + """Update return types. |
| 334 | +
|
| 335 | + If the return type contains any type that resolves to an `ast.Name`, |
| 336 | + convert it to an `ast.Constant`. We only need the type for type checking |
| 337 | + and can avoid importing the type in the global scope unless needed. |
| 338 | +
|
| 339 | + :param node: The ast node used as return type |
| 340 | + :returns: A modified ast node |
| 341 | + """ |
| 342 | + if isinstance(node, ast.Name): |
| 343 | + if node.id in self.imported_classes: |
| 344 | + self.input_and_return_types.add(node.id) |
| 345 | + return ast.Constant(value=node.id) |
| 346 | + |
| 347 | + if isinstance(node, ast.Subscript): |
| 348 | + node.slice = self._update_name_to_constant(node.slice) |
| 349 | + return node |
| 350 | + |
| 351 | + if isinstance(node, ast.Tuple): |
| 352 | + for i, _ in enumerate(node.elts): |
| 353 | + node.elts[i] = self._update_name_to_constant(node.elts[i]) # type: ignore |
| 354 | + |
| 355 | + return node |
| 356 | + |
| 357 | + return node |
0 commit comments