Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/autoflake.py: 49%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

689 statements  

1#!/usr/bin/env python 

2# Copyright (C) Steven Myint 

3# 

4# Permission is hereby granted, free of charge, to any person obtaining 

5# a copy of this software and associated documentation files (the 

6# "Software"), to deal in the Software without restriction, including 

7# without limitation the rights to use, copy, modify, merge, publish, 

8# distribute, sublicense, and/or sell copies of the Software, and to 

9# permit persons to whom the Software is furnished to do so, subject to 

10# the following conditions: 

11# 

12# The above copyright notice and this permission notice shall be included 

13# in all copies or substantial portions of the Software. 

14# 

15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, 

16# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF 

17# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. 

18# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY 

19# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, 

20# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE 

21# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. 

22"""Removes unused imports and unused variables as reported by pyflakes.""" 

23 

24from __future__ import annotations 

25 

26import ast 

27import collections 

28import difflib 

29import fnmatch 

30import io 

31import logging 

32import os 

33import pathlib 

34import re 

35import signal 

36import string 

37import sys 

38import sysconfig 

39import tokenize 

40from collections.abc import Callable, Iterable, Mapping, MutableMapping, Sequence 

41from typing import IO, Any, cast 

42 

43import pyflakes.api 

44import pyflakes.messages 

45import pyflakes.reporter 

46 

47__version__ = "2.4.0" 

48 

49 

50_LOGGER = logging.getLogger("autoflake") 

51_LOGGER.propagate = False 

52 

53ATOMS = frozenset([tokenize.NAME, tokenize.NUMBER, tokenize.STRING]) 

54 

55EXCEPT_REGEX = re.compile(r"^\s*except [\s,()\w]+ as \w+:$") 

56PYTHON_SHEBANG_REGEX = re.compile(r"^#!.*\bpython[3]?\b\s*$") 

57 

58MAX_PYTHON_FILE_DETECTION_BYTES = 1024 

59 

60IGNORE_COMMENT_REGEX = re.compile( 

61 r"\s*#\s{1,}autoflake:\s{1,}\bskip_file\b", 

62 re.MULTILINE, 

63) 

64 

65 

66def standard_paths() -> Iterable[str]: 

67 """Yield paths to standard modules.""" 

68 paths = sysconfig.get_paths() 

69 path_names = ("stdlib", "platstdlib") 

70 for path_name in path_names: 

71 # Yield lib paths. 

72 if path_name in paths: 

73 path = paths[path_name] 

74 if os.path.isdir(path): 

75 yield from os.listdir(path) 

76 

77 # Yield lib-dynload paths. 

78 dynload_path = os.path.join(path, "lib-dynload") 

79 if os.path.isdir(dynload_path): 

80 yield from os.listdir(dynload_path) 

81 

82 

83def standard_package_names() -> Iterable[str]: 

84 """Yield standard module names.""" 

85 for name in standard_paths(): 

86 if name.startswith("_") or "-" in name: 

87 continue 

88 

89 if "." in name and not name.endswith(("so", "py", "pyc")): 

90 continue 

91 

92 yield name.split(".")[0] 

93 

94 

95IMPORTS_WITH_SIDE_EFFECTS = {"antigravity", "rlcompleter", "this"} 

96 

97# In case they are built into CPython. 

98# 

99# ``standard_package_names()`` only discovers modules that have a file in the 

100# standard library directory, so modules compiled into the interpreter (e.g. 

101# ``itertools``) are invisible to it. ``sys.builtin_module_names`` reports them 

102# for the running interpreter; the literals below cover builds where these are 

103# extension modules on some platforms but not others. 

104BINARY_IMPORTS = { 

105 "datetime", 

106 "grp", 

107 "io", 

108 "json", 

109 "math", 

110 "multiprocessing", 

111 "parser", 

112 "pwd", 

113 "string", 

114 "operator", 

115 "os", 

116 "sys", 

117 "time", 

118} | set(sys.builtin_module_names) 

119 

120SAFE_IMPORTS = ( 

121 frozenset(standard_package_names()) - IMPORTS_WITH_SIDE_EFFECTS | BINARY_IMPORTS 

122) 

123 

124 

125def unused_import_line_numbers( 

126 messages: Iterable[pyflakes.messages.Message], 

127) -> Iterable[int]: 

128 """Yield line numbers of unused imports.""" 

129 for message in messages: 

130 if isinstance(message, pyflakes.messages.UnusedImport): 

131 yield message.lineno 

132 

133 

134def unused_import_module_name( 

135 messages: Iterable[pyflakes.messages.Message], 

136) -> Iterable[tuple[int, str]]: 

137 """Yield line number and module name of unused imports.""" 

138 pattern = re.compile(r"\'(.+?)\'") 

139 for message in messages: 

140 if isinstance(message, pyflakes.messages.UnusedImport): 

141 module_name = pattern.search(str(message)) 

142 if module_name: 

143 module_name = module_name.group()[1:-1] 

144 yield (message.lineno, module_name) 

145 

146 

147def star_import_used_line_numbers( 

148 messages: Iterable[pyflakes.messages.Message], 

149) -> Iterable[int]: 

150 """Yield line number of star import usage.""" 

151 for message in messages: 

152 if isinstance(message, pyflakes.messages.ImportStarUsed): 

153 yield message.lineno 

154 

155 

156def star_import_usage_undefined_name( 

157 messages: Iterable[pyflakes.messages.Message], 

158) -> Iterable[tuple[int, str, str]]: 

159 """Yield line number, undefined name, and its possible origin module.""" 

160 for message in messages: 

161 if isinstance(message, pyflakes.messages.ImportStarUsage): 

162 # pyflakes annotates message_args as an empty tuple 

163 undefined_name = message.message_args[0] # ty: ignore[index-out-of-bounds] 

164 module_name = message.message_args[1] # ty: ignore[index-out-of-bounds] 

165 yield (message.lineno, undefined_name, module_name) 

166 

167 

168def unused_variable_line_numbers( 

169 messages: Iterable[pyflakes.messages.Message], 

170) -> Iterable[int]: 

171 """Yield line numbers of unused variables.""" 

172 for message in messages: 

173 if isinstance(message, pyflakes.messages.UnusedVariable): 

174 yield message.lineno 

175 

176 

177def _split_lines(source: str) -> list[str]: 

178 """Split source into physical lines, keeping line endings. 

179 

180 Unlike ``io.StringIO(source).readlines()``, honor the same line 

181 terminators as the tokenizer (``\r\n``, ``\r``, and ``\n``), so 

182 that enumerating the result stays consistent with the line numbers 

183 that pyflakes reports. A bare ``\r`` counts as a line terminator, 

184 while other characters ``str.splitlines()`` splits on (such as form 

185 feeds) do not. 

186 """ 

187 return re.findall(r"[^\r\n]*(?:\r\n|\r|\n)|[^\r\n]+", source) 

188 

189 

190def duplicate_key_line_numbers( 

191 messages: Iterable[pyflakes.messages.Message], 

192 source: str, 

193) -> Iterable[int]: 

194 """Yield line numbers of duplicate keys.""" 

195 messages = [ 

196 message 

197 for message in messages 

198 if isinstance(message, pyflakes.messages.MultiValueRepeatedKeyLiteral) 

199 ] 

200 

201 if messages: 

202 # Filter out complex cases. We don't want to bother trying to parse 

203 # this stuff and get it right. We can do it on a key-by-key basis. 

204 

205 key_to_messages = create_key_to_messages_dict(messages) 

206 

207 lines = _split_lines(source) 

208 

209 for key, messages in key_to_messages.items(): 

210 good = True 

211 for message in messages: 

212 line = lines[message.lineno - 1] 

213 key = message.message_args[0] # ty: ignore[index-out-of-bounds] 

214 

215 if not dict_entry_has_key(line, key): 

216 good = False 

217 

218 if good: 

219 for message in messages: 

220 yield message.lineno 

221 

222 

223def create_key_to_messages_dict( 

224 messages: Iterable[pyflakes.messages.MultiValueRepeatedKeyLiteral], 

225) -> Mapping[Any, Iterable[pyflakes.messages.MultiValueRepeatedKeyLiteral]]: 

226 """Return dict mapping the key to list of messages.""" 

227 dictionary: dict[ 

228 Any, 

229 list[pyflakes.messages.MultiValueRepeatedKeyLiteral], 

230 ] = collections.defaultdict(list) 

231 for message in messages: 

232 dictionary[message.message_args[0]].append(message) # ty: ignore[index-out-of-bounds] 

233 return dictionary 

234 

235 

236def check(source: str) -> Iterable[pyflakes.messages.Message]: 

237 """Return messages from pyflakes.""" 

238 reporter = ListReporter() 

239 try: 

240 pyflakes.api.check(source, filename="<string>", reporter=reporter) 

241 except (AttributeError, RecursionError, UnicodeDecodeError): 

242 pass 

243 return reporter.messages 

244 

245 

246class StubFile: 

247 """Stub out file for pyflakes.""" 

248 

249 def write(self, *_: Any) -> None: 

250 """Stub out.""" 

251 

252 

253class ListReporter(pyflakes.reporter.Reporter): 

254 """Accumulate messages in messages list.""" 

255 

256 def __init__(self) -> None: 

257 """Initialize. 

258 

259 Ignore errors from Reporter. 

260 """ 

261 ignore = StubFile() 

262 pyflakes.reporter.Reporter.__init__(self, ignore, ignore) 

263 self.messages: list[pyflakes.messages.Message] = [] 

264 

265 def flake(self, message: pyflakes.messages.Message) -> None: 

266 """Accumulate messages.""" 

267 self.messages.append(message) 

268 

269 

270def extract_package_name(line: str) -> str | None: 

271 """Return package name in import statement.""" 

272 assert "\\" not in line 

273 assert "(" not in line 

274 assert ")" not in line 

275 assert ";" not in line 

276 

277 if line.lstrip().startswith(("import", "from")): 

278 parts = line.split() 

279 if len(parts) < 2: 

280 return None 

281 word = parts[1] 

282 else: 

283 # Ignore doctests. 

284 return None 

285 

286 package = word.split(".")[0] 

287 assert " " not in package 

288 

289 return package 

290 

291 

292def multiline_import(line: str, previous_line: str = "") -> bool: 

293 """Return True if import is spans multiples lines.""" 

294 for symbol in "()": 

295 if symbol in line: 

296 return True 

297 

298 return multiline_statement(line, previous_line) 

299 

300 

301def multiline_statement(line: str, previous_line: str = "") -> bool: 

302 """Return True if this is part of a multiline statement.""" 

303 for symbol in "\\:;": 

304 if symbol in line: 

305 return True 

306 

307 sio = io.StringIO(line) 

308 try: 

309 list(tokenize.generate_tokens(sio.readline)) 

310 return previous_line.rstrip().endswith("\\") 

311 except (SyntaxError, tokenize.TokenError): 

312 return True 

313 

314 

315class PendingFix: 

316 """Allows a rewrite operation to span multiple lines. 

317 

318 In the main rewrite loop, every time a helper function returns a 

319 ``PendingFix`` object instead of a string, this object will be called 

320 with the following line. 

321 """ 

322 

323 def __init__(self, line: str) -> None: 

324 """Analyse and store the first line.""" 

325 self.accumulator = collections.deque([line]) 

326 

327 def __call__(self, line: str) -> PendingFix | str: 

328 """Process line considering the accumulator. 

329 

330 Return self to keep processing the following lines or a string 

331 with the final result of all the lines processed at once. 

332 """ 

333 raise NotImplementedError("Abstract method needs to be overwritten") 

334 

335 

336def _valid_char_in_line(char: str, line: str) -> bool: 

337 """Return True if a char appears in the line and is not commented.""" 

338 comment_index = line.find("#") 

339 char_index = line.find(char) 

340 valid_char_in_line = char_index >= 0 and ( 

341 comment_index > char_index or comment_index < 0 

342 ) 

343 return valid_char_in_line 

344 

345 

346def _top_module(module_name: str) -> str: 

347 """Return the name of the top level module in the hierarchy.""" 

348 if module_name[0] == ".": 

349 return "%LOCAL_MODULE%" 

350 return module_name.split(".")[0] 

351 

352 

353def _modules_to_remove( 

354 unused_modules: Iterable[str], 

355 safe_to_remove: Iterable[str] = SAFE_IMPORTS, 

356) -> Iterable[str]: 

357 """Discard unused modules that are not safe to remove from the list.""" 

358 return [x for x in unused_modules if _top_module(x) in safe_to_remove] 

359 

360 

361def _segment_module(segment: str) -> str: 

362 """Extract the module identifier inside the segment. 

363 

364 It might be the case the segment does not have a module (e.g. is composed 

365 just by a parenthesis or line continuation and whitespace). In this 

366 scenario we just keep the segment... These characters are not valid in 

367 identifiers, so they will never be contained in the list of unused modules 

368 anyway. 

369 """ 

370 return segment.strip(string.whitespace + ",\\()") or segment 

371 

372 

373class FilterMultilineImport(PendingFix): 

374 """Remove unused imports from multiline import statements. 

375 

376 This class handles both the cases: "from imports" and "direct imports". 

377 

378 Some limitations exist (e.g. imports with comments, lines joined by ``;``, 

379 etc). In these cases, the statement is left unchanged to avoid problems. 

380 """ 

381 

382 IMPORT_RE = re.compile(r"\bimport\b\s*") 

383 INDENTATION_RE = re.compile(r"^\s*") 

384 BASE_RE = re.compile(r"\bfrom\s+([^ ]+)") 

385 SEGMENT_RE = re.compile( 

386 r"([^,\s]+(?:[\s\\]+as[\s\\]+[^,\s]+)?[,\s\\)]*)", 

387 re.MULTILINE, 

388 ) 

389 # ^ module + comma + following space (including new line and continuation) 

390 IDENTIFIER_RE = re.compile(r"[^,\s]+") 

391 

392 def __init__( 

393 self, 

394 line: str, 

395 unused_module: Iterable[str] = (), 

396 remove_all_unused_imports: bool = False, 

397 safe_to_remove: Iterable[str] = SAFE_IMPORTS, 

398 previous_line: str = "", 

399 ): 

400 """Receive the same parameters as ``filter_unused_import``.""" 

401 self.remove: Iterable[str] = unused_module 

402 self.parenthesized: bool = "(" in line 

403 self.from_, imports = self.IMPORT_RE.split(line, maxsplit=1) 

404 match = self.BASE_RE.search(self.from_) 

405 self.base = match.group(1) if match else None 

406 self.give_up: bool = False 

407 

408 if not remove_all_unused_imports: 

409 if self.base and _top_module(self.base) not in safe_to_remove: 

410 self.give_up = True 

411 else: 

412 self.remove = _modules_to_remove(self.remove, safe_to_remove) 

413 

414 if "\\" in previous_line: 

415 # Ignore tricky things like "try: \<new line> import" ... 

416 self.give_up = True 

417 

418 self.analyze(line) 

419 

420 PendingFix.__init__(self, imports) 

421 

422 def is_over(self, line: str | None = None) -> bool: 

423 """Return True if the multiline import statement is over.""" 

424 line = line or self.accumulator[-1] 

425 

426 if self.parenthesized: 

427 return _valid_char_in_line(")", line) 

428 

429 return not _valid_char_in_line("\\", line) 

430 

431 def analyze(self, line: str) -> None: 

432 """Decide if the statement will be fixed or left unchanged.""" 

433 if any(ch in line for ch in ";:#"): 

434 self.give_up = True 

435 

436 def fix(self, accumulated: Iterable[str]) -> str: 

437 """Given a collection of accumulated lines, fix the entire import.""" 

438 old_imports = "".join(accumulated) 

439 ending = get_line_ending(old_imports) 

440 # Split imports into segments that contain the module name + 

441 # comma + whitespace and eventual <newline> \ ( ) chars 

442 segments = [x for x in self.SEGMENT_RE.findall(old_imports) if x] 

443 modules = [_segment_module(x) for x in segments] 

444 keep = _filter_imports(modules, self.base, self.remove) 

445 

446 # Short-circuit if no import was discarded 

447 if len(keep) == len(segments): 

448 return self.from_ + "import " + "".join(accumulated) 

449 

450 fixed = "" 

451 if keep: 

452 # Since it is very difficult to deal with all the line breaks and 

453 # continuations, let's use the code layout that already exists and 

454 # just replace the module identifiers inside the first N-1 segments 

455 # + the last segment 

456 templates = list(zip(modules, segments)) 

457 templates = templates[: len(keep) - 1] + templates[-1:] 

458 # It is important to keep the last segment, since it might contain 

459 # important chars like `)` 

460 fixed = "".join( 

461 template.replace(module, keep[i]) 

462 for i, (module, template) in enumerate(templates) 

463 ) 

464 

465 # Fix the edge case: inline parenthesis + just one surviving import 

466 if self.parenthesized and any(ch not in fixed for ch in "()"): 

467 fixed = fixed.strip(string.whitespace + "()") + ending 

468 

469 # Replace empty imports with a "pass" statement 

470 empty = len(fixed.strip(string.whitespace + "\\(),")) < 1 

471 if empty: 

472 match = self.INDENTATION_RE.search(self.from_) 

473 assert match is not None 

474 indentation = match.group(0) 

475 return indentation + "pass" + ending 

476 

477 return self.from_ + "import " + fixed 

478 

479 def __call__(self, line: str | None = None) -> PendingFix | str: 

480 """Accumulate all the lines in the import and then trigger the fix.""" 

481 if line: 

482 self.accumulator.append(line) 

483 self.analyze(line) 

484 if not self.is_over(line): 

485 return self 

486 if self.give_up: 

487 return self.from_ + "import " + "".join(self.accumulator) 

488 

489 return self.fix(self.accumulator) 

490 

491 

492def _filter_imports( 

493 imports: Iterable[str], 

494 parent: str | None = None, 

495 unused_module: Iterable[str] = (), 

496) -> Sequence[str]: 

497 # We compare full module name (``a.module`` not `module`) to 

498 # guarantee the exact same module as detected from pyflakes. 

499 sep = "" if parent and parent[-1] == "." else "." 

500 

501 def full_name(name: str) -> str: 

502 return name if parent is None else parent + sep + name 

503 

504 return [x for x in imports if full_name(x) not in unused_module] 

505 

506 

507def filter_from_import(line: str, unused_module: Iterable[str]) -> str: 

508 """Parse and filter ``from something import a, b, c``. 

509 

510 Return line without unused import modules, or `pass` if all of the 

511 module in import is unused. 

512 """ 

513 indentation, imports = re.split( 

514 pattern=r"\bimport\b", 

515 string=line, 

516 maxsplit=1, 

517 ) 

518 match = re.search( 

519 pattern=r"\bfrom\s+([^ ]+)", 

520 string=indentation, 

521 ) 

522 assert match is not None 

523 base_module = match.group(1) 

524 

525 imports = re.split(pattern=r"\s*,\s*", string=imports.strip()) 

526 filtered_imports = _filter_imports(imports, base_module, unused_module) 

527 

528 # All of the import in this statement is unused 

529 if not filtered_imports: 

530 return get_indentation(line) + "pass" + get_line_ending(line) 

531 

532 indentation += "import " 

533 

534 return indentation + ", ".join(filtered_imports) + get_line_ending(line) 

535 

536 

537def break_up_import(line: str) -> str: 

538 """Return line with imports on separate lines.""" 

539 assert "\\" not in line 

540 assert "(" not in line 

541 assert ")" not in line 

542 assert ";" not in line 

543 assert "#" not in line 

544 assert not line.lstrip().startswith("from") 

545 

546 newline = get_line_ending(line) 

547 if not newline: 

548 return line 

549 

550 indentation, imports = re.split( 

551 pattern=r"\bimport\b", 

552 string=line, 

553 maxsplit=1, 

554 ) 

555 

556 indentation += "import " 

557 assert newline 

558 

559 return "".join( 

560 [indentation + i.strip() + newline for i in imports.split(",")], 

561 ) 

562 

563 

564def filter_code( 

565 source: str, 

566 additional_imports: Iterable[str] | None = None, 

567 expand_star_imports: bool = False, 

568 remove_all_unused_imports: bool = False, 

569 remove_duplicate_keys: bool = False, 

570 remove_unused_variables: bool = False, 

571 remove_rhs_for_unused_variables: bool = False, 

572 ignore_init_module_imports: bool = False, 

573) -> Iterable[str]: 

574 """Yield code with unused imports removed.""" 

575 imports = SAFE_IMPORTS 

576 if additional_imports: 

577 imports |= frozenset(additional_imports) 

578 del additional_imports 

579 

580 messages = check(source) 

581 

582 if ignore_init_module_imports: 

583 marked_import_line_numbers: frozenset[int] = frozenset() 

584 else: 

585 marked_import_line_numbers = frozenset( 

586 unused_import_line_numbers(messages), 

587 ) 

588 marked_unused_module: dict[int, list[str]] = collections.defaultdict(list) 

589 for line_number, module_name in unused_import_module_name(messages): 

590 marked_unused_module[line_number].append(module_name) 

591 

592 undefined_names: list[str] = [] 

593 if expand_star_imports and not ( 

594 # See explanations in #18. 

595 re.search(r"\b__all__\b", source) or re.search(r"\bdel\b", source) 

596 ): 

597 marked_star_import_line_numbers = frozenset( 

598 star_import_used_line_numbers(messages), 

599 ) 

600 if len(marked_star_import_line_numbers) > 1: 

601 # Auto expanding only possible for single star import 

602 marked_star_import_line_numbers = frozenset() 

603 else: 

604 for line_number, undefined_name, _ in star_import_usage_undefined_name( 

605 messages, 

606 ): 

607 undefined_names.append(undefined_name) 

608 if not undefined_names: 

609 marked_star_import_line_numbers = frozenset() 

610 else: 

611 marked_star_import_line_numbers = frozenset() 

612 

613 if remove_unused_variables: 

614 marked_variable_line_numbers = frozenset( 

615 unused_variable_line_numbers(messages), 

616 ) 

617 else: 

618 marked_variable_line_numbers = frozenset() 

619 

620 if remove_duplicate_keys: 

621 marked_key_line_numbers: frozenset[int] = frozenset( 

622 duplicate_key_line_numbers(messages, source), 

623 ) 

624 else: 

625 marked_key_line_numbers = frozenset() 

626 

627 line_messages = get_messages_by_line(messages) 

628 

629 previous_line = "" 

630 result: str | PendingFix = "" 

631 for line_number, line in enumerate(_split_lines(source), start=1): 

632 if isinstance(result, PendingFix): 

633 result = result(line) 

634 elif "#" in line: 

635 result = line 

636 elif line_number in marked_import_line_numbers: 

637 result = filter_unused_import( 

638 line, 

639 unused_module=marked_unused_module[line_number], 

640 remove_all_unused_imports=remove_all_unused_imports, 

641 imports=imports, 

642 previous_line=previous_line, 

643 ) 

644 elif line_number in marked_variable_line_numbers: 

645 result = filter_unused_variable( 

646 line, 

647 drop_rhs=remove_rhs_for_unused_variables, 

648 ) 

649 elif line_number in marked_key_line_numbers: 

650 result = filter_duplicate_key( 

651 line, 

652 line_messages[line_number], 

653 line_number, 

654 marked_key_line_numbers, 

655 source, 

656 ) 

657 elif line_number in marked_star_import_line_numbers: 

658 result = filter_star_import(line, undefined_names) 

659 else: 

660 result = line 

661 

662 if not isinstance(result, PendingFix): 

663 yield result 

664 

665 previous_line = line 

666 

667 

668def get_messages_by_line( 

669 messages: Iterable[pyflakes.messages.Message], 

670) -> Mapping[int, pyflakes.messages.Message]: 

671 """Return dictionary that maps line number to message.""" 

672 line_messages: dict[int, pyflakes.messages.Message] = {} 

673 for message in messages: 

674 line_messages[message.lineno] = message 

675 return line_messages 

676 

677 

678def filter_star_import( 

679 line: str, 

680 marked_star_import_undefined_name: Iterable[str], 

681) -> str: 

682 """Return line with the star import expanded.""" 

683 undefined_name = sorted(set(marked_star_import_undefined_name)) 

684 return re.sub(r"\*", ", ".join(undefined_name), line) 

685 

686 

687def filter_unused_import( 

688 line: str, 

689 unused_module: Iterable[str], 

690 remove_all_unused_imports: bool, 

691 imports: Iterable[str], 

692 previous_line: str = "", 

693) -> PendingFix | str: 

694 """Return line if used, otherwise return None.""" 

695 # Ignore doctests. 

696 if line.lstrip().startswith(">"): 

697 return line 

698 

699 if multiline_import(line, previous_line): 

700 if not FilterMultilineImport.IMPORT_RE.search(line): 

701 # Ignore imports with the ``import`` keyword on a continuation 

702 # line (e.g. ``from x \<newline> import y``). 

703 return line 

704 filt = FilterMultilineImport( 

705 line, 

706 unused_module, 

707 remove_all_unused_imports, 

708 imports, 

709 previous_line, 

710 ) 

711 return filt() 

712 

713 is_from_import = line.lstrip().startswith("from") 

714 

715 if "," in line and not is_from_import: 

716 return break_up_import(line) 

717 

718 package = extract_package_name(line) 

719 if not remove_all_unused_imports and package is not None and package not in imports: 

720 return line 

721 

722 if "," in line: 

723 assert is_from_import 

724 return filter_from_import(line, unused_module) 

725 else: 

726 # We need to replace import with "pass" in case the import is the 

727 # only line inside a block. For example, 

728 # "if True:\n import os". In such cases, if the import is 

729 # removed, the block will be left hanging with no body. 

730 return get_indentation(line) + "pass" + get_line_ending(line) 

731 

732 

733def filter_unused_variable( 

734 line: str, 

735 previous_line: str = "", 

736 drop_rhs: bool = False, 

737) -> str: 

738 """Return line if used, otherwise return None.""" 

739 if re.match(EXCEPT_REGEX, line): 

740 return re.sub(r" as \w+:$", ":", line, count=1) 

741 elif multiline_statement(line, previous_line): 

742 return line 

743 elif line.count("=") == 1: 

744 split_line = line.split("=") 

745 assert len(split_line) == 2 

746 value = split_line[1].lstrip() 

747 if "," in split_line[0]: 

748 return line 

749 

750 if is_literal_or_name(value): 

751 # Rather than removing the line, replace with it "pass" to avoid 

752 # a possible hanging block with no body. 

753 value = "pass" + get_line_ending(line) 

754 if drop_rhs: 

755 return get_indentation(line) + value 

756 

757 if drop_rhs: 

758 return "" 

759 return get_indentation(line) + value 

760 else: 

761 return line 

762 

763 

764def filter_duplicate_key( 

765 line: str, 

766 message: pyflakes.messages.Message, 

767 line_number: int, 

768 marked_line_numbers: Iterable[int], 

769 source: str, 

770 previous_line: str = "", 

771) -> str: 

772 """Return '' if first occurrence of the key otherwise return `line`.""" 

773 if marked_line_numbers and line_number == min(marked_line_numbers): 

774 return "" 

775 

776 return line 

777 

778 

779def dict_entry_has_key(line: str, key: Any) -> bool: 

780 """Return True if `line` is a dict entry that uses `key`. 

781 

782 Return False for multiline cases where the line should not be removed by 

783 itself. 

784 

785 """ 

786 if "#" in line: 

787 return False 

788 

789 result = re.match(r"\s*(.*)\s*:\s*(.*),\s*$", line) 

790 if not result: 

791 return False 

792 

793 try: 

794 candidate_key = ast.literal_eval(result.group(1)) 

795 except (SyntaxError, ValueError): 

796 return False 

797 

798 if multiline_statement(result.group(2)): 

799 return False 

800 

801 return cast(bool, candidate_key == key) 

802 

803 

804def is_literal_or_name(value: str) -> bool: 

805 """Return True if value is a literal or a name.""" 

806 try: 

807 ast.literal_eval(value) 

808 return True 

809 except (SyntaxError, TypeError, ValueError): 

810 pass 

811 

812 if value.strip() in ["dict()", "list()", "set()"]: 

813 return True 

814 

815 # Support removal of variables on the right side. But make sure 

816 # there are no dots, which could mean an access of a property. 

817 return re.match(r"^\w+\s*$", value) is not None 

818 

819 

820def useless_pass_line_numbers( 

821 source: str, 

822 ignore_pass_after_docstring: bool = False, 

823) -> Iterable[int]: 

824 """Yield line numbers of unneeded "pass" statements.""" 

825 sio = io.StringIO(source) 

826 previous_token_type = None 

827 last_pass_row = None 

828 last_pass_indentation = None 

829 previous_line = "" 

830 previous_non_empty_line = "" 

831 for token in tokenize.generate_tokens(sio.readline): 

832 token_type = token[0] 

833 start_row = token[2][0] 

834 line = token[4] 

835 

836 is_pass = token_type == tokenize.NAME and line.strip() == "pass" 

837 

838 # Leading "pass". 

839 if ( 

840 start_row - 1 == last_pass_row 

841 and get_indentation(line) == last_pass_indentation 

842 and token_type in ATOMS 

843 and not is_pass 

844 ): 

845 yield start_row - 1 

846 

847 if is_pass: 

848 last_pass_row = start_row 

849 last_pass_indentation = get_indentation(line) 

850 

851 is_trailing_pass = ( 

852 previous_token_type != tokenize.INDENT 

853 and not previous_line.rstrip().endswith("\\") 

854 ) 

855 

856 is_pass_after_docstring = previous_non_empty_line.rstrip().endswith( 

857 ("'''", '"""'), 

858 ) 

859 

860 # Trailing "pass". 

861 if is_trailing_pass: 

862 if is_pass_after_docstring and ignore_pass_after_docstring: 

863 continue 

864 else: 

865 yield start_row 

866 

867 previous_token_type = token_type 

868 previous_line = line 

869 if line.strip(): 

870 previous_non_empty_line = line 

871 

872 

873def filter_useless_pass( 

874 source: str, 

875 ignore_pass_statements: bool = False, 

876 ignore_pass_after_docstring: bool = False, 

877) -> Iterable[str]: 

878 """Yield code with useless "pass" lines removed.""" 

879 if ignore_pass_statements: 

880 marked_lines: frozenset[int] = frozenset() 

881 else: 

882 try: 

883 marked_lines = frozenset( 

884 useless_pass_line_numbers( 

885 source, 

886 ignore_pass_after_docstring, 

887 ), 

888 ) 

889 except (SyntaxError, tokenize.TokenError): 

890 marked_lines = frozenset() 

891 

892 sio = io.StringIO(source) 

893 for line_number, line in enumerate(sio.readlines(), start=1): 

894 if line_number not in marked_lines: 

895 yield line 

896 

897 

898def get_indentation(line: str) -> str: 

899 """Return leading whitespace.""" 

900 if line.strip(): 

901 non_whitespace_index = len(line) - len(line.lstrip()) 

902 return line[:non_whitespace_index] 

903 else: 

904 return "" 

905 

906 

907def get_line_ending(line: str) -> str: 

908 """Return line ending.""" 

909 non_whitespace_index = len(line.rstrip()) - len(line) 

910 if not non_whitespace_index: 

911 return "" 

912 else: 

913 return line[non_whitespace_index:] 

914 

915 

916def fix_code( 

917 source: str, 

918 additional_imports: Iterable[str] | None = None, 

919 expand_star_imports: bool = False, 

920 remove_all_unused_imports: bool = False, 

921 remove_duplicate_keys: bool = False, 

922 remove_unused_variables: bool = False, 

923 remove_rhs_for_unused_variables: bool = False, 

924 ignore_init_module_imports: bool = False, 

925 ignore_pass_statements: bool = False, 

926 ignore_pass_after_docstring: bool = False, 

927) -> str: 

928 """Return code with all filtering run on it.""" 

929 if not source: 

930 return source 

931 

932 if IGNORE_COMMENT_REGEX.search(source): 

933 return source 

934 

935 # pyflakes does not handle "nonlocal" correctly. 

936 if "nonlocal" in source: 

937 remove_unused_variables = False 

938 

939 filtered_source = None 

940 while True: 

941 filtered_source = "".join( 

942 filter_useless_pass( 

943 "".join( 

944 filter_code( 

945 source, 

946 additional_imports=additional_imports, 

947 expand_star_imports=expand_star_imports, 

948 remove_all_unused_imports=remove_all_unused_imports, 

949 remove_duplicate_keys=remove_duplicate_keys, 

950 remove_unused_variables=remove_unused_variables, 

951 remove_rhs_for_unused_variables=( 

952 remove_rhs_for_unused_variables 

953 ), 

954 ignore_init_module_imports=ignore_init_module_imports, 

955 ), 

956 ), 

957 ignore_pass_statements=ignore_pass_statements, 

958 ignore_pass_after_docstring=ignore_pass_after_docstring, 

959 ), 

960 ) 

961 

962 if filtered_source == source: 

963 break 

964 source = filtered_source 

965 

966 return filtered_source 

967 

968 

969def fix_file( 

970 filename: str, 

971 args: Mapping[str, Any], 

972 standard_out: IO[str] | None = None, 

973) -> int: 

974 """Run fix_code() on a file.""" 

975 if standard_out is None: 

976 standard_out = sys.stdout 

977 encoding = detect_encoding(filename) 

978 with open_with_encoding(filename, encoding=encoding) as input_file: 

979 return _fix_file( 

980 input_file, 

981 filename, 

982 args, 

983 args["write_to_stdout"], 

984 cast(IO[str], standard_out), 

985 encoding=encoding, 

986 ) 

987 

988 

989def _fix_file( 

990 input_file: IO[str], 

991 filename: str, 

992 args: Mapping[str, Any], 

993 write_to_stdout: bool, 

994 standard_out: IO[str], 

995 encoding: str | None = None, 

996) -> int: 

997 source = input_file.read() 

998 original_source = source 

999 

1000 isInitFile = os.path.basename(filename) == "__init__.py" 

1001 

1002 if args["ignore_init_module_imports"] and isInitFile: 

1003 ignore_init_module_imports = True 

1004 else: 

1005 ignore_init_module_imports = False 

1006 

1007 filtered_source = fix_code( 

1008 source, 

1009 additional_imports=(args["imports"].split(",") if "imports" in args else None), 

1010 expand_star_imports=args["expand_star_imports"], 

1011 remove_all_unused_imports=args["remove_all_unused_imports"], 

1012 remove_duplicate_keys=args["remove_duplicate_keys"], 

1013 remove_unused_variables=args["remove_unused_variables"], 

1014 remove_rhs_for_unused_variables=(args["remove_rhs_for_unused_variables"]), 

1015 ignore_init_module_imports=ignore_init_module_imports, 

1016 ignore_pass_statements=args["ignore_pass_statements"], 

1017 ignore_pass_after_docstring=args["ignore_pass_after_docstring"], 

1018 ) 

1019 

1020 if original_source != filtered_source: 

1021 if args["check"]: 

1022 standard_out.write( 

1023 f"{filename}: Unused imports/variables detected{os.linesep}", 

1024 ) 

1025 return 1 

1026 if args["check_diff"]: 

1027 diff = get_diff_text( 

1028 io.StringIO(original_source).readlines(), 

1029 io.StringIO(filtered_source).readlines(), 

1030 filename, 

1031 ) 

1032 standard_out.write("".join(diff)) 

1033 return 1 

1034 if write_to_stdout: 

1035 standard_out.write(filtered_source) 

1036 elif args["in_place"]: 

1037 with open_with_encoding( 

1038 filename, 

1039 mode="w", 

1040 encoding=encoding, 

1041 ) as output_file: 

1042 output_file.write(filtered_source) 

1043 _LOGGER.info("Fixed %s", filename) 

1044 else: 

1045 diff = get_diff_text( 

1046 io.StringIO(original_source).readlines(), 

1047 io.StringIO(filtered_source).readlines(), 

1048 filename, 

1049 ) 

1050 standard_out.write("".join(diff)) 

1051 elif write_to_stdout: 

1052 standard_out.write(filtered_source) 

1053 else: 

1054 if (args["check"] or args["check_diff"]) and not args["quiet"]: 

1055 standard_out.write(f"{filename}: No issues detected!{os.linesep}") 

1056 else: 

1057 _LOGGER.debug("Clean %s: nothing to fix", filename) 

1058 

1059 return 0 

1060 

1061 

1062def open_with_encoding( 

1063 filename: str, 

1064 encoding: str | None, 

1065 mode: str = "r", 

1066 limit_byte_check: int = -1, 

1067) -> IO[str]: 

1068 """Return opened file with a specific encoding.""" 

1069 if not encoding: 

1070 encoding = detect_encoding(filename, limit_byte_check=limit_byte_check) 

1071 

1072 return open( 

1073 filename, 

1074 mode=mode, 

1075 encoding=encoding, 

1076 newline="", # Preserve line endings 

1077 ) 

1078 

1079 

1080def detect_encoding(filename: str, limit_byte_check: int = -1) -> str: 

1081 """Return file encoding.""" 

1082 try: 

1083 with open(filename, "rb") as input_file: 

1084 encoding = _detect_encoding(input_file.readline) 

1085 

1086 # Check for correctness of encoding. 

1087 with open_with_encoding(filename, encoding) as input_file: 

1088 input_file.read(limit_byte_check) 

1089 

1090 return encoding 

1091 except (LookupError, SyntaxError, UnicodeDecodeError): 

1092 return "latin-1" 

1093 

1094 

1095def _detect_encoding(readline: Callable[[], bytes]) -> str: 

1096 """Return file encoding.""" 

1097 try: 

1098 encoding = tokenize.detect_encoding(readline)[0] 

1099 return encoding 

1100 except (LookupError, SyntaxError, UnicodeDecodeError): 

1101 return "latin-1" 

1102 

1103 

1104def get_diff_text(old: Sequence[str], new: Sequence[str], filename: str) -> str: 

1105 """Return text of unified diff between old and new.""" 

1106 newline = "\n" 

1107 diff = difflib.unified_diff( 

1108 old, 

1109 new, 

1110 "original/" + filename, 

1111 "fixed/" + filename, 

1112 lineterm=newline, 

1113 ) 

1114 

1115 text = "" 

1116 for line in diff: 

1117 text += line 

1118 

1119 # Work around missing newline (http://bugs.python.org/issue2142). 

1120 if not line.endswith(newline): 

1121 text += newline + r"\ No newline at end of file" + newline 

1122 

1123 return text 

1124 

1125 

1126def _split_comma_separated(string: str) -> set[str]: 

1127 """Return a set of strings.""" 

1128 return {text.strip() for text in string.split(",") if text.strip()} 

1129 

1130 

1131def is_python_file(filename: str) -> bool: 

1132 """Return True if filename is Python file.""" 

1133 if filename.endswith(".py"): 

1134 return True 

1135 

1136 try: 

1137 with open_with_encoding( 

1138 filename, 

1139 None, 

1140 limit_byte_check=MAX_PYTHON_FILE_DETECTION_BYTES, 

1141 ) as f: 

1142 text = f.read(MAX_PYTHON_FILE_DETECTION_BYTES) 

1143 if not text: 

1144 return False 

1145 first_line = text.splitlines()[0] 

1146 except (OSError, IndexError): 

1147 return False 

1148 

1149 return PYTHON_SHEBANG_REGEX.match(first_line) is not None 

1150 

1151 

1152def is_exclude_file(filename: str, exclude: Iterable[str]) -> bool: 

1153 """Return True if file matches exclude pattern.""" 

1154 base_name = os.path.basename(filename) 

1155 

1156 if base_name.startswith("."): 

1157 return True 

1158 

1159 for pattern in exclude: 

1160 if fnmatch.fnmatch(base_name, pattern): 

1161 return True 

1162 if fnmatch.fnmatch(filename, pattern): 

1163 return True 

1164 return False 

1165 

1166 

1167def match_file(filename: str, exclude: Iterable[str]) -> bool: 

1168 """Return True if file is okay for modifying/recursing.""" 

1169 if is_exclude_file(filename, exclude): 

1170 _LOGGER.debug("Skipped %s: matched to exclude pattern", filename) 

1171 return False 

1172 

1173 return os.path.isdir(filename) or is_python_file(filename) 

1174 

1175 

1176def find_files( 

1177 filenames: list[str], 

1178 recursive: bool, 

1179 exclude: Iterable[str], 

1180) -> Iterable[str]: 

1181 """Yield filenames.""" 

1182 while filenames: 

1183 name = filenames.pop(0) 

1184 if recursive and os.path.isdir(name): 

1185 for root, directories, children in os.walk(name): 

1186 filenames += [ 

1187 os.path.join(root, f) 

1188 for f in children 

1189 if match_file( 

1190 os.path.join(root, f), 

1191 exclude, 

1192 ) 

1193 ] 

1194 directories[:] = [ 

1195 d 

1196 for d in directories 

1197 if match_file( 

1198 os.path.join(root, d), 

1199 exclude, 

1200 ) 

1201 ] 

1202 else: 

1203 if not is_exclude_file(name, exclude): 

1204 yield name 

1205 else: 

1206 _LOGGER.debug("Skipped %s: matched to exclude pattern", name) 

1207 

1208 

1209def process_pyproject_toml(toml_file_path: str) -> MutableMapping[str, Any] | None: 

1210 """Extract config mapping from pyproject.toml file.""" 

1211 if sys.version_info >= (3, 11): 

1212 import tomllib 

1213 else: 

1214 import tomli as tomllib 

1215 

1216 with open(toml_file_path, "rb") as f: 

1217 return tomllib.load(f).get("tool", {}).get("autoflake", None) 

1218 

1219 

1220def process_config_file(config_file_path: str) -> MutableMapping[str, Any] | None: 

1221 """Extract config mapping from config file.""" 

1222 import configparser 

1223 

1224 reader = configparser.ConfigParser() 

1225 reader.read(config_file_path, encoding="utf-8") 

1226 if not reader.has_section("autoflake"): 

1227 return None 

1228 

1229 return reader["autoflake"] 

1230 

1231 

1232def find_and_process_config(args: Mapping[str, Any]) -> MutableMapping[str, Any] | None: 

1233 # Configuration file parsers {filename: parser function}. 

1234 CONFIG_FILES: Mapping[str, Callable[[str], MutableMapping[str, Any] | None]] = { 

1235 "pyproject.toml": process_pyproject_toml, 

1236 "setup.cfg": process_config_file, 

1237 } 

1238 # Traverse the file tree common to all files given as argument looking for 

1239 # a configuration file 

1240 config_path = os.path.commonpath([os.path.abspath(file) for file in args["files"]]) 

1241 config: Mapping[str, Any] | None = None 

1242 while True: 

1243 for config_file, processor in CONFIG_FILES.items(): 

1244 config_file_path = os.path.join( 

1245 os.path.join(config_path, config_file), 

1246 ) 

1247 if os.path.isfile(config_file_path): 

1248 config = processor(config_file_path) 

1249 if config is not None: 

1250 break 

1251 if config is not None: 

1252 break 

1253 config_path, tail = os.path.split(config_path) 

1254 if not tail: 

1255 break 

1256 return config 

1257 

1258 

1259def merge_configuration_file( 

1260 flag_args: MutableMapping[str, Any], 

1261) -> tuple[MutableMapping[str, Any], bool]: 

1262 """Merge configuration from a file into args.""" 

1263 BOOL_TYPES = { 

1264 "1": True, 

1265 "yes": True, 

1266 "true": True, 

1267 "on": True, 

1268 "0": False, 

1269 "no": False, 

1270 "false": False, 

1271 "off": False, 

1272 } 

1273 

1274 if "config_file" in flag_args: 

1275 config_file = pathlib.Path(flag_args["config_file"]).resolve() 

1276 process_method = process_config_file 

1277 if config_file.suffix == ".toml": 

1278 process_method = process_pyproject_toml 

1279 

1280 config = process_method(str(config_file)) 

1281 

1282 if not config: 

1283 _LOGGER.error( 

1284 "can't parse config file '%s'", 

1285 config_file, 

1286 ) 

1287 return flag_args, False 

1288 else: 

1289 config = find_and_process_config(flag_args) 

1290 

1291 BOOL_FLAGS = { 

1292 "check", 

1293 "check_diff", 

1294 "expand_star_imports", 

1295 "ignore_init_module_imports", 

1296 "ignore_pass_after_docstring", 

1297 "ignore_pass_statements", 

1298 "in_place", 

1299 "quiet", 

1300 "recursive", 

1301 "remove_all_unused_imports", 

1302 "remove_duplicate_keys", 

1303 "remove_rhs_for_unused_variables", 

1304 "remove_unused_variables", 

1305 "write_to_stdout", 

1306 } 

1307 

1308 config_args: dict[str, Any] = {} 

1309 if config is not None: 

1310 for name, value in config.items(): 

1311 arg = name.replace("-", "_") 

1312 if arg in BOOL_FLAGS: 

1313 # boolean properties 

1314 if isinstance(value, str): 

1315 value = BOOL_TYPES.get(value.lower(), value) 

1316 if not isinstance(value, bool): 

1317 _LOGGER.error( 

1318 "'%s' in the config file should be a boolean", 

1319 name, 

1320 ) 

1321 return flag_args, False 

1322 config_args[arg] = value 

1323 else: 

1324 if isinstance(value, list) and all( 

1325 isinstance(val, str) for val in value 

1326 ): 

1327 value = ",".join(str(val) for val in value) 

1328 if not isinstance(value, str): 

1329 _LOGGER.error( 

1330 "'%s' in the config file should be a comma separated" 

1331 " string or list of strings", 

1332 name, 

1333 ) 

1334 return flag_args, False 

1335 

1336 config_args[arg] = value 

1337 

1338 # merge args that can be merged 

1339 merged_args = {} 

1340 mergeable_keys = {"imports", "exclude"} 

1341 for key in mergeable_keys: 

1342 values = ( 

1343 v for v in (config_args.get(key), flag_args.get(key)) if v is not None 

1344 ) 

1345 value = ",".join(values) 

1346 if value != "": 

1347 merged_args[key] = value 

1348 

1349 default_args = {arg: False for arg in BOOL_FLAGS} 

1350 return { 

1351 **default_args, 

1352 **config_args, 

1353 **flag_args, 

1354 **merged_args, 

1355 }, True 

1356 

1357 

1358def _main( 

1359 argv: Sequence[str], 

1360 standard_out: IO[str] | None, 

1361 standard_error: IO[str] | None, 

1362 standard_input: IO[str] | None = None, 

1363) -> int: 

1364 """Return exit status. 

1365 

1366 0 means no error. 

1367 """ 

1368 import argparse 

1369 

1370 parser = argparse.ArgumentParser( 

1371 description=__doc__, 

1372 prog="autoflake", 

1373 argument_default=argparse.SUPPRESS, 

1374 ) 

1375 check_group = parser.add_mutually_exclusive_group() 

1376 check_group.add_argument( 

1377 "-c", 

1378 "--check", 

1379 action="store_true", 

1380 help="return error code if changes are needed", 

1381 ) 

1382 check_group.add_argument( 

1383 "-cd", 

1384 "--check-diff", 

1385 action="store_true", 

1386 help="return error code if changes are needed, also display file diffs", 

1387 ) 

1388 

1389 imports_group = parser.add_mutually_exclusive_group() 

1390 imports_group.add_argument( 

1391 "--imports", 

1392 help="by default, only unused standard library " 

1393 "imports are removed; specify a comma-separated " 

1394 "list of additional modules/packages", 

1395 ) 

1396 imports_group.add_argument( 

1397 "--remove-all-unused-imports", 

1398 action="store_true", 

1399 help="remove all unused imports (not just those from the standard library)", 

1400 ) 

1401 

1402 parser.add_argument( 

1403 "-r", 

1404 "--recursive", 

1405 action="store_true", 

1406 help="drill down directories recursively", 

1407 ) 

1408 parser.add_argument( 

1409 "-j", 

1410 "--jobs", 

1411 type=int, 

1412 metavar="n", 

1413 default=0, 

1414 help="number of parallel jobs; match CPU count if value is 0 (default: 0)", 

1415 ) 

1416 parser.add_argument( 

1417 "--exclude", 

1418 metavar="globs", 

1419 help="exclude file/directory names that match these comma-separated globs", 

1420 ) 

1421 parser.add_argument( 

1422 "--expand-star-imports", 

1423 action="store_true", 

1424 help="expand wildcard star imports with undefined " 

1425 "names; this only triggers if there is only " 

1426 "one star import in the file; this is skipped if " 

1427 "there are any uses of `__all__` or `del` in the " 

1428 "file", 

1429 ) 

1430 parser.add_argument( 

1431 "--ignore-init-module-imports", 

1432 action="store_true", 

1433 help="exclude __init__.py when removing unused imports", 

1434 ) 

1435 parser.add_argument( 

1436 "--remove-duplicate-keys", 

1437 action="store_true", 

1438 help="remove all duplicate keys in objects", 

1439 ) 

1440 parser.add_argument( 

1441 "--remove-unused-variables", 

1442 action="store_true", 

1443 help="remove unused variables", 

1444 ) 

1445 parser.add_argument( 

1446 "--remove-rhs-for-unused-variables", 

1447 action="store_true", 

1448 help="remove RHS of statements when removing unused variables (unsafe)", 

1449 ) 

1450 parser.add_argument( 

1451 "--ignore-pass-statements", 

1452 action="store_true", 

1453 help="ignore all pass statements", 

1454 ) 

1455 parser.add_argument( 

1456 "--ignore-pass-after-docstring", 

1457 action="store_true", 

1458 help='ignore pass statements after a newline ending on \'"""\'', 

1459 ) 

1460 parser.add_argument( 

1461 "--version", 

1462 action="version", 

1463 version="%(prog)s " + __version__, 

1464 ) 

1465 parser.add_argument( 

1466 "--quiet", 

1467 action="store_true", 

1468 help="Suppress output if there are no issues", 

1469 ) 

1470 parser.add_argument( 

1471 "-v", 

1472 "--verbose", 

1473 action="count", 

1474 dest="verbosity", 

1475 default=0, 

1476 help="print more verbose logs (you can repeat `-v` to make it more verbose)", 

1477 ) 

1478 parser.add_argument( 

1479 "--stdin-display-name", 

1480 dest="stdin_display_name", 

1481 default="stdin", 

1482 help="the name used when processing input from stdin", 

1483 ) 

1484 

1485 parser.add_argument( 

1486 "--config", 

1487 dest="config_file", 

1488 help=( 

1489 "Explicitly set the config file " 

1490 "instead of auto determining based on file location" 

1491 ), 

1492 ) 

1493 

1494 parser.add_argument("files", nargs="+", help="files to format") 

1495 

1496 output_group = parser.add_mutually_exclusive_group() 

1497 output_group.add_argument( 

1498 "-i", 

1499 "--in-place", 

1500 action="store_true", 

1501 help="make changes to files instead of printing diffs", 

1502 ) 

1503 output_group.add_argument( 

1504 "-s", 

1505 "--stdout", 

1506 action="store_true", 

1507 dest="write_to_stdout", 

1508 help=( 

1509 "print changed text to stdout. defaults to true " 

1510 "when formatting stdin, or to false otherwise" 

1511 ), 

1512 ) 

1513 

1514 args: MutableMapping[str, Any] = vars(parser.parse_args(argv[1:])) 

1515 

1516 if standard_error is None: 

1517 _LOGGER.addHandler(logging.NullHandler()) 

1518 else: 

1519 _LOGGER.addHandler(logging.StreamHandler(standard_error)) 

1520 loglevels = [logging.WARNING, logging.INFO, logging.DEBUG] 

1521 try: 

1522 loglevel = loglevels[args["verbosity"]] 

1523 except IndexError: # Too much -v 

1524 loglevel = loglevels[-1] 

1525 _LOGGER.setLevel(loglevel) 

1526 

1527 args, success = merge_configuration_file(args) 

1528 if not success: 

1529 return 1 

1530 

1531 if ( 

1532 args["remove_rhs_for_unused_variables"] 

1533 and not (args["remove_unused_variables"]) 

1534 ): 

1535 _LOGGER.error( 

1536 "Using --remove-rhs-for-unused-variables only makes sense when " 

1537 "used with --remove-unused-variables", 

1538 ) 

1539 return 1 

1540 

1541 if "exclude" in args: 

1542 args["exclude"] = _split_comma_separated(args["exclude"]) 

1543 else: 

1544 args["exclude"] = set() 

1545 

1546 if args["jobs"] < 1: 

1547 worker_count = os.cpu_count() 

1548 if sys.platform == "win32": 

1549 # Work around https://bugs.python.org/issue26903 

1550 worker_count = min(worker_count, 60) 

1551 args["jobs"] = worker_count or 1 

1552 

1553 filenames = list(set(args["files"])) 

1554 

1555 # convert argparse namespace to a dict so that it can be serialized 

1556 # by multiprocessing 

1557 exit_status = 0 

1558 files = list(find_files(filenames, args["recursive"], args["exclude"])) 

1559 if ( 

1560 args["jobs"] == 1 

1561 or len(files) == 1 

1562 or args["jobs"] == 1 

1563 or "-" in files 

1564 or standard_out is not None 

1565 ): 

1566 for name in files: 

1567 if name == "-" and standard_input is not None: 

1568 exit_status |= _fix_file( 

1569 standard_input, 

1570 args["stdin_display_name"], 

1571 args=args, 

1572 write_to_stdout=True, 

1573 standard_out=standard_out or sys.stdout, 

1574 ) 

1575 else: 

1576 try: 

1577 exit_status |= fix_file( 

1578 name, 

1579 args=args, 

1580 standard_out=standard_out, 

1581 ) 

1582 except OSError as exception: 

1583 _LOGGER.error(str(exception)) 

1584 exit_status |= 1 

1585 else: 

1586 import multiprocessing 

1587 

1588 with multiprocessing.Pool(args["jobs"]) as pool: 

1589 futs = [] 

1590 for name in files: 

1591 fut = pool.apply_async(fix_file, args=(name, args)) 

1592 futs.append(fut) 

1593 for fut in futs: 

1594 try: 

1595 exit_status |= fut.get() 

1596 except OSError as exception: 

1597 _LOGGER.error(str(exception)) 

1598 exit_status |= 1 

1599 

1600 return exit_status 

1601 

1602 

1603def main() -> int: 

1604 """Command-line entry point.""" 

1605 try: 

1606 # Exit on broken pipe. 

1607 signal.signal(signal.SIGPIPE, signal.SIG_DFL) 

1608 except AttributeError: # pragma: no cover 

1609 # SIGPIPE is not available on Windows. 

1610 pass 

1611 

1612 try: 

1613 return _main( 

1614 sys.argv, 

1615 standard_out=None, 

1616 standard_error=sys.stderr, 

1617 standard_input=sys.stdin, 

1618 ) 

1619 except KeyboardInterrupt: # pragma: no cover 

1620 return 2 # pragma: no cover 

1621 

1622 

1623if __name__ == "__main__": 

1624 sys.exit(main())