def _apply_patch()

in sync/commit.py [0:0]

• 50 lines of code
• 13 McCabe index (conditional complexity)

def _apply_patch(patch: bytes,
                 message: bytes,
                 rev_name: str,
                 dest_repo: Repo,
                 skip_empty: bool = True,
                 msg_filter: MsgFilterFunc | None = None,
                 metadata: dict[str, str] | None = None,
                 src_prefix: str | None = None,
                 dest_prefix: str | None = None,
                 amend: bool = False,
                 three_way: bool = True,
                 author: bytes | None = None,
                 exclude: set[str] | None = None,
                 patch_fallback: bool = False,
                 allow_empty: bool = False,
                 ) -> Commit | None:
    assert isinstance(patch, bytes)

    if skip_empty and (not patch or patch.isspace() or
                       not any(line.startswith(b"diff ") for line in patch.splitlines())):
        return None

    if metadata is None:
        metadata = {}

    if msg_filter:
        msg, metadata_extra = msg_filter(message)
    else:
        msg, metadata_extra = message, {}

    if metadata_extra:
        metadata.update(metadata_extra)

    msg = Commit.make_commit_msg(msg, metadata)

    working_dir = dest_repo.working_dir

    assert working_dir is not None

    with Store(dest_repo, rev_name + ".message", msg) as message_path:
        strip_dirs = len(src_prefix.split("/")) + 1 if src_prefix else 1
        with Store(dest_repo, rev_name + ".diff", patch) as patch_path:
            # Without this tests were failing with "Index does not match"
            dest_repo.git.update_index(refresh=True)
            apply_kwargs: dict[str, Any] = {}
            if dest_prefix:
                apply_kwargs["directory"] = dest_prefix
            if three_way:
                apply_kwargs["3way"] = True
            else:
                apply_kwargs["reject"] = True

            err_msg: str | None = None
            try:
                logger.info("Trying to apply patch")
                dest_repo.git.apply(patch_path, index=True, binary=True,
                                    p=strip_dirs, **apply_kwargs)
                logger.info("Patch applied")
            except git.GitCommandError as e:
                err_msg = """git apply failed