Fast repository status for repos in git folder, Python Rewrite.

git-all.py

#!/usr/bin/env python3

from plumbum import local, cli, FG, BG, TF, ProcessExecutionError, colors
from plumbum.cmd import tput, git
from contextlib import contextmanager
from functools import partial

# This can be hard coded to a repo location, or cwd, etc.
# REPOLOC = local.path(__file__) / '..'
REPOLOC = local.env.home / 'git' 
valid_repos = [d / '../..' for d in REPOLOC // '*/.git/config']

printf = partial(print, end='', flush=True)

def git_on_all():
    'Loop over all the repos, inside the loop you are in the directory of the repo.'
    for n,repo in enumerate(valid_repos):
        with local.cwd(repo):
            with (colors[1:7][n%6] & colors.bold):
                yield repo.basename

                
def fetch():
    'Fetch on all repos, with message printed for failures.'
    bg = []
    for repo in git_on_all():
        bg.append((repo, git['fetch','-q'] & BG))
    printf('Waiting for the repos to report: ')
    for repo, fut in bg:
        try:
            fut.wait()
        except ProcessExecutionError:
            print(repo, 'Failed to fetch, not valid?')
    print('done.')

    
# Git commands, collected and named for clarity
            
def unstaged_changes():
    'True if there are unstaged changes in working tree.'
    return not git['diff-index', '--cached', '--quiet', 'HEAD', '--ignore-submodules', '--'] & TF
    
def print_unstaged_changes():
    printf(~colors.bold & git('diff-files', '--name-status', '-r', '--ignore-submodules', '--'))
    
def uncommited_changes_in_index():
    return not git['diff-index', '--cached', '--quiet', 'HEAD', '--ignore-submodules', '--'] & TF
    
def print_uncommited_changes_in_index():
    printf(~colors.bold & git('diff-index', '--cached', '--name-status', '-r', '--ignore-submodules', 'HEAD', '--'))
    
def unpushed_changes():
    'Note that this is a list of unpushed changes compared to empty, not as programatic as above solutions'
    return git('rev-list', '@{u}..') != ''

def print_unpushed_changes():
    
    printf(~colors.bold & git('log', '@{u}..', '--oneline'))
    
def unmerged_changes():
    return git('rev-list', '..@{u}') != ''
    
def print_unmerged_changes():
    printf(~colors.bold & git('log', '..@{u}', '--oneline'))
    
# Actual Application and commands
    
    
class GitAll(cli.Application):
    """Multiple repository checker"""
    VERSION = "1.0"
    
    
@GitAll.subcommand("status")
class Status(cli.Application):
    'Checks all the git repos in git folder'
    quick = cli.Flag("-q", help = "Quick status (no fetch)")
    
    def main(self):
        if not self.quick:
            fetch()
        for repo in git_on_all():
            printf("Checking {}: ".format(repo))
            changed = False
            if unstaged_changes():
                print("There are unstaged changes:")
                print_unstaged_changes()
                changed = True
            if uncommited_changes_in_index():
                print("There are uncommented changes in the index:")
                print_uncommited_changes_in_index()
                changed = True
            if unpushed_changes():
                print("There are unpushed changes:")
                print_unpushed_changes()
                changed = True
            if unmerged_changes():
                print("There are unmerged changes:")
                print_unmerged_changes()
                changed = True
            if not changed:
                print("No changes")

        
@GitAll.subcommand("pull")
class Pull(cli.Application):
    'Pulls all repos in the folder, not threaded.'
    def main(self):
        for repo in git_on_all():
            git['pull'] & FG
        
        
@GitAll.subcommand("fetch")
class Fetch(cli.Application):
    'Fetches all repos in the folder'
    quiet = cli.Flag("-q", help = "Quiet fetch")
    def main(self):
        fetch()
        if not self.quiet:
            for repo in git_on_all():
                printf("Fetched {}: ".format(repo))
                if unmerged_changes():
                    print("There are unmerged changes:")
                    print_unmerged_changes()
                else:
                    print("Up to date.")
                
                
                
@GitAll.subcommand("merge")
class Merge(cli.Application):
    'Merges all repos in the folder (origin/master)'
    def main(self):
        for repo in git_on_all():
            printf("Merging {}: ".format(repo))
            if unmerged_changes():
                git['merge','origin/master'] & FG
            else:
                print("Merge not needed.")


if __name__ == "__main__":
    GitAll.run()