Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
import os
import sys
import requests
from git import Repo
from string import Template
# Get the command-line arguments
user_type = sys.argv[1]
user_id = sys.argv[2]
github_access_token = sys.argv[3]
url_type = sys.argv[4]
backup_folder_path = sys.argv[5]
# Defines a template for a GraphQL query to retrieve repositories
query = """
query {
$user_type(login:"$user_id") {
repositories(first:100,orderBy: {field: NAME, direction: ASC}$after) {
pageInfo {
startCursor
hasNextPage
endCursor
}
nodes {
name
$url_type
}
}
}
}
"""
query_template = Template(query)
# Creates a new query with the specified 'after' parameter value
def create_query(after=None):
after_str = f',after:"{after}"' if after else ''
return query_template.substitute(
user_type=user_type,
user_id=user_id,
url_type=url_type,
after=after_str)
# Makes the query request to the GitHub API
def send_query(query):
response = requests.post(
"https://api.github.com/graphql",
json={'query': query},
headers={"Authorization": "Token " + github_access_token})
if response.status_code == 200:
return response.json()
else:
raise Exception(f'{response.status_code}: {response.json()}')
# Fetches one page of the results and return the data as a tuple
def fetch_page(after=None):
q = create_query(after)
result = send_query(q)
data = result['data']
repos = data[user_type]['repositories']['nodes']
page_info = data[user_type]['repositories']['pageInfo']
after = page_info['endCursor']
has_next_page = page_info['hasNextPage']
return (has_next_page, after, repos)
# Gets a 'friendly' short SHA hash for a commit
def short_hash(ref):
return ref.object.hexsha[0:8]
after = None
# Keep fetching pages until has_next_page is false
while True:
(has_next_page, new_after, repos) = fetch_page(after)
for repo in repos:
name = repo['name']
url = repo[url_type]
mirror_path = os.path.join(
backup_folder_path,
f'{name}.git')
msg = None
print(name, end=': ')
if os.path.exists(mirror_path):
# We already have a mirror of this repo, so update it
repo = Repo(mirror_path)
existing_sha = short_hash(repo.head)
repo.remotes.origin.update()
updated_sha = short_hash(repo.head)
if existing_sha != updated_sha:
msg = f"Updated [{existing_sha}->{updated_sha}]"
else:
msg = f"No changes [{existing_sha}]"
else:
# This repo has been created since the last backup
Repo.clone_from(url, mirror_path, multi_options=['--mirror'])
repo = Repo(mirror_path)
msg = f"Cloned [{short_hash(repo.head)}]"
print(msg)
if has_next_page:
after = new_after
else:
break
print("All repositories backed up")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.