Example of async usage in Python by pulling and checking the status of local repos.

async_example.py

""" Pull Git repos and check the status of them using Python's async.

Simple test using async subprocess calls and async itself.

Some simple tests showed this took a 3rd the time of the synchronous
counterpart.
"""
import asyncio
import time
from pathlib import Path


# Repos
REPOS = [
    Path.home() / 'vimfiles',
]

DEV_PATH = Path.home() / 'Development'
REPOS.extend(
    [
        repo
        for repo in DEV_PATH.glob('*')
        if DEV_PATH.is_dir() and not DEV_PATH.is_symlink()
    ]
)

# Bare Repos
DOTFILES_GIT_DIRECTORY = Path.home() / '.dotfiles'
DOTFILES_WORK_DIR = Path.home()

BARE_REPOS = [(DOTFILES_GIT_DIRECTORY, DOTFILES_WORK_DIR)]


async def async_run_command(command):
    """ Run an async command and return stdout and stderr. """
    process = await asyncio.create_subprocess_shell(
        command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await process.communicate()
    stdout, stderr = stdout.decode(), stderr.decode()

    return stdout, stderr


async def async_git_pull(repo_path=None, git_dir=None, work_tree=None):
    """`git pull` on a directory, or git directory and work tree for bare
    repos."""
    if repo_path:
        command = ['git', '-C', fr'"{repo_path}"', 'pull']
        command = ' '.join(command)
    else:
        command = [
            'git',
            f'--git-dir="{git_dir}"',
            f'--work-tree="{work_tree}"',
            'pull',
        ]
        command = ' '.join(command)
    stdout, stderr = await async_run_command(command)

    return stdout, stderr


async def async_git_status(repo_path=None, git_dir=None, work_tree=None):
    """`git status` on a directory, or git directory and work tree for bare
    repos."""

    if repo_path:
        command = ['git', '-C', fr'"{ repo_path }"', 'status']
        command = " ".join(command)
    else:
        command = [
            'git',
            f'--git-dir="{git_dir}"',
            f'--work-tree="{work_tree}"',
            'status',
        ]
        command = ' '.join(command)
    stdout, stderr = await async_run_command(command)

    return stdout, stderr


async def chain(repo_path=None, git_dir=None, work_tree=None):
    """ Chain multiple async parts together. """
    if repo_path:
        name = Path(repo_path).name
    else:
        name = Path(git_dir).name
    print(f'Starting {name}...')
    stdout, stderr = await async_git_pull(repo_path, git_dir, work_tree)
    stdout, stderr = await async_git_status(repo_path, git_dir, work_tree)
    print(f'Ending {name}...')


async def async_main():
    """Main function, but async.

    Chain pulls and status checks to each repo/bare repo, then run them.
    """
    await asyncio.gather(
        *(
            [chain(repo) for repo in REPOS]
            + [
                chain(git_dir=bare_repo[0], work_tree=bare_repo[1])
                for bare_repo in BARE_REPOS
            ]
        )
    )


if __name__ == '__main__':

    start = time.perf_counter()

    asyncio.run(async_main())

    elapsed = time.perf_counter() - start
    print()
    print(f'Executed in {elapsed:0.2f} seconds.')