大きなサイズやバイナリファイルをGitで上手く管理する方法について考える

しむどん： 2024-12-02

Gitはバイナリファイルを管理する事が苦手で、リポジトリのサイズがどんどん増えてしまう。これを回避する機構としてGit LFSがある。Git LFSは、登録したデータとポインタファイルを置き換える事により、サイズの肥大化という問題を解消している。GitHubもGit LFSをサポートしている。しかしこのGit LFSにはいくつか問題がある。GitHubでは料金がかかるし、CIによってはサポートしていなかったり、逆に有無を言わさずフェッチしてしまったりと、状況の足並みが揃わず対策がない事さえある。GitHubにホスティングしているリポジトリでは、GitHubが用意しているLFSサーバーを使う事になるが、料金の制限により、CIが通らない事が発生した。またGitHubのLFSサーバーの使用料制限にはバグがあり、1ヶ月経過してもリセットされないという事象が発生した(これはサポートに連絡して解決)。GitHubの提供していないLFSサーバーを使う事もできる。Git LFSのCustom Transfer Agentを使えば、S3やGoogle Driveやローカルの別ディレクトリなど、ファイルを配置したい所に置く事ができる。実際にやってみたらできた。

それではGit LFSでいいだろうとも思えるが、ちょっと立ち止まる事にした。そもそも僕はバイナリや大きなファイルをリポジトリ外部に登録して、Gitにデータ自体を管理させないようにしたかっただけだ。やりたい事はシンプルだ。しかし僕にとっては Git LFS や Git LFSのCustom Transfer Agent は、やりたい事に対して少し複雑すぎるように思えた。そこで Git LFS のポインタファイルの仕組みなどはそのままに、データの保存と取得だけを供えたスクリプトを、自分で実装する事にした。転送自体は rclone を使う事にし、 Git LFS 自体には依存しないようにした。

これを使うメリットは何をしようとしているのか、どこに何があるのか、なにが良くて何が出来て何がダメなのか、すぐに理解できる事だ。これで対応が難しくなったら、その時始めてもう少し複雑なものの導入を検討しようと思う。スクリプトを貼っておく。

#! /usr/bin/env python3
import configparser
import hashlib
import os
import re
import shutil
import sys
import tempfile

__author__ = "TakesxiSximada"
__version__ = "2"

pointer_file_format = """# LFS POINTER
version https://git-lfs.github.com/spec/v1
oid {hash_type}:{oid}
size {size}
"""

def traverse_config_file(path):
    cur_dir = os.path.abspath(path)
    for i in range(100):
        conf_path = os.path.join(cur_dir, "storebin.conf")
        if os.path.exists(conf_path):
            return conf_path
        cur_dir = os.path.dirname(cur_dir)

def parse_config(path) -> configparser.ConfigParser:
    conf = configparser.ConfigParser()
    conf.read(path)
    return conf

sub_command = sys.argv[1]
if "put" == sub_command:
    target_file = sys.argv[2]
    pointer_file = f"{target_file}.storebin"

    config_file = traverse_config_file(target_file)
    if not config_file:
        print("no config file: storebin.conf")
        sys.exit(1)

    conf = parse_config(config_file)
    rclone_config_label = conf["storebin"]["rclone_config_label"]

    if os.path.exists(pointer_file):
        print("already exists: ", pointer_file)
        sys.exit(1)

    with open(target_file, "rb") as fp:
        hash_value_obj = hashlib.sha256(fp.read())

    oid = hash_value_obj.hexdigest()
    file_size = os.path.getsize(target_file)
    object_storage_key = f"{oid[:2]}/{oid[2:4]}/{oid}"
    remote_location = f"{rclone_config_label}:{object_storage_key}"
    pointer_file_content = pointer_file_format.format(
        hash_type="sha256",
        oid=oid,
        size=file_size,
    )

    tmp_data = tempfile.mkstemp()
    tmp_file = tmp_data[1]
    rc = os.system(f"rclone copyto --verbose {remote_location} {tmp_file}")
    if rc == 0:
        print("Already exist remote file: %s", remote_location)
        sys.exit(1)
        
    cmd = f"rclone copyto --verbose {target_file} {remote_location}"
    print("> ", cmd)

    rc = os.system(cmd)
    if rc == 0:
        with open(pointer_file, "w+") as fp:
            fp.write(pointer_file_content)
    else:
        print(f"error: target_file={target_file} key={object_storage_key}: {cmd}")

elif "get" == sub_command:
    pointer_file = sys.argv[2]
    target_file = pointer_file.rstrip(".storebin")

    config_file = traverse_config_file(target_file)
    if not config_file:
        print("no config file: storebin.conf")
        sys.exit(1)

    conf = parse_config(config_file)
    rclone_config_label = conf["storebin"]["rclone_config_label"]

    if os.path.exists(target_file):
        print("already exists: ", target_file)
        sys.exit(1)
    with open(pointer_file) as fp:
        m = re.search(r"^oid sha256:(?P<oid>[0-9a-f]+)$", fp.read(), re.M)

    if not m:
        print("bad pointer file: ", target_file)
        sys.exit(1)

    oid = m.group("oid")
    if not oid:
        print("bad oid: ", oid)
        sys.exit(1)

    object_storage_key = f"{oid[:2]}/{oid[2:4]}/{oid}"
    remote_location = f"{rclone_config_label}:{object_storage_key}"

    object_storage_key = f"{oid[:2]}/{oid[2:4]}/{oid}"
    cmd = f"rclone copyto --immutable --verbose {remote_location} {target_file}"
    print("> ", cmd)
    rc = os.system(cmd)
    sys.exit(rc)

elif "delete" == sub_command:
    pointer_file = sys.argv[2]
    target_file = pointer_file.rstrip(".storebin")

    config_file = traverse_config_file(target_file)
    if not config_file:
        print("no config file: storebin.conf")
        sys.exit(1)

    conf = parse_config(config_file)
    rclone_config_label = conf["storebin"]["rclone_config_label"]
    with open(pointer_file) as fp:
        m = re.search(r"^oid sha256:(?P<oid>[0-9a-f]+)$", fp.read(), re.M)

    if not m:
        print("bad pointer file: ", target_file)
        sys.exit(1)

    oid = m.group("oid")
    if not oid:
        print("bad oid: ", oid)
        sys.exit(1)

    object_storage_key = f"{oid[:2]}/{oid[2:4]}/{oid}"
    cmd = f"rclone deletefile --verbose {rclone_config_label}:{object_storage_key}"
    print("> ", cmd)
    os.system(cmd)
    shutil.move(pointer_file,  pointer_file + ".deleted")

else:
    print("not support subcommand")
    sys.exit(1)