Git LFS S3 transfer agent

lfsta.py

# [lfs]
# 	repositoryformatversion = 0
# 	standalonetransferagent = lfsta
# [lfs "customtransfer.lfsta"]
# 	path = python
# 	args = lfs.py

import sys
import json
from pathlib import Path
import requests
import os
import boto3

def json_message(payload):
    sys.stdout.write(json.dumps(payload) + "\n")
    sys.stdout.flush()

def error(msg: str):
    payload = {
        "error": {
            "code": 1,
            "message": msg,
        },
    }
    json_message(payload)
    sys.exit(1)

class Progress(object):
    def __init__(self, oid: str):
        self.oid = oid
        self.seen_so_far = 0

    def __call__(self, bytes_amount):
        self.seen_so_far += bytes_amount
        payload = {
            "event": "progress",
            "oid": self.oid,
            "bytesSoFar": self.seen_so_far,
            "bytesSinceLast": bytes_amount
        }
        json_message(payload)

BUCKET = "..."
REGION = "..."

S3_CLIENT = None

def init_s3_client():
    global S3_CLIENT
    session = boto3.Session()
    s3 = session.resource("s3",
        region_name=REGION,
        aws_access_key_id="...",
        aws_secret_access_key="...")
    S3_CLIENT = s3.Bucket(BUCKET)

def do_upload(oid: str, path: str):
    complete_payload = {
        "event": "complete",
        "oid": oid,
    }

    if S3_CLIENT == None:
        init_s3_client()

    try:
        S3_CLIENT.upload_file(path, f"lfs/{oid}", Callback=Progress(oid))
        json_message(complete_payload)
    except error as err:
        complete_payload["error"] = {
            "code": r.status_code,
            "message": "Error when uploading",
        }
        json_message(complete_payload)

def do_download(oid: str):
    path = ".git/lfs/tmp/" + oid

    complete_payload = {
        "event": "complete",
        "oid": oid,
        "path": path,
    }

    if S3_CLIENT == None:
        init_s3_client()

    try:
        S3_CLIENT.download_file(Key=f"lfs/{oid}", Filename=path, Callback=Progress(oid))
        json_message(complete_payload)
    except error as err:
        complete_payload["error"] = {
            "code": r.status_code,
            "message": "Error when uploading",
        }
        json_message(complete_payload)

if __name__ == "__main__":
    while True:
        payload = json.loads(sys.stdin.readline())
        if payload["event"] == "init" and payload["operation"] == "upload":
            json_message({})
        elif payload["event"] == "init" and payload["operation"] == "download":
            json_message({})
        elif payload["event"] == "upload":
            do_upload(payload.get("oid", None), payload.get("path", None))
        elif payload["event"] == "download":
            do_download(payload.get("oid", None))
        elif payload["event"] == "terminate":
            break
        else:
            error("Unknown initial operation")