diff --git a/isort/_vendored/tomli/_parser.py b/isort/_vendored/tomli/_parser.py index ab36adc02..cddcb67df 100644 --- a/isort/_vendored/tomli/_parser.py +++ b/isort/_vendored/tomli/_parser.py @@ -1,21 +1,30 @@ import string import warnings +import sys from types import MappingProxyType from typing import IO, Any, Callable, Dict, FrozenSet, Iterable, NamedTuple, Optional, Tuple -from ._re import ( - RE_DATETIME, - RE_LOCALTIME, - RE_NUMBER, - match_to_datetime, - match_to_localtime, - match_to_number, -) +try: + from ._re import ( + RE_DATETIME, + RE_LOCALTIME, + RE_NUMBER, + match_to_datetime, + match_to_localtime, + match_to_number, + ) +except ImportError: + from _re import ( + RE_DATETIME, + RE_LOCALTIME, + RE_NUMBER, + match_to_datetime, + match_to_localtime, + match_to_number, + ) ASCII_CTRL = frozenset(chr(i) for i in range(32)) | frozenset(chr(127)) -# Neither of these sets include quotation mark or backslash. They are -# currently handled as separate cases in the parser functions. ILLEGAL_BASIC_STR_CHARS = ASCII_CTRL - frozenset("\t") ILLEGAL_MULTILINE_BASIC_STR_CHARS = ASCII_CTRL - frozenset("\t\n\r") @@ -32,17 +41,16 @@ BASIC_STR_ESCAPE_REPLACEMENTS = MappingProxyType( { - "\\b": "\u0008", # backspace - "\\t": "\u0009", # tab - "\\n": "\u000a", # linefeed - "\\f": "\u000c", # form feed - "\\r": "\u000d", # carriage return - '\\"': "\u0022", # quote - "\\\\": "\u005c", # backslash + "\\b": "\u0008", + "\\t": "\u0009", + "\\n": "\u000a", + "\\f": "\u000c", + "\\r": "\u000d", + '\\"': "\u0022", + "\\\\": "\u005c", } ) -# Type annotations ParseFloat = Callable[[str], Any] Key = Tuple[str, ...] Pos = int @@ -53,7 +61,6 @@ class TOMLDecodeError(ValueError): def load(fp: IO, *, parse_float: ParseFloat = float) -> Dict[str, Any]: - """Parse TOML from a file object.""" s = fp.read() if isinstance(s, bytes): s = s.decode() @@ -67,33 +74,16 @@ def load(fp: IO, *, parse_float: ParseFloat = float) -> Dict[str, Any]: def loads(s: str, *, parse_float: ParseFloat = float) -> Dict[str, Any]: # noqa: C901 - """Parse TOML from a string.""" - - # The spec allows converting "\r\n" to "\n", even in string - # literals. Let's do so to simplify parsing. src = s.replace("\r\n", "\n") pos = 0 out = Output(NestedDict(), Flags()) header: Key = () - # Parse one statement at a time - # (typically means one line in TOML source) while True: - # 1. Skip line leading whitespace pos = skip_chars(src, pos, TOML_WS) - - # 2. Parse rules. Expect one of the following: - # - end of file - # - end of line - # - comment - # - key/value pair - # - append dict to list (and move to its namespace) - # - create dict (and move to its namespace) - # Skip trailing whitespace when applicable. - try: - char = src[pos] - except IndexError: + if pos >= len(src): break + char = src[pos] if char == "\n": pos += 1 continue @@ -101,10 +91,7 @@ def loads(s: str, *, parse_float: ParseFloat = float) -> Dict[str, Any]: # noqa pos = key_value_rule(src, pos, out, header, parse_float) pos = skip_chars(src, pos, TOML_WS) elif char == "[": - try: - second_char: Optional[str] = src[pos + 1] - except IndexError: - second_char = None + second_char = src[pos + 1] if pos + 1 < len(src) else None if second_char == "[": pos, header = create_list_rule(src, pos, out) else: @@ -113,14 +100,11 @@ def loads(s: str, *, parse_float: ParseFloat = float) -> Dict[str, Any]: # noqa elif char != "#": raise suffixed_err(src, pos, "Invalid statement") - # 3. Skip comment pos = skip_comment(src, pos) - # 4. Expect end of line or end of file - try: - char = src[pos] - except IndexError: + if pos >= len(src): break + char = src[pos] if char != "\n": raise suffixed_err(src, pos, "Expected newline or end of document after a statement") pos += 1 @@ -129,12 +113,7 @@ def loads(s: str, *, parse_float: ParseFloat = float) -> Dict[str, Any]: # noqa class Flags: - """Flags that map to parsed keys/namespaces.""" - - # Marks an immutable namespace (inline array or inline table). FROZEN = 0 - # Marks a nest that has been explicitly created and can no longer - # be opened using the "[table]" syntax. EXPLICIT_NEST = 1 def __init__(self) -> None: @@ -174,7 +153,7 @@ def set(self, key: Key, flag: int, *, recursive: bool) -> None: # noqa: A003 def is_(self, key: Key, flag: int) -> bool: if not key: - return False # document root has no flags + return False cont = self._flags for k in key[:-1]: if k not in cont: @@ -192,7 +171,6 @@ def is_(self, key: Key, flag: int) -> bool: class NestedDict: def __init__(self) -> None: - # The parsed content of the TOML document self.dict: Dict[str, Any] = {} def get_or_create_nest( @@ -230,11 +208,9 @@ class Output(NamedTuple): def skip_chars(src: str, pos: Pos, chars: Iterable[str]) -> Pos: - try: - while src[pos] in chars: - pos += 1 - except IndexError: - pass + length = len(src) + while pos < length and src[pos] in chars: + pos += 1 return pos @@ -252,7 +228,6 @@ def skip_until( new_pos = len(src) if error_on_eof: raise suffixed_err(src, new_pos, f'Expected "{expect!r}"') - if not error_on.isdisjoint(src[pos:new_pos]): while src[pos] not in error_on: pos += 1 @@ -261,11 +236,7 @@ def skip_until( def skip_comment(src: str, pos: Pos) -> Pos: - try: - char: Optional[str] = src[pos] - except IndexError: - char = None - if char == "#": + if pos < len(src) and src[pos] == "#": return skip_until(src, pos + 1, "\n", error_on=ILLEGAL_COMMENT_CHARS, error_on_eof=False) return pos @@ -280,7 +251,7 @@ def skip_comments_and_array_ws(src: str, pos: Pos) -> Pos: def create_dict_rule(src: str, pos: Pos, out: Output) -> Tuple[Pos, Key]: - pos += 1 # Skip "[" + pos += 1 pos = skip_chars(src, pos, TOML_WS) pos, key = parse_key(src, pos) @@ -298,15 +269,13 @@ def create_dict_rule(src: str, pos: Pos, out: Output) -> Tuple[Pos, Key]: def create_list_rule(src: str, pos: Pos, out: Output) -> Tuple[Pos, Key]: - pos += 2 # Skip "[[" + pos += 2 pos = skip_chars(src, pos, TOML_WS) pos, key = parse_key(src, pos) if out.flags.is_(key, Flags.FROZEN): raise suffixed_err(src, pos, f"Can not mutate immutable namespace {key}") - # Free the namespace now that it points to another empty list item... out.flags.unset_all(key) - # ...but this key precisely is still prohibited from table declaration out.flags.set(key, Flags.EXPLICIT_NEST, recursive=False) try: out.data.append_nest_to_list(key) @@ -325,7 +294,6 @@ def key_value_rule(src: str, pos: Pos, out: Output, header: Key, parse_float: Pa if out.flags.is_(abs_key_parent, Flags.FROZEN): raise suffixed_err(src, pos, f"Can not mutate immutable namespace {abs_key_parent}") - # Containers in the relative path can't be opened with the table syntax after this out.flags.set_for_relative_key(header, key, Flags.EXPLICIT_NEST) try: nest = out.data.get_or_create_nest(abs_key_parent) @@ -333,7 +301,6 @@ def key_value_rule(src: str, pos: Pos, out: Output, header: Key, parse_float: Pa raise suffixed_err(src, pos, "Can not overwrite a value") if key_stem in nest: raise suffixed_err(src, pos, "Can not overwrite a value") - # Mark inline table and array namespaces recursively immutable if isinstance(value, (dict, list)): out.flags.set(header + key, Flags.FROZEN, recursive=True) nest[key_stem] = value @@ -342,11 +309,7 @@ def key_value_rule(src: str, pos: Pos, out: Output, header: Key, parse_float: Pa def parse_key_value_pair(src: str, pos: Pos, parse_float: ParseFloat) -> Tuple[Pos, Key, Any]: pos, key = parse_key(src, pos) - try: - char: Optional[str] = src[pos] - except IndexError: - char = None - if char != "=": + if pos >= len(src) or src[pos] != "=": raise suffixed_err(src, pos, 'Expected "=" after a key in a key/value pair') pos += 1 pos = skip_chars(src, pos, TOML_WS) @@ -358,12 +321,9 @@ def parse_key(src: str, pos: Pos) -> Tuple[Pos, Key]: pos, key_part = parse_key_part(src, pos) key: Key = (key_part,) pos = skip_chars(src, pos, TOML_WS) + length = len(src) while True: - try: - char: Optional[str] = src[pos] - except IndexError: - char = None - if char != ".": + if pos >= length or src[pos] != ".": return pos, key pos += 1 pos = skip_chars(src, pos, TOML_WS) @@ -373,10 +333,7 @@ def parse_key(src: str, pos: Pos) -> Tuple[Pos, Key]: def parse_key_part(src: str, pos: Pos) -> Tuple[Pos, str]: - try: - char: Optional[str] = src[pos] - except IndexError: - char = None + char = src[pos] if pos < len(src) else None if char in BARE_KEY_CHARS: start_pos = pos pos = skip_chars(src, pos, BARE_KEY_CHARS) @@ -405,7 +362,9 @@ def parse_array(src: str, pos: Pos, parse_float: ParseFloat) -> Tuple[Pos, list] array.append(val) pos = skip_comments_and_array_ws(src, pos) - c = src[pos : pos + 1] + if pos >= len(src): + raise suffixed_err(src, pos, "Unclosed array") + c = src[pos] if c == "]": return pos + 1, array if c != ",": @@ -438,7 +397,9 @@ def parse_inline_table(src: str, pos: Pos, parse_float: ParseFloat) -> Tuple[Pos raise suffixed_err(src, pos, f'Duplicate inline table key "{key_stem}"') nest[key_stem] = value pos = skip_chars(src, pos, TOML_WS) - c = src[pos : pos + 1] + if pos >= len(src): + raise suffixed_err(src, pos, "Unclosed inline table") + c = src[pos] if c == "}": return pos + 1, nested_dict.dict if c != ",": @@ -455,8 +416,6 @@ def parse_basic_str_escape( # noqa: C901 escape_id = src[pos : pos + 2] pos += 2 if multiline and escape_id in {"\\ ", "\\\t", "\\\n"}: - # Skip whitespace until next non-whitespace character or end of - # the doc. Error if non-whitespace is found before newline. if escape_id != "\\\n": pos = skip_chars(src, pos, TOML_WS) try: @@ -496,10 +455,10 @@ def parse_hex_char(src: str, pos: Pos, hex_len: int) -> Tuple[Pos, str]: def parse_literal_str(src: str, pos: Pos) -> Tuple[Pos, str]: - pos += 1 # Skip starting apostrophe + pos += 1 start_pos = pos pos = skip_until(src, pos, "'", error_on=ILLEGAL_LITERAL_STR_CHARS, error_on_eof=True) - return pos + 1, src[start_pos:pos] # Skip ending apostrophe + return pos + 1, src[start_pos:pos] def parse_multiline_str(src: str, pos: Pos, *, literal: bool) -> Tuple[Pos, str]: @@ -522,8 +481,6 @@ def parse_multiline_str(src: str, pos: Pos, *, literal: bool) -> Tuple[Pos, str] delim = '"' pos, result = parse_basic_str(src, pos, multiline=True) - # Add at maximum two extra apostrophes/quotes if the end sequence - # is 4 or 5 chars long instead of just 3. if not src.startswith(delim, pos): return pos, result pos += 1 @@ -540,58 +497,50 @@ def parse_basic_str(src: str, pos: Pos, *, multiline: bool) -> Tuple[Pos, str]: else: error_on = ILLEGAL_BASIC_STR_CHARS parse_escapes = parse_basic_str_escape - result = "" - start_pos = pos - while True: - try: - char = src[pos] - except IndexError: - raise suffixed_err(src, pos, "Unterminated string") + parts: list[str] = [] + start = pos + length = len(src) + while pos < length: + char = src[pos] if char == '"': if not multiline: - return pos + 1, result + src[start_pos:pos] + parts.append(src[start:pos]) + return pos + 1, "".join(parts) if src.startswith('"""', pos): - return pos + 3, result + src[start_pos:pos] + parts.append(src[start:pos]) + return pos + 3, "".join(parts) pos += 1 continue if char == "\\": - result += src[start_pos:pos] - pos, parsed_escape = parse_escapes(src, pos) - result += parsed_escape - start_pos = pos + parts.append(src[start:pos]) + pos, esc = parse_escapes(src, pos) + parts.append(esc) + start = pos continue if char in error_on: raise suffixed_err(src, pos, f'Illegal character "{char!r}"') pos += 1 + raise suffixed_err(src, pos, "Unterminated string") def parse_value(src: str, pos: Pos, parse_float: ParseFloat) -> Tuple[Pos, Any]: # noqa: C901 - try: - char: Optional[str] = src[pos] - except IndexError: - char = None + char = src[pos] if pos < len(src) else None - # Basic strings if char == '"': if src.startswith('"""', pos): return parse_multiline_str(src, pos, literal=False) return parse_one_line_basic_str(src, pos) - # Literal strings if char == "'": if src.startswith("'''", pos): return parse_multiline_str(src, pos, literal=True) return parse_literal_str(src, pos) - # Booleans - if char == "t": - if src.startswith("true", pos): - return pos + 4, True - if char == "f": - if src.startswith("false", pos): - return pos + 5, False + if char == "t" and src.startswith("true", pos): + return pos + 4, True + if char == "f" and src.startswith("false", pos): + return pos + 5, False - # Dates and times datetime_match = RE_DATETIME.match(src, pos) if datetime_match: try: @@ -603,22 +552,16 @@ def parse_value(src: str, pos: Pos, parse_float: ParseFloat) -> Tuple[Pos, Any]: if localtime_match: return localtime_match.end(), match_to_localtime(localtime_match) - # Integers and "normal" floats. - # The regex will greedily match any type starting with a decimal - # char, so needs to be located after handling of dates and times. number_match = RE_NUMBER.match(src, pos) if number_match: return number_match.end(), match_to_number(number_match, parse_float) - # Arrays if char == "[": return parse_array(src, pos, parse_float) - # Inline tables if char == "{": return parse_inline_table(src, pos, parse_float) - # Special floats first_three = src[pos : pos + 3] if first_three in {"inf", "nan"}: return pos + 3, parse_float(first_three) @@ -630,9 +573,6 @@ def parse_value(src: str, pos: Pos, parse_float: ParseFloat) -> Tuple[Pos, Any]: def suffixed_err(src: str, pos: Pos, msg: str) -> TOMLDecodeError: - """Return a `TOMLDecodeError` where error message is suffixed with - coordinates in source.""" - def coord_repr(src: str, pos: Pos) -> str: if pos >= len(src): return "end of document" @@ -647,4 +587,4 @@ def coord_repr(src: str, pos: Pos) -> str: def is_unicode_scalar_value(codepoint: int) -> bool: - return (0 <= codepoint <= 55295) or (57344 <= codepoint <= 1114111) + return (0 <= codepoint <= 55295) or (57344 <= codepoint <= 1114111) \ No newline at end of file diff --git a/isort/main.py b/isort/main.py index 9369ddd16..049d4c1ff 100644 --- a/isort/main.py +++ b/isort/main.py @@ -114,7 +114,7 @@ def _build_arg_parser() -> argparse.ArgumentParser: " " "If you've used isort 4 but are new to isort 5, see the upgrading guide: " "https://isort.readthedocs.io/en/latest/upgrade_guides/5.0.0.html", - add_help=False, # prevent help option from appearing in "optional arguments" group + add_help=False, ) general_group = parser.add_argument_group("general options") @@ -864,7 +864,7 @@ def parse_args(argv: Sequence[str] | None = None) -> dict[str, Any]: argv = sys.argv[1:] if argv is None else list(argv) parser = _build_arg_parser() - arguments = {key: value for key, value in vars(parser.parse_args(argv)).items() if value} + arguments = {k: v for k, v in vars(parser.parse_args(argv)).items() if v} if "dont_order_by_type" in arguments: arguments["order_by_type"] = False del arguments["dont_order_by_type"] @@ -877,7 +877,7 @@ def parse_args(argv: Sequence[str] | None = None) -> dict[str, Any]: sys.exit("Can't set both --float-to-top and --dont-float-to-top.") else: arguments["float_to_top"] = False - multi_line_output = arguments.get("multi_line_output", None) + multi_line_output = arguments.get("multi_line_output") if multi_line_output: if multi_line_output.isdigit(): arguments["multi_line_output"] = WrapModes(int(multi_line_output)) @@ -888,7 +888,6 @@ def parse_args(argv: Sequence[str] | None = None) -> dict[str, Any]: def _preconvert(item: Any) -> str | list[Any]: - """Preconverts objects from native types into JSONifyiable types""" if isinstance(item, (set, frozenset)): return list(item) if isinstance(item, WrapModes): @@ -986,9 +985,6 @@ def identify_imports_main( print(str(identified_import)) -# Ignore DeepSource cyclomatic complexity check for this function. It is one -# the main entrypoints so sort of expected to be complex. -# skipcq: PY-R1000 def main(argv: Sequence[str] | None = None, stdin: TextIOWrapper | None = None) -> None: arguments = parse_args(argv) if arguments.get("show_version"): @@ -1047,8 +1043,6 @@ def main(argv: Sequence[str] | None = None, stdin: TextIOWrapper | None = None) config_trie = find_all_configs(config_dict.pop("config_root", ".")) if "src_paths" in config_dict: - # Keep CLI-provided values as-is so wildcard patterns can be expanded later - # relative to the resolved config directory. config_dict["src_paths"] = set(config_dict.get("src_paths", ())) config = Config(**config_dict) @@ -1069,7 +1063,6 @@ def main(argv: Sequence[str] | None = None, stdin: TextIOWrapper | None = None) file_path=file_path, extension=ext_format, ) - wrong_sorted_files = incorrectly_sorted else: try: @@ -1132,23 +1125,20 @@ def main(argv: Sequence[str] | None = None, stdin: TextIOWrapper | None = None) with executor_ctx as executor: if executor is not None: - attempt_iterator = executor.imap( - functools.partial( - sort_imports, - config=config, - check=check, - ask_to_apply=ask_to_apply, - show_diff=show_diff, - write_to_stdout=write_to_stdout, - extension=ext_format, - config_trie=config_trie, - ), - file_names, + partial_sort = functools.partial( + sort_imports, + config=config, + check=check, + ask_to_apply=ask_to_apply, + show_diff=show_diff, + write_to_stdout=write_to_stdout, + extension=ext_format, + config_trie=config_trie, ) + attempt_iterator = executor.imap(partial_sort, file_names) else: - # https://github.com/python/typeshed/pull/2814 attempt_iterator = ( - sort_imports( # type: ignore + sort_imports( file_name, config=config, check=check, @@ -1161,25 +1151,20 @@ def main(argv: Sequence[str] | None = None, stdin: TextIOWrapper | None = None) for file_name in file_names ) - # If any files passed in are missing considered as error, should be removed is_no_attempt = True any_encoding_valid = False for sort_attempt in attempt_iterator: if not sort_attempt: - continue # pragma: no cover - shouldn't happen, satisfies type constraint + continue incorrectly_sorted = sort_attempt.incorrectly_sorted - if arguments.get("check", False) and incorrectly_sorted: + if check and incorrectly_sorted: wrong_sorted_files = True if sort_attempt.skipped: - num_skipped += ( - 1 # pragma: no cover - shouldn't happen, due to skip in iter_source_code - ) - + num_skipped += 1 if not sort_attempt.supported_encoding: num_invalid_encoding += 1 else: any_encoding_valid = True - is_no_attempt = False num_skipped += len(skipped) diff --git a/isort/output.py b/isort/output.py index 9d9ce9aa2..8c62e690e 100644 --- a/isort/output.py +++ b/isort/output.py @@ -1,4 +1,3 @@ -import copy import itertools from collections.abc import Iterable from functools import partial @@ -13,24 +12,17 @@ from .settings import DEFAULT_CONFIG, Config -# Ignore DeepSource cyclomatic complexity check for this function. -# skipcq: PY-R1000 def sorted_imports( parsed: parse.ParsedContent, config: Config = DEFAULT_CONFIG, extension: str = "py", import_type: str = "import", ) -> str: - """Adds the imports back to the file. - - (at the index of the first import) sorted alphabetically and split between groups - - """ if parsed.import_index == -1: return _output_as_string(parsed.lines_without_imports, parsed.line_separator) formatted_output: list[str] = parsed.lines_without_imports.copy() - remove_imports = [format_simplified(removal) for removal in config.remove_imports] + remove_imports = {format_simplified(r) for r in config.remove_imports} sections: Iterable[str] = itertools.chain(parsed.sections, config.forced_separate) @@ -42,157 +34,140 @@ def sorted_imports( "lazy_from": {}, } base_sections: tuple[str, ...] = () - for section in sections: - if section == "FUTURE": + for sec in sections: + if sec == "FUTURE": base_sections = ("FUTURE",) continue - parsed.imports["no_sections"]["straight"].update(parsed.imports[section]["straight"]) - parsed.imports["no_sections"]["from"].update(parsed.imports[section]["from"]) + parsed.imports["no_sections"]["straight"].update(parsed.imports[sec]["straight"]) + parsed.imports["no_sections"]["from"].update(parsed.imports[sec]["from"]) parsed.imports["no_sections"]["lazy_straight"].update( - parsed.imports[section]["lazy_straight"] + parsed.imports[sec]["lazy_straight"] ) - parsed.imports["no_sections"]["lazy_from"].update(parsed.imports[section]["lazy_from"]) + parsed.imports["no_sections"]["lazy_from"].update(parsed.imports[sec]["lazy_from"]) sections = (*base_sections, "no_sections") output: list[str] = [] seen_headings: set[str] = set() pending_lines_before = False + cfg = config for section in sections: - section_output = _build_import_group( - parsed, config, section, remove_imports, import_type, is_lazy=False + sec_out = _build_import_group( + parsed, cfg, section, list(remove_imports), import_type, is_lazy=False ) - - # PEP 810 lazy imports always follow all eager imports within the section. - lazy_section_output = _build_import_group( - parsed, config, section, remove_imports, import_type, is_lazy=True + lazy_sec_out = _build_import_group( + parsed, cfg, section, list(remove_imports), import_type, is_lazy=True ) + if lazy_sec_out: + sec_out = ( + sec_out + [""] * cfg.lines_between_types + lazy_sec_out + if sec_out + else lazy_sec_out + ) - if lazy_section_output: - if section_output: - section_output += [""] * config.lines_between_types + lazy_section_output - else: - section_output = lazy_section_output - - section_name = section - no_lines_before = section_name in config.no_lines_before - - if section_output: - if section_name in parsed.place_imports: - parsed.place_imports[section_name] = section_output + no_lines_before = section in cfg.no_lines_before + if sec_out: + if section in parsed.place_imports: + parsed.place_imports[section] = sec_out continue - section_title = config.import_headings.get(section_name.lower(), "") - if section_title and section_title not in seen_headings: - if config.dedup_headings: - seen_headings.add(section_title) - section_comment = f"# {section_title}" - if section_comment not in parsed.lines_without_imports[0:1]: # pragma: no branch - section_output.insert(0, section_comment) - - section_footer = config.import_footers.get(section_name.lower(), "") - if section_footer and section_footer not in seen_headings: - if config.dedup_headings: - seen_headings.add(section_footer) - section_comment_end = f"# {section_footer}" - if ( - section_comment_end not in parsed.lines_without_imports[-1:] - ): # pragma: no branch - section_output.append("") # Empty line for black compatibility - section_output.append(section_comment_end) - - if section in config.separate_packages: - section_output = _separate_packages(section_output, config) + title = cfg.import_headings.get(section.lower(), "") + if title and title not in seen_headings: + if cfg.dedup_headings: + seen_headings.add(title) + comment = f"# {title}" + if comment not in parsed.lines_without_imports[:1]: + sec_out.insert(0, comment) + + footer = cfg.import_footers.get(section.lower(), "") + if footer and footer not in seen_headings: + if cfg.dedup_headings: + seen_headings.add(footer) + foot_comment = f"# {footer}" + if foot_comment not in parsed.lines_without_imports[-1:]: + sec_out.append("") + sec_out.append(foot_comment) + + if section in cfg.separate_packages: + sec_out = _separate_packages(sec_out, cfg) if pending_lines_before or not no_lines_before: - output += [""] * config.lines_between_sections - - output += section_output - + output += [""] * cfg.lines_between_sections + output += sec_out pending_lines_before = False else: pending_lines_before = pending_lines_before or not no_lines_before - if config.ensure_newline_before_comments: + if cfg.ensure_newline_before_comments: output = _ensure_newline_before_comment(output) while output and output[-1].strip() == "": - output.pop() # pragma: no cover + output.pop() while output and output[0].strip() == "": output.pop(0) - if config.formatting_function: - output = config.formatting_function( - parsed.line_separator.join(output), extension, config + if cfg.formatting_function: + output = cfg.formatting_function( + parsed.line_separator.join(output), extension, cfg ).splitlines() - output_at = 0 - if parsed.import_index < parsed.original_line_count: - output_at = parsed.import_index - formatted_output[output_at:0] = output + output_at = parsed.import_index if parsed.import_index < parsed.original_line_count else 0 + formatted_output[output_at:output_at] = output if output: imports_tail = output_at + len(output) - while [ - character.strip() for character in formatted_output[imports_tail : imports_tail + 1] - ] == [""]: + while imports_tail < len(formatted_output) and formatted_output[imports_tail].strip() == "": formatted_output.pop(imports_tail) - if config.lines_before_imports != -1: - lines_before_imports = config.lines_before_imports - if config.profile == "black" and extension == "pyi": # special case for black - lines_before_imports = 1 - formatted_output[:0] = ["" for line in range(lines_before_imports)] - imports_tail += lines_before_imports + if cfg.lines_before_imports != -1: + lines_before = cfg.lines_before_imports + if cfg.profile == "black" and extension == "pyi": + lines_before = 1 + formatted_output[:0] = ["" for _ in range(lines_before)] + imports_tail += lines_before if len(formatted_output) > imports_tail: next_construct = "" tail = formatted_output[imports_tail:] - - for index, line in enumerate(tail): # pragma: no branch + for idx, line in enumerate(tail): should_skip, in_quote = _parse_utils.skip_line( line, in_quote="", needs_import=False ) if not should_skip and line.strip(): if ( line.strip().startswith("#") - and len(tail) > (index + 1) - and tail[index + 1].strip() + and idx + 1 < len(tail) + and tail[idx + 1].strip() ): continue next_construct = line break - if in_quote: # pragma: no branch + if in_quote: next_construct = line break - if config.lines_after_imports != -1: - lines_after_imports = config.lines_after_imports - if config.profile == "black" and extension == "pyi": # special case for black - lines_after_imports = 1 - formatted_output[imports_tail:0] = ["" for line in range(lines_after_imports)] + if cfg.lines_after_imports != -1: + lines_after = cfg.lines_after_imports + if cfg.profile == "black" and extension == "pyi": + lines_after = 1 + formatted_output[imports_tail:imports_tail] = ["" for _ in range(lines_after)] elif extension != "pyi" and next_construct.startswith(STATEMENT_DECLARATIONS): - formatted_output[imports_tail:0] = ["", ""] + formatted_output[imports_tail:imports_tail] = ["", ""] else: - formatted_output[imports_tail:0] = [""] + formatted_output[imports_tail:imports_tail] = [""] if parsed.place_imports: - new_out_lines = [] - for index, line in enumerate(formatted_output): - new_out_lines.append(line) + new_out = [] + for idx, line in enumerate(formatted_output): + new_out.append(line) if line in parsed.import_placements: - new_out_lines.extend(parsed.place_imports[parsed.import_placements[line]]) - if ( - len(formatted_output) <= (index + 1) - or formatted_output[index + 1].strip() != "" - ): - new_out_lines.append("") - formatted_output = new_out_lines + new_out.extend(parsed.place_imports[parsed.import_placements[line]]) + if idx + 1 >= len(formatted_output) or formatted_output[idx + 1].strip() != "": + new_out.append("") + formatted_output = new_out return _output_as_string(formatted_output, parsed.line_separator) -# Ignore DeepSource cyclomatic complexity check for this function. -# skipcq: PY-R1000 def _build_import_group( parsed: parse.ParsedContent, config: Config, @@ -202,39 +177,31 @@ def _build_import_group( *, is_lazy: bool, ) -> list[str]: - """Build the sorted import lines for one group (eager or lazy) within a section.""" - straight_key: Literal["lazy_straight", "straight"] = "lazy_straight" if is_lazy else "straight" - from_key: Literal["lazy_from", "from"] = "lazy_from" if is_lazy else "from" + straight_key = "lazy_straight" if is_lazy else "straight" + from_key = "lazy_from" if is_lazy else "from" - straight_modules: Iterable[str] = parsed.imports[section][straight_key] + straight_modules = parsed.imports[section][straight_key] if not config.only_sections: straight_modules = sorting.sort( config, straight_modules, - key=lambda key: sorting.module_key( - key, config, section_name=section, straight_import=True - ), + key=lambda k: sorting.module_key(k, config, section_name=section, straight_import=True), reverse=config.reverse_sort, ) - from_modules: Iterable[str] = parsed.imports[section][from_key] + from_modules = parsed.imports[section][from_key] if not config.only_sections: from_modules = sorting.sort( config, from_modules, - key=lambda key: sorting.module_key(key, config, section_name=section), + key=lambda k: sorting.module_key(k, config, section_name=section), reverse=config.reverse_sort, ) - if not is_lazy and config.star_first: - star_modules = [] - other_modules = [] - for module in from_modules: - if "*" in parsed.imports[section]["from"][module]: - star_modules.append(module) - else: - other_modules.append(module) - from_modules = star_modules + other_modules + stars, others = [], [] + for mod in from_modules: + (stars if "*" in parsed.imports[section]["from"][mod] else others).append(mod) + from_modules = stars + others straight_imports = _with_straight_imports( parsed, config, straight_modules, section, remove_imports, import_type, is_lazy=is_lazy @@ -244,46 +211,42 @@ def _build_import_group( ) lines_between = [""] * (config.lines_between_types if from_modules and straight_modules else 0) - if config.from_first or section == "FUTURE": - group_output = from_imports + lines_between + straight_imports - else: - group_output = straight_imports + lines_between + from_imports + group_output = ( + from_imports + lines_between + straight_imports + if config.from_first or section == "FUTURE" + else straight_imports + lines_between + from_imports + ) if config.force_sort_within_sections: - # collapse comments - comments_above: list[str] = [] - new_group_output: list[str] = [] + comments_buf: list[str] = [] + sortable: list[Any] = [] for line in group_output: if not line: continue if line.startswith("#"): - comments_above.append(line) - elif comments_above: - new_group_output.append(_LineWithComments(line, comments_above)) - comments_above = [] + comments_buf.append(line) + elif comments_buf: + sortable.append(_LineWithComments(line, comments_buf)) + comments_buf = [] else: - new_group_output.append(line) - # only_sections option is not imposed if force_sort_within_sections is True - new_group_output = sorting.sort( + sortable.append(line) + sortable = sorting.sort( config, - new_group_output, + sortable, key=partial(sorting.section_key, config=config), reverse=config.reverse_sort, ) - # uncollapse comments group_output = [] - for line in new_group_output: - line_comments = getattr(line, "comments", ()) - if line_comments: - group_output.extend(line_comments) - group_output.append(str(line)) + for item in sortable: + if isinstance(item, _LineWithComments): + group_output.extend(item.comments) + group_output.append(str(item)) + else: + group_output.append(item) return group_output -# Ignore DeepSource cyclomatic complexity check for this function. It was -# already complex when this check was enabled. -# skipcq: PY-R1000 def _with_from_imports( parsed: parse.ParsedContent, config: Config, @@ -295,10 +258,12 @@ def _with_from_imports( is_lazy: bool, ) -> list[str]: output: list[str] = [] - import_key: Literal["lazy_from", "from"] = "lazy_from" if is_lazy else "from" + import_key = "lazy_from" if is_lazy else "from" + cfg = config + rem_set = set(remove_imports) for module in from_modules: - if module in remove_imports: + if module in rem_set: continue import_start = f"from {module} {import_type} " @@ -306,370 +271,262 @@ def _with_from_imports( import_start = f"lazy {import_start}" from_imports = list(parsed.imports[section][import_key][module]) - if ( - not config.no_inline_sort - or (config.force_single_line and module not in config.single_line_exclusions) - ) and not config.only_sections: + if (not cfg.no_inline_sort or (cfg.force_single_line and module not in cfg.single_line_exclusions)) and not cfg.only_sections: from_imports = sorting.sort( - config, + cfg, from_imports, - key=lambda key: sorting.module_key( - key, - config, + key=lambda k: sorting.module_key( + k, + cfg, True, - config.force_alphabetical_sort_within_sections, + cfg.force_alphabetical_sort_within_sections, section_name=section, ), - reverse=config.reverse_sort, + reverse=cfg.reverse_sort, ) - if remove_imports: - from_imports = [ - line for line in from_imports if f"{module}.{line}" not in remove_imports - ] + if rem_set: + from_imports = [ln for ln in from_imports if f"{module}.{ln}" not in rem_set] - sub_modules = [f"{module}.{from_import}" for from_import in from_imports] + sub_modules = [f"{module}.{fi}" for fi in from_imports] as_imports = { - from_import: [ - f"{from_import} as {as_module}" for as_module in parsed.as_map["from"][sub_module] - ] - for from_import, sub_module in zip(from_imports, sub_modules, strict=False) - if sub_module in parsed.as_map["from"] + fi: [f"{fi} as {as_mod}" for as_mod in parsed.as_map["from"][sub]] + for fi, sub in zip(from_imports, sub_modules, strict=False) + if sub in parsed.as_map["from"] } - if config.combine_as_imports and not ("*" in from_imports and config.combine_star): - if not config.no_inline_sort: - for as_import in as_imports: - if not config.only_sections: - as_imports[as_import] = sorting.sort(config, as_imports[as_import]) - for from_import in copy.copy(from_imports): - if from_import in as_imports: - idx = from_imports.index(from_import) - if parsed.imports[section][import_key][module][from_import]: - from_imports[(idx + 1) : (idx + 1)] = as_imports.pop(from_import) + + if cfg.combine_as_imports and not ("*" in from_imports and cfg.combine_star): + if not cfg.no_inline_sort: + for key in as_imports: + if not cfg.only_sections: + as_imports[key] = sorting.sort(cfg, as_imports[key]) + for fi in list(from_imports): + if fi in as_imports: + idx = from_imports.index(fi) + if parsed.imports[section][import_key][module][fi]: + from_imports[idx + 1 : idx + 1] = as_imports.pop(fi) else: - from_imports[idx : (idx + 1)] = as_imports.pop(from_import) + from_imports[idx : idx + 1] = as_imports.pop(fi) - only_show_as_imports = False - comments: list[str] | None = parsed.categorized_comments["from"].pop(module, None) + only_show_as = False + comments = parsed.categorized_comments["from"].pop(module, None) above_comments = parsed.categorized_comments["above"]["from"].pop(module, None) + while from_imports: if above_comments: output.extend(above_comments) above_comments = None - if "*" in from_imports and config.combine_star: - import_statement = wrap.line( + if "*" in from_imports and cfg.combine_star: + stmt = wrap.line( with_comments( _with_star_comments(parsed, module, list(comments or ())), f"{import_start}*", - removed=config.ignore_comments, - comment_prefix=config.comment_prefix, + removed=cfg.ignore_comments, + comment_prefix=cfg.comment_prefix, ), parsed.line_separator, - config, + cfg, ) - from_imports = [ - from_import for from_import in from_imports if from_import in as_imports - ] - only_show_as_imports = True - elif config.force_single_line and module not in config.single_line_exclusions: - import_statement = "" + output.append(stmt) + from_imports = [fi for fi in from_imports if fi in as_imports] + only_show_as = True + continue + + if cfg.force_single_line and module not in cfg.single_line_exclusions: while from_imports: - from_import = from_imports.pop(0) - single_import_line = with_comments( + fi = from_imports.pop(0) + line = with_comments( comments, - import_start + from_import, - removed=config.ignore_comments, - comment_prefix=config.comment_prefix, + import_start + fi, + removed=cfg.ignore_comments, + comment_prefix=cfg.comment_prefix, ) - comment = ( - parsed.categorized_comments["nested"].get(module, {}).pop(from_import, None) - ) - if comment is not None: - comment_text = f" {comment}" if comment else "" - single_import_line += ( - f"{(comments and ';') or config.comment_prefix}{comment_text}" - ) - if from_import in as_imports: - if ( - parsed.imports[section][import_key][module][from_import] - and not only_show_as_imports - ): + nested = parsed.categorized_comments["nested"].get(module, {}).pop(fi, None) + if nested is not None: + line += f"{(comments and ';') or cfg.comment_prefix} {nested}" + if fi in as_imports: + if parsed.imports[section][import_key][module][fi] and not only_show_as: + output.append(wrap.line(line, parsed.line_separator, cfg)) + f_comments = parsed.categorized_comments["straight"].get(f"{module}.{fi}") + sorted_as = sorting.sort(cfg, as_imports[fi]) if not cfg.only_sections else as_imports[fi] + for as_imp in sorted_as: output.append( - wrap.line(single_import_line, parsed.line_separator, config) - ) - from_comments = parsed.categorized_comments["straight"].get( - f"{module}.{from_import}" - ) - - if not config.only_sections: - output.extend( with_comments( - from_comments, - wrap.line( - import_start + as_import, parsed.line_separator, config - ), - removed=config.ignore_comments, - comment_prefix=config.comment_prefix, + f_comments, + wrap.line(import_start + as_imp, parsed.line_separator, cfg), + removed=cfg.ignore_comments, + comment_prefix=cfg.comment_prefix, ) - for as_import in sorting.sort(config, as_imports[from_import]) - ) - - else: - output.extend( - with_comments( - from_comments, - wrap.line( - import_start + as_import, parsed.line_separator, config - ), - removed=config.ignore_comments, - comment_prefix=config.comment_prefix, - ) - for as_import in as_imports[from_import] ) else: - output.append(wrap.line(single_import_line, parsed.line_separator, config)) + output.append(wrap.line(line, parsed.line_separator, cfg)) comments = None - else: - # Tracks whether any aliased imports were emitted before the grouped - # non-aliased imports in this pass of the outer loop. When True it - # suppresses the split_on_trailing_comma explode behaviour for the - # non-aliased group, because those imports are not the sole content - # of the statement and forcing them onto individual lines would break - # the intended output structure. - processed_as_imports_this_iteration = False - while from_imports and from_imports[0] in as_imports: - processed_as_imports_this_iteration = True - from_import = from_imports.pop(0) - - if not config.only_sections: - as_imports[from_import] = sorting.sort(config, as_imports[from_import]) - from_comments = ( - parsed.categorized_comments["straight"].get(f"{module}.{from_import}") or [] + continue + + processed_as = False + while from_imports and from_imports[0] in as_imports: + processed_as = True + fi = from_imports.pop(0) + if not cfg.only_sections: + as_imports[fi] = sorting.sort(cfg, as_imports[fi]) + f_comments = parsed.categorized_comments["straight"].get(f"{module}.{fi}") or [] + if parsed.imports[section][import_key][module][fi] and not only_show_as: + spec_comm = parsed.categorized_comments["nested"].get(module, {}).pop(fi, None) + if spec_comm is not None: + f_comments.append(spec_comm) + output.append( + wrap.line( + with_comments( + f_comments, + import_start + fi, + removed=cfg.ignore_comments, + comment_prefix=cfg.comment_prefix, + ), + parsed.line_separator, + cfg, + ) ) - if ( - parsed.imports[section][import_key][module][from_import] - and not only_show_as_imports - ): - specific_comment = ( - parsed.categorized_comments["nested"] - .get(module, {}) - .pop(from_import, None) + for as_imp in as_imports[fi]: + opening_comments = list(f_comments) + spec_comm = parsed.categorized_comments["nested"].get(module, {}).pop(as_imp, None) + if spec_comm is not None: + f_comments.append(spec_comm) + imp_line = import_start + as_imp + if opening_comments and cfg.use_parentheses: + lines = wrap.line( + with_comments( + [spec_comm] if spec_comm else [], + imp_line, + removed=cfg.ignore_comments, + comment_prefix=cfg.comment_prefix, + ), + parsed.line_separator, + cfg, + ).split(parsed.line_separator) + opening = with_comments( + opening_comments, + "", + removed=cfg.ignore_comments, + comment_prefix=cfg.comment_prefix, ) - if specific_comment is not None: - from_comments.append(specific_comment) + if opening: + lines[0] += opening + output.append(parsed.line_separator.join(lines)) + else: output.append( wrap.line( with_comments( - from_comments, - import_start + from_import, - removed=config.ignore_comments, - comment_prefix=config.comment_prefix, + f_comments, + imp_line, + removed=cfg.ignore_comments, + comment_prefix=cfg.comment_prefix, ), parsed.line_separator, - config, + cfg, ) ) - from_comments = [] - - for as_import in as_imports[from_import]: - # `from_comments` at this point contains any comments that appeared on - # the *opening* "from X import" line. These are distinct from - # `specific_comment`, which is an inline comment on the attribute line - # itself. We snapshot `from_comments` here so that we can later distinguish - # the two types: opening-line comments must stay on the "import (" line - # when parentheses are used, while attribute-line comments stay on the - # import attribute line. - opening_line_comments = list(from_comments) - specific_comment = ( - parsed.categorized_comments["nested"] - .get(module, {}) - .pop(as_import, None) - ) - # Collect the attribute-line comment (if any) separately so it can be - # embedded in the attribute line regardless of wrapping mode. - if specific_comment is not None: - from_comments.append(specific_comment) - - import_line = import_start + as_import - if opening_line_comments and config.use_parentheses: - # When parentheses are used, opening-line comments (e.g. "# noqa") must - # remain on the "from X import (" line. If we naively embedded them in - # the attribute string and then called wrap.line(), the comment would - # end up inside the parentheses on the alias attribute line. - # Wrap the import with only the attribute-line comment. Afterwards, add - # the opening-line comment back to the first line of the wrapped import - # statement. - lines = wrap.line( - with_comments( - [specific_comment] if specific_comment else [], - import_line, - removed=config.ignore_comments, - comment_prefix=config.comment_prefix, - ), - parsed.line_separator, - config, - ).split(parsed.line_separator) - - opening_comment = with_comments( - opening_line_comments, - "", - removed=config.ignore_comments, - comment_prefix=config.comment_prefix, - ) - if opening_comment: - lines[0] += opening_comment - output.append(parsed.line_separator.join(lines)) - else: - output.append( - wrap.line( - with_comments( - from_comments, - import_line, - removed=config.ignore_comments, - comment_prefix=config.comment_prefix, - ), - parsed.line_separator, - config, - ) - ) + f_comments = [] - from_comments = [] - - if "*" in from_imports: - output.append( - with_comments( - _with_star_comments(parsed, module, []), - f"{import_start}*", - removed=config.ignore_comments, - comment_prefix=config.comment_prefix, - ) - ) - from_imports.remove("*") - - for from_import in copy.copy(from_imports): - comment = ( - parsed.categorized_comments["nested"].get(module, {}).pop(from_import, None) - ) - if comment is not None: - # If the comment is a noqa and hanging indent wrapping is used, - # keep the name in the main list and hoist the comment to the statement. - if ( - comment.lower().startswith("noqa") - and config.multi_line_output == wrap.Modes.HANGING_INDENT # type: ignore[attr-defined] # noqa: E501 - ): - comments = list(comments) if comments else [] - comments.append(comment) - continue - - from_imports.remove(from_import) - if from_imports: - use_comments: list[str] | None = [] - else: - use_comments = comments - comments = None - single_import_line = with_comments( - use_comments, - import_start + from_import, - removed=config.ignore_comments, - comment_prefix=config.comment_prefix, - ) - comment_text = f" {comment}" if comment else "" - single_import_line += ( - f"{(use_comments and ';') or config.comment_prefix}{comment_text}" - ) - output.append(wrap.line(single_import_line, parsed.line_separator, config)) - - from_import_section = [] - while from_imports and ( - from_imports[0] not in as_imports - or ( - config.combine_as_imports - and parsed.imports[section][import_key][module][from_import] + if "*" in from_imports: + output.append( + with_comments( + _with_star_comments(parsed, module, []), + f"{import_start}*", + removed=cfg.ignore_comments, + comment_prefix=cfg.comment_prefix, ) - ): - from_import_section.append(from_imports.pop(0)) - if config.combine_as_imports: - comments = (comments or []) + list( - parsed.categorized_comments["from"].pop(f"{module}.__combined_as__", ()) + ) + from_imports.remove("*") + + for fi in list(from_imports): + nested_comm = parsed.categorized_comments["nested"].get(module, {}).pop(fi, None) + if nested_comm is not None: + if nested_comm.lower().startswith("noqa") and cfg.multi_line_output == wrap.Modes.HANGING_INDENT: # type: ignore[attr-defined] + comments = list(comments) if comments else [] + comments.append(nested_comm) + continue + from_imports.remove(fi) + use_comm = [] if from_imports else comments + if not from_imports: + comments = None + line = with_comments( + use_comm, + import_start + fi, + removed=cfg.ignore_comments, + comment_prefix=cfg.comment_prefix, ) - import_statement = with_comments( - comments, - import_start + (", ").join(from_import_section), - removed=config.ignore_comments, - comment_prefix=config.comment_prefix, + line += f"{(use_comm and ';') or cfg.comment_prefix} {nested_comm}" if nested_comm else "" + output.append(wrap.line(line, parsed.line_separator, cfg)) + + remaining = [] + while from_imports and ( + from_imports[0] not in as_imports + or (cfg.combine_as_imports and parsed.imports[section][import_key][module][from_imports[0]]) + ): + remaining.append(from_imports.pop(0)) + if cfg.combine_as_imports: + comments = (comments or []) + list( + parsed.categorized_comments["from"].pop(f"{module}.__combined_as__", ()) + ) + stmt = with_comments( + comments, + import_start + (", ").join(remaining), + removed=cfg.ignore_comments, + comment_prefix=cfg.comment_prefix, + ) + if not remaining: + stmt = "" + + do_multi = False + if cfg.force_grid_wrap and len(remaining) >= cfg.force_grid_wrap: + do_multi = True + if len(stmt) > cfg.line_length and len(remaining) > 1: + do_multi = True + if ( + len(stmt) > cfg.line_length + and remaining + and cfg.multi_line_output not in (wrap.Modes.GRID, wrap.Modes.VERTICAL) # type: ignore + ): + do_multi = True + + if stmt and cfg.split_on_trailing_comma and module in parsed.trailing_commas and not processed_as: + stmt = wrap.import_statement( + import_start=import_start, + from_imports=remaining, + comments=comments, + line_separator=parsed.line_separator, + config=cfg, + explode=True, + ) + elif do_multi: + stmt = wrap.import_statement( + import_start=import_start, + from_imports=remaining, + comments=comments, + line_separator=parsed.line_separator, + config=cfg, ) - if not from_import_section: - import_statement = "" - - do_multiline_reformat = False - - force_grid_wrap = config.force_grid_wrap - if force_grid_wrap and len(from_import_section) >= force_grid_wrap: - do_multiline_reformat = True - - if len(import_statement) > config.line_length and len(from_import_section) > 1: - do_multiline_reformat = True - - # If line too long AND have imports AND we are - # NOT using GRID or VERTICAL wrap modes - if ( - len(import_statement) > config.line_length - and len(from_import_section) > 0 - and config.multi_line_output not in (wrap.Modes.GRID, wrap.Modes.VERTICAL) # type: ignore # noqa: E501 - ): - do_multiline_reformat = True - - if ( - import_statement - and config.split_on_trailing_comma - and module in parsed.trailing_commas - and not processed_as_imports_this_iteration - ): - import_statement = wrap.import_statement( + if cfg.multi_line_output == wrap.Modes.GRID: # type: ignore + alt = wrap.import_statement( import_start=import_start, - from_imports=from_import_section, + from_imports=remaining, comments=comments, line_separator=parsed.line_separator, - config=config, - explode=True, + config=cfg, + multi_line_output=wrap.Modes.VERTICAL_GRID, # type: ignore ) + if max(len(l) for l in stmt.split(parsed.line_separator)) > cfg.line_length: + stmt = alt + elif len(stmt) > cfg.line_length: + stmt = wrap.line(stmt, parsed.line_separator, cfg) + + comments = None + if stmt: + output.append(stmt) - elif do_multiline_reformat: - import_statement = wrap.import_statement( - import_start=import_start, - from_imports=from_import_section, - comments=comments, - line_separator=parsed.line_separator, - config=config, - ) - if config.multi_line_output == wrap.Modes.GRID: # type: ignore - other_import_statement = wrap.import_statement( - import_start=import_start, - from_imports=from_import_section, - comments=comments, - line_separator=parsed.line_separator, - config=config, - multi_line_output=wrap.Modes.VERTICAL_GRID, # type: ignore - ) - if ( - max( - len(import_line) - for import_line in import_statement.split(parsed.line_separator) - ) - > config.line_length - ): - import_statement = other_import_statement - elif len(import_statement) > config.line_length: - import_statement = wrap.line(import_statement, parsed.line_separator, config) - comments = None - - if import_statement: - output.append(import_statement) return output -# Ignore DeepSource cyclomatic complexity check for this function. -# skipcq: PY-R1000 def _with_straight_imports( parsed: parse.ParsedContent, config: Config, @@ -681,70 +538,56 @@ def _with_straight_imports( is_lazy: bool, ) -> list[str]: output: list[str] = [] - import_type = f"lazy {import_type}" if is_lazy else import_type + rem_set = set(remove_imports) - as_imports = any(module in parsed.as_map["straight"] for module in straight_modules) + as_present = any(m in parsed.as_map["straight"] for m in straight_modules) - # combine_straight_imports only works for bare imports, 'as' imports not included - if config.combine_straight_imports and not as_imports: + if config.combine_straight_imports and not as_present: if not straight_modules: return [] - - above_comments: list[str] = [] - inline_comments: list[str] = [] - - for module in straight_modules: - if module in parsed.categorized_comments["above"]["straight"]: - above_comments.extend(parsed.categorized_comments["above"]["straight"].pop(module)) - if module in parsed.categorized_comments["straight"]: - inline_comments.extend(parsed.categorized_comments["straight"][module]) - - combined_straight_imports = ", ".join(straight_modules) - - output.extend(above_comments) - - if inline_comments: - combined_inline_comments = " ".join(c for c in inline_comments if c) - if combined_inline_comments: - output.append( - f"{import_type} {combined_straight_imports} # {combined_inline_comments}" - ) - else: - output.append(f"{import_type} {combined_straight_imports} #") + above: list[str] = [] + inline: list[str] = [] + for mod in straight_modules: + if mod in parsed.categorized_comments["above"]["straight"]: + above.extend(parsed.categorized_comments["above"]["straight"].pop(mod)) + if mod in parsed.categorized_comments["straight"]: + inline.extend(parsed.categorized_comments["straight"][mod]) + combined = ", ".join(straight_modules) + output.extend(above) + if inline: + inline_str = " ".join(c for c in inline if c) + output.append(f"{import_type} {combined} # {inline_str}" if inline_str else f"{import_type} {combined} #") else: - output.append(f"{import_type} {combined_straight_imports}") - + output.append(f"{import_type} {combined}") return output - for module in straight_modules: - if module in remove_imports: + for mod in straight_modules: + if mod in rem_set: continue - - import_definition = [] - if module in parsed.as_map["straight"]: - if parsed.imports[section]["lazy_straight" if is_lazy else "straight"][module]: - import_definition.append((f"{import_type} {module}", module)) - import_definition.extend( - (f"{import_type} {module} as {as_import}", f"{module} as {as_import}") - for as_import in parsed.as_map["straight"][module] + definitions: list[tuple[str, str]] = [] + if mod in parsed.as_map["straight"]: + if parsed.imports[section]["lazy_straight" if is_lazy else "straight"][mod]: + definitions.append((f"{import_type} {mod}", mod)) + definitions.extend( + (f"{import_type} {mod} as {as_imp}", f"{mod} as {as_imp}") + for as_imp in parsed.as_map["straight"][mod] ) else: - import_definition.append((f"{import_type} {module}", module)) - - comments_above = parsed.categorized_comments["above"]["straight"].pop(module, None) - if comments_above: - output.extend(comments_above) - output.extend( - with_comments( - parsed.categorized_comments["straight"].get(imodule), - idef, - removed=config.ignore_comments, - comment_prefix=config.comment_prefix, + definitions.append((f"{import_type} {mod}", mod)) + + above = parsed.categorized_comments["above"]["straight"].pop(mod, None) + if above: + output.extend(above) + for idef, imodule in definitions: + output.append( + with_comments( + parsed.categorized_comments["straight"].get(imodule), + idef, + removed=config.ignore_comments, + comment_prefix=config.comment_prefix, + ) ) - for idef, imodule in import_definition - ) - return output @@ -754,8 +597,7 @@ def _output_as_string(lines: list[str], line_separator: str) -> str: def _normalize_empty_lines(lines: list[str]) -> list[str]: while lines and lines[-1].strip() == "": - lines.pop(-1) - + lines.pop() lines.append("") return lines @@ -763,9 +605,7 @@ def _normalize_empty_lines(lines: list[str]) -> list[str]: class _LineWithComments(str): comments: list[str] - def __new__( - cls: type["_LineWithComments"], value: Any, comments: list[str] - ) -> "_LineWithComments": + def __new__(cls: type["_LineWithComments"], value: Any, comments: list[str]) -> "_LineWithComments": instance = super().__new__(cls, value) instance.comments = comments return instance @@ -775,7 +615,7 @@ def _ensure_newline_before_comment(output: list[str]) -> list[str]: new_output: list[str] = [] def is_comment(line: str | None) -> bool: - return line.startswith("#") if line else False + return bool(line and line.startswith("#")) for line, prev_line in zip(output, [None, *output], strict=False): if is_comment(line) and prev_line != "" and not is_comment(prev_line): @@ -794,32 +634,31 @@ def _with_star_comments(parsed: parse.ParsedContent, module: str, comments: list def _separate_packages(section_output: list[str], config: Config) -> list[str]: group_keys: set[str] = set() comments_above: list[str] = [] - processed_section_output: list[str] = [] + processed: list[str] = [] - for section_line in section_output: - if section_line.startswith("#"): - comments_above.append(section_line) + for line in section_output: + if line.startswith("#"): + comments_above.append(line) continue - package_name: str = section_line.split(" ")[1] + package_name = line.split(" ")[1] _, reason = module_with_reason(package_name, config) if "Matched configured known pattern" in reason: - package_depth = len(reason.split(".")) - 1 # minus 1 for re.compile - key = ".".join(package_name.split(".")[: package_depth + 1]) + depth = len(reason.split(".")) - 1 + key = ".".join(package_name.split(".")[: depth + 1]) else: key = package_name.split(".")[0] if key not in group_keys: if group_keys: - processed_section_output.append("") - + processed.append("") group_keys.add(key) if comments_above: - processed_section_output.extend(comments_above) + processed.extend(comments_above) comments_above = [] - processed_section_output.append(section_line) + processed.append(line) - return processed_section_output + return processed \ No newline at end of file diff --git a/isort/parse.py b/isort/parse.py index 036783975..47e284afb 100644 --- a/isort/parse.py +++ b/isort/parse.py @@ -76,17 +76,15 @@ class ParsedContent(NamedTuple): # skipcq: PY-R1000 def file_contents(contents: str, config: Config = DEFAULT_CONFIG) -> ParsedContent: """Parses a python file taking out and categorizing imports.""" - line_separator: str = config.line_ending or _infer_line_separator(contents) + line_separator = config.line_ending or _infer_line_separator(contents) in_lines = contents.splitlines() if contents and contents[-1] in ("\n", "\r"): in_lines.append("") - - out_lines = [] + out_lines: list[str] = [] original_line_count = len(in_lines) finder = partial(place.module, config=config) line_count = len(in_lines) - place_imports: dict[str, list[str]] = {} import_placements: dict[str, str] = {} as_map: dict[str, dict[str, list[str]]] = { @@ -103,6 +101,7 @@ def file_contents(contents: str, config: Config = DEFAULT_CONFIG) -> ParsedConte "lazy_straight": OrderedDict(), "lazy_from": OrderedDict(), } + categorized_comments: CommentsDict = { "from": {}, "straight": {}, @@ -115,16 +114,27 @@ def file_contents(contents: str, config: Config = DEFAULT_CONFIG) -> ParsedConte index = 0 import_index = -1 in_quote = "" + + # Local copies of frequently accessed config attributes + sect_comments = config.section_comments + sect_comments_end = config.section_comments_end + float_to_top = config.float_to_top + treat_all_comments_as_code = config.treat_all_comments_as_code + treat_comments_as_code = config.treat_comments_as_code + verbose = config.verbose + only_modified = config.only_modified + force_single_line = config.force_single_line + combine_as_imports = config.combine_as_imports + remove_redundant_aliases = config.remove_redundant_aliases + while index < line_count: line = in_lines[index] index += 1 statement_index = index - (skipping_line, in_quote) = skip_line(line, in_quote=in_quote) + skipping_line, in_quote = skip_line(line, in_quote=in_quote) - if ( - line in config.section_comments or line in config.section_comments_end - ) and not skipping_line: - if import_index == -1: # pragma: no branch + if (line in sect_comments or line in sect_comments_end) and not skipping_line: + if import_index == -1: import_index = index - 1 continue @@ -143,7 +153,7 @@ def file_contents(contents: str, config: Config = DEFAULT_CONFIG) -> ParsedConte lstripped_line = line.lstrip() if ( - config.float_to_top + float_to_top and import_index == -1 and line and not in_quote @@ -151,43 +161,30 @@ def file_contents(contents: str, config: Config = DEFAULT_CONFIG) -> ParsedConte and not lstripped_line.startswith("'''") and not lstripped_line.startswith('"""') ): - if not lstripped_line.startswith("import") and not lstripped_line.startswith("from"): + if not lstripped_line.startswith(("import", "from")): import_index = index - 1 while import_index and not in_lines[import_index - 1]: import_index -= 1 else: commentless = line.split("#", 1)[0].strip() - if ( - ("isort:skip" in line or "isort: skip" in line) - and "(" in commentless - and ")" not in commentless - ): + if ("isort:skip" in line or "isort: skip" in line) and "(" in commentless and ")" not in commentless: import_index = index - starting_line = line while "isort:skip" in starting_line or "isort: skip" in starting_line: commentless = starting_line.split("#", 1)[0] - if ( - "(" in commentless - and not commentless.rstrip().endswith(")") - and import_index < line_count - ): - while import_index < line_count and not commentless.rstrip().endswith( - ")" - ): + if "(" in commentless and not commentless.rstrip().endswith(")") and import_index < line_count: + while import_index < line_count and not commentless.rstrip().endswith(")"): commentless = in_lines[import_index].split("#", 1)[0] import_index += 1 else: import_index += 1 - if import_index >= line_count: break - starting_line = in_lines[import_index] line, *end_of_line_comment = line.split("#", 1) if ";" in line: - statements = [line.strip() for line in line.split(";")] + statements = [stmt.strip() for stmt in line.split(";")] else: statements = [line] if end_of_line_comment: @@ -201,10 +198,6 @@ def file_contents(contents: str, config: Config = DEFAULT_CONFIG) -> ParsedConte out_lines.append(raw_line) continue - # Detect PEP 810 lazy imports (``lazy import X`` / ``lazy from X import Y``). - # We strip the ``lazy `` prefix so the rest of the parsing logic works normally - # on the resulting ``import X`` / ``from X import Y`` string. The original - # lazy type is remembered in ``is_lazy`` and used later when storing the result. is_lazy = type_of_import in ("lazy_straight", "lazy_from") if is_lazy: line = line[len("lazy ") :] @@ -212,10 +205,12 @@ def file_contents(contents: str, config: Config = DEFAULT_CONFIG) -> ParsedConte if import_index == -1: import_index = index - 1 - nested_comments = {} + + nested_comments: dict[str, str] = {} import_string, comment = parse_comments(line) comments = [comment] if comment is not None else [] - line_parts = [part for part in strip_syntax(import_string).strip().split(" ") if part] + + line_parts = [p for p in strip_syntax(import_string).strip().split(" ") if p] if type_of_import == "from" and len(line_parts) == 2 and comments: nested_comments[line_parts[-1]] = comments[0] @@ -232,7 +227,6 @@ def _get_next_line() -> tuple[str, str | None]: ) for extra_line in extra_lines: raw_lines.append(extra_line.line) - # If during parsing of the continuation lines we encounter a comment, we record it. if extra_line.comment is not None: comments.append(extra_line.comment) stripped_line = strip_syntax(extra_line.line).strip() @@ -250,14 +244,15 @@ def _get_next_line() -> tuple[str, str | None]: continue just_imports = [ - item.replace("{|", "{ ").replace("|}", " }") - for item in strip_syntax(import_string).split() + itm.replace("{|", "{ ").replace("|}", " }") + for itm in strip_syntax(import_string).split() ] attach_comments_to: list[str] | None = None direct_imports = just_imports[1:] straight_import = True top_level_module = "" + if "as" in just_imports and (just_imports.index("as") + 1) < len(just_imports): straight_import = False while "as" in just_imports: @@ -266,15 +261,14 @@ def _get_next_line() -> tuple[str, str | None]: if type_of_import == "from": nested_module = just_imports[as_index - 1] top_level_module = just_imports[0] - module = top_level_module + "." + nested_module + module = f"{top_level_module}.{nested_module}" as_name = just_imports[as_index + 1] direct_imports.remove(nested_module) direct_imports.remove(as_name) direct_imports.remove("as") - if nested_module == as_name and config.remove_redundant_aliases: - pass - elif as_name not in as_map["from"][module]: # pragma: no branch - as_map["from"][module].append(as_name) + if not (nested_module == as_name and remove_redundant_aliases): + if as_name not in as_map["from"][module]: + as_map["from"][module].append(as_name) full_name = f"{nested_module} as {as_name}" associated_comment = nested_comments.get(full_name) @@ -282,24 +276,23 @@ def _get_next_line() -> tuple[str, str | None]: categorized_comments["nested"].setdefault(top_level_module, {})[ full_name ] = associated_comment - if associated_comment in comments: # pragma: no branch + if associated_comment in comments: comments.pop(comments.index(associated_comment)) else: module = just_imports[as_index - 1] as_name = just_imports[as_index + 1] - if module == as_name and config.remove_redundant_aliases: - pass - elif as_name not in as_map["straight"][module]: - as_map["straight"][module].append(as_name) + if not (module == as_name and remove_redundant_aliases): + if as_name not in as_map["straight"][module]: + as_map["straight"][module].append(as_name) if comments and attach_comments_to is None: - if nested_module and config.combine_as_imports: + if nested_module and combine_as_imports: attach_comments_to = categorized_comments["from"].setdefault( f"{top_level_module}.__combined_as__", [] ) else: if type_of_import == "from" or ( - config.remove_redundant_aliases and as_name == module.split(".")[-1] + remove_redundant_aliases and as_name == module.split(".")[-1] ): attach_comments_to = categorized_comments["straight"].setdefault( module, [] @@ -313,10 +306,9 @@ def _get_next_line() -> tuple[str, str | None]: if type_of_import == "from": import_from = just_imports.pop(0) placed_module = finder(import_from) - if config.verbose and not config.only_modified: + if verbose and not only_modified: print(f"from-type place_module for {import_from} returned {placed_module}") - - elif config.verbose: + elif verbose: verbose_output.append( f"from-type place_module for {import_from} returned {placed_module}" ) @@ -326,7 +318,6 @@ def _get_next_line() -> tuple[str, str | None]: " Do you need to define a default section?", stacklevel=2, ) - if placed_module and placed_module not in imports: raise MissingSection(import_module=import_from, section=placed_module) @@ -337,10 +328,11 @@ def _get_next_line() -> tuple[str, str | None]: categorized_comments["nested"].setdefault(import_from, {})[import_name] = ( associated_comment ) - if associated_comment in comments: # pragma: no branch + if associated_comment in comments: comments.pop(comments.index(associated_comment)) + if ( - config.force_single_line + force_single_line and comments and attach_comments_to is None and len(just_imports) == 1 @@ -357,7 +349,8 @@ def _get_next_line() -> tuple[str, str | None]: if comments and attach_comments_to is None: attach_comments_to = categorized_comments["from"].setdefault(import_from, []) - if len(out_lines) > max(import_index, 1) - 1: + threshold = max(import_index, 1) - 1 + if len(out_lines) > threshold: last = out_lines[-1].rstrip() if out_lines else "" while ( last.startswith("#") @@ -365,42 +358,35 @@ def _get_next_line() -> tuple[str, str | None]: and not last.endswith("'''") and "isort:imports-" not in last and "isort: imports-" not in last - and not config.treat_all_comments_as_code - and last.strip() not in config.treat_comments_as_code + and not treat_all_comments_as_code + and last.strip() not in treat_comments_as_code ): categorized_comments["above"]["from"].setdefault(import_from, []).insert( 0, out_lines.pop(-1) ) - if out_lines: - last = out_lines[-1].rstrip() - else: - last = "" - if statement_index - 1 == import_index: # pragma: no cover + last = out_lines[-1].rstrip() if out_lines else "" + if statement_index - 1 == import_index: import_index -= len( categorized_comments["above"]["from"].get(import_from, []) ) if import_from not in root: root[import_from] = OrderedDict( - (module, module in direct_imports) for module in just_imports + (mod, mod in direct_imports) for mod in just_imports ) else: root[import_from].update( - (module, root[import_from].get(module, False) or module in direct_imports) - for module in just_imports + (mod, root[import_from].get(mod, False) or mod in direct_imports) + for mod in just_imports ) if comments and attach_comments_to is not None: attach_comments_to.extend(comments) - if ( - just_imports - and just_imports[-1] - and "," in import_string.split(just_imports[-1])[-1] - ): + if just_imports and just_imports[-1] and "," in import_string.split(just_imports[-1])[-1]: trailing_commas.add(import_from) else: - assert type_of_import == "straight" # noqa: S101 # Only needed for type checker + # straight import handling if comments and attach_comments_to is not None: attach_comments_to.extend(comments) comments = [] @@ -410,7 +396,8 @@ def _get_next_line() -> tuple[str, str | None]: categorized_comments["straight"][module] = comments comments = [] - if len(out_lines) > max(import_index, +1, 1) - 1: + threshold = max(import_index, 1) - 1 + if len(out_lines) > threshold: last = out_lines[-1].rstrip() if out_lines else "" while ( last.startswith("#") @@ -418,25 +405,21 @@ def _get_next_line() -> tuple[str, str | None]: and not last.endswith("'''") and "isort:imports-" not in last and "isort: imports-" not in last - and not config.treat_all_comments_as_code - and last.strip() not in config.treat_comments_as_code + and not treat_all_comments_as_code + and last.strip() not in treat_comments_as_code ): categorized_comments["above"]["straight"].setdefault(module, []).insert( 0, out_lines.pop(-1) ) - if out_lines: - last = out_lines[-1].rstrip() - else: - last = "" + last = out_lines[-1].rstrip() if out_lines else "" if index - 1 == import_index: import_index -= len( categorized_comments["above"]["straight"].get(module, []) ) placed_module = finder(module) - if config.verbose and not config.only_modified: + if verbose and not only_modified: print(f"else-type place_module for {module} returned {placed_module}") - - elif config.verbose: + elif verbose: verbose_output.append( f"else-type place_module for {module} returned {placed_module}" ) @@ -455,7 +438,6 @@ def _get_next_line() -> tuple[str, str | None]: "lazy_from": OrderedDict(), }, ) - if placed_module and placed_module not in imports: raise MissingSection(import_module=module, section=placed_module) @@ -485,4 +467,4 @@ def _get_next_line() -> tuple[str, str | None]: sections=config.sections, verbose_output=verbose_output, trailing_commas=trailing_commas, - ) + ) \ No newline at end of file diff --git a/isort/place.py b/isort/place.py index 2f863b35c..fadb45ce8 100644 --- a/isort/place.py +++ b/isort/place.py @@ -52,8 +52,8 @@ def _local(name: str, config: Config) -> tuple[str, str] | None: def _known_pattern(name: str, config: Config) -> tuple[str, str] | None: parts = name.split(".") - module_names_to_check = (".".join(parts[:first_k]) for first_k in range(len(parts), 0, -1)) - for module_name_to_check in module_names_to_check: + for first_k in range(len(parts), 0, -1): + module_name_to_check = ".".join(parts[:first_k]) for pattern, placement in config.known_patterns: if placement in config.sections and pattern.match(module_name_to_check): return (placement, f"Matched configured known pattern {pattern}") @@ -96,6 +96,7 @@ def _src_path( return None +@lru_cache(maxsize=1000) def _is_module(path: Path) -> bool: return ( exists_case_sensitive(str(path.with_suffix(".py"))) @@ -107,10 +108,12 @@ def _is_module(path: Path) -> bool: ) +@lru_cache(maxsize=1000) def _is_package(path: Path) -> bool: return exists_case_sensitive(str(path)) and path.is_dir() +@lru_cache(maxsize=1000) def _is_namespace_package(path: Path, src_extensions: frozenset[str]) -> bool: if not _is_package(path): return False @@ -131,16 +134,15 @@ def _is_namespace_package(path: Path, src_extensions: frozenset[str]) -> bool: if ( b"__import__('pkg_resources').declare_namespace(__name__)" not in file_start and b'__import__("pkg_resources").declare_namespace(__name__)' not in file_start - and b"__path__ = __import__('pkgutil').extend_path(__path__, __name__)" - not in file_start - and b'__path__ = __import__("pkgutil").extend_path(__path__, __name__)' - not in file_start + and b"__path__ = __import__('pkgutil').extend_path(__path__, __name__)" not in file_start + and b'__path__ = __import__("pkgutil").extend_path(__path__, __name__)' not in file_start ): return False return True +@lru_cache(maxsize=1000) def _src_path_is_module(src_path: Path, module_name: str) -> bool: return ( module_name == src_path.name and src_path.is_dir() and exists_case_sensitive(str(src_path)) - ) + ) \ No newline at end of file