1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
|
import os
import argparse
from pathlib import Path
import datetime
from functools import wraps
from sh.contrib import git
import boto3
s3 = boto3.resource('s3')
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest='cmd', help='sub-command help')
parser_backup= subparsers.add_parser('backup', help='backup repositories')
parser_backup.add_argument('repositories', help='directory with repos')
# create the parser for the "b" command
parser_b = subparsers.add_parser('b', help='b help')
parser_b.add_argument('--baz', choices='XYZ', help='baz help')
args = parser.parse_args()
def logit(func):
@wraps(func)
def with_logging(*args, **kwargs):
print(func.__name__ + " was called")
return func(*args, **kwargs)
return with_logging
def cloud_object(bundle, prefix, bucket):
obj = s3.Object(bucket, f'{prefix}/{bundle.name}')
obj.upload_file(str(bundle))
return obj
def backup_repo(repo, repo_name):
# new repo or backup
ts = datetime.datetime.utcnow().timestamp()
bundle_path = Path(f'{repo_name}.{ts}.bundle')
repo.bundle('create', bundle_path, '--all')
return bundle_path
def tag_checkpoint(repo):
last_hash = repo('rev-list', '-n', 1, '--all').strip()
repo.tag('-f', 'CHECKPOINT', last_hash)
def requires_backup(repo):
last_hash = repo.rev_list('rev-list', '-n', 1, '--all').strip()
# empty repo
if not last_hash:
return False
try:
checkpoint = repo.rev_list('rev-list', '-n', 1, 'CHECKPOINT').strip()
# no checkpoint exists
except sh.ErrorReturnCode_1:
return True
return True ^ (last_hash == checkpoint)
def run_restore(bundle):
pass
def run_backup(base):
repo_base_path = Path(base)
for repo_path in repo_base_path.glob('*.git'):
repo = git.bake(f'--git-dir={repo_path}/')
if requires_backup:
bundle_path = backup_repo(repo, repo_path.name.split('.')[0])
obj = cloud_object(bundle_path, '2', 'privategit')
tag_checkpoint(repo)
if args.cmd == 'backup':
run_backup(args.repositories)
elif args.cmd == 'restore':
run_restore(args.bundle)
|
