Git

Git helper functions.

ConflictFile dataclass

Container for path and different versions of conflicted notebooks.

UnmergedBlob dataclass

Container for git unmerged blobs.

blob2commit(blob, repo)

Get the short commit message from blob hash.

Source code in databooks/git_utils.py

def blob2commit(blob: Blob, repo: Repo) -> str:
    """Get the short commit message from blob hash."""
    _git = Git(working_dir=repo.working_dir)
    commit_id = _git.log(find_object=blob, max_count=1, all=True, oneline=True)
    return (
        commit_id
        if len(commit_id) > 0
        else _git.stash("list", "--oneline", "--max-count", "1", "--find-object", blob)
    )

get_conflict_blobs(repo)

Get the source files for conflicts.

Source code in databooks/git_utils.py

def get_conflict_blobs(repo: Repo) -> List[ConflictFile]:
    """Get the source files for conflicts."""
    unmerged_blobs = repo.index.unmerged_blobs()
    blobs = (
        UnmergedBlob(filename=Path(k), stage=dict(v))
        for k, v in unmerged_blobs.items()
        if 0 not in dict(v).keys()  # only get blobs that could not be merged
    )
    # Type checking: `repo.working_dir` is not None if repo is passed
    repo.working_dir = cast(str, repo.working_dir)
    return [
        ConflictFile(
            filename=repo.working_dir / blob.filename,
            first_log=blob2commit(blob=blob.stage[2], repo=repo),
            last_log=blob2commit(blob=blob.stage[3], repo=repo),
            first_contents=repo.git.show(blob.stage[2]),
            last_contents=repo.git.show(blob.stage[3]),
        )
        for blob in blobs
    ]

get_repo(path=PosixPath('/home/runner/work/databooks/databooks'))

Find git repo in current or parent directories.

Source code in databooks/git_utils.py

def get_repo(path: Path = Path.cwd()) -> Repo:
    """Find git repo in current or parent directories."""
    repo = Repo(path=path, search_parent_directories=True)
    logger.info(f"Repo found at: {repo.working_dir}")
    return repo