dc-ops/lib/helper.py

import logging as log
import subprocess

import git


def run_subprocess(command: str, workdir: str) -> bool:
    """execute subprocess and print output

    Args:
        cmd (str): command
        cwd (str): working directory

    Returns:
        bool: False if subprocess fails
    """
    try:
        log.debug(subprocess.run(command, cwd=workdir, shell=True, text=True))
    except subprocess.CalledProcessError:
        return False

    return True


def update_git_repo(repo_path: str, ignore_git_status: bool) -> bool:
    """update repo and check for new commits

    Args:
        repo_path (str): path to git repo
        ignore_git_status (bool): continue even if no changes

    Returns:
        bool: False if any step fails
    """
    # create repo instance if it exists
    try:
        repo = git.Repo(repo_path)
    except git.exc.NoSuchPathError:
        log.error("directory not found")
        return False

    if not ignore_git_status:
        # try to fetch latest changes
        try:
            repo.git.fetch()
        except git.exc.GitCommandError as e:
            log.error(str(e))
            return False

        # check for new commits
        if repo.rev_parse("HEAD") == repo.rev_parse(f"origin/{repo.active_branch}"):
            log.info("no changes - skipping")
            return False

    # pull remote changes to local branch
    try:
        log.info(repo.git.pull())
    except git.exc.GitCommandError as e:
        log.error(str(e))
        return False

    return True