Script to clone or pull all of my gists, public and private, to a specified directory

.gitignore

.mypy_cache/

n8_gist_backup.py

#!/usr/bin/env python3
"""n8_gist_backup.py
Module to backup all private and public GitHub gists. Uses a GitHub Token
stored in Keyring which requires with scope `gists` to access one's private
gists. Clones all gists into `target_dir`.

Depends on [`internet_on`](https://gist.github.com/n8henrie/9330143). Requires
python >= 3.5 for `subprocess.run()`.

## Update 20160307:

- Now uses asyncio with the Python 3.5 syntax
    - About 21x faster for me
    - Old code:
        3.60user 3.50system 3:11.94elapsed 3%CPU (0avgtext+0avgdata
        96534528maxresident)k
        703inputs+421outputs (1187major+588760minor)pagefaults 0swaps
    - New code:
        3.57user 3.21system 0:08.87elapsed 76%CPU (0avgtext+0avgdata
        111230976maxresident)k
        176inputs+178outputs (1225major+623779minor)pagefaults 0swaps
"""

import asyncio
import logging
import os
import os.path
import sys
from asyncio import subprocess

import requests
from n8scripts.internet_on import internet_on

logging.basicConfig(
    level=logging.WARNING,
    format="%(asctime)s %(name)-12s %(levelname)-8s %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

logger_name = str(__file__) + " :: " + str(__name__)
logger = logging.getLogger(logger_name)


def gen_gist_ids(user, access_token):
    api_base = "https://api.github.com"
    endpoint = "/users/{}/gists".format(user)
    params = {"page": 1, "per_page": 100}
    headers = {"Authorization": f"token {access_token}"}
    while True:
        page = requests.get(
            api_base + endpoint, params=params, headers=headers
        )
        logger.debug(page.json())
        for gist in page.json():
            yield gist["id"]

        if 'rel="next"' not in page.headers["Link"]:
            break
        params["page"] += 1


async def get_keychain_credential(account, service):
    cmd = [
        "security",
        "find-generic-password",
        "-a",
        account,
        "-s",
        service,
        "-w",
    ]
    create_process = asyncio.create_subprocess_exec(
        *cmd, stdout=subprocess.PIPE
    )
    process = await create_process
    stdout, stderr = await process.communicate()
    return stdout.decode("utf8").strip()


async def clone_or_pull(gist_id, target_dir, sem):
    async with sem:
        logger.info("Working on {}".format(gist_id))
        ssh_url = "[email protected]:{}.git".format(gist_id)

        cmd = ["git", "-C", target_dir, "clone", ssh_url]
        logger.debug(" ".join(cmd))

        process = await subprocess.create_subprocess_exec(
            *cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
        )
        stdout_data, stderr_data = await process.communicate()

        logger.debug("Return code: {}".format(process.returncode))
        if process.returncode == 0:
            logger.info("cloned {}".format(gist_id))
        elif process.returncode == 128:
            logger.debug("Clone didn't work, trying to pull")

            # ?Already been cloned, try to pull
            logger.debug(" ".join(cmd))
            cmd = [
                "git",
                "-C",
                os.path.join(target_dir, gist_id),
                "pull",
                "origin",
                "master",
            ]
            process = await subprocess.create_subprocess_exec(
                *cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
            )
            stdout_data, stderr_data = await process.communicate()
            if process.returncode == 0:
                logger.info("Pulled gist {}".format(gist_id))
            else:
                logger.warning(
                    "Unable to pull gist {}. Return code: "
                    "{}".format(gist_id, process.returncode)
                )


async def main(target_dir):
    if not os.path.isdir(target_dir):
        os.mkdir(target_dir, mode=0o755)

    user = await get_keychain_credential("user", "n8_gist_backup.py")
    token = await get_keychain_credential("access_token", "n8_gist_backup.py")
    sem = asyncio.Semaphore(20)
    coroutines = [
        clone_or_pull(gist_id, target_dir, sem)
        for gist_id in gen_gist_ids(user=user, access_token=token)
    ]
    for coroutine in asyncio.as_completed(coroutines):
        await coroutine


if __name__ == "__main__":

    # Ensure internet connectivity
    if not internet_on():
        logging.error("No internet connectivity")
        sys.exit(1)

    target_dir = os.path.expanduser("~/gists")

    asyncio.run(main(target_dir=target_dir))