in src/huggingface_hub/commands/lfs.py [0:0]
def run(self) -> None:
# Immediately after invoking a custom transfer process, git-lfs
# sends initiation data to the process over stdin.
# This tells the process useful information about the configuration.
init_msg = json.loads(sys.stdin.readline().strip())
if not (init_msg.get("event") == "init" and init_msg.get("operation") == "upload"):
write_msg({"error": {"code": 32, "message": "Wrong lfs init operation"}})
sys.exit(1)
# The transfer process should use the information it needs from the
# initiation structure, and also perform any one-off setup tasks it
# needs to do. It should then respond on stdout with a simple empty
# confirmation structure, as follows:
write_msg({})
# After the initiation exchange, git-lfs will send any number of
# transfer requests to the stdin of the transfer process, in a serial sequence.
while True:
msg = read_msg()
if msg is None:
# When all transfers have been processed, git-lfs will send
# a terminate event to the stdin of the transfer process.
# On receiving this message the transfer process should
# clean up and terminate. No response is expected.
sys.exit(0)
oid = msg["oid"]
filepath = msg["path"]
completion_url = msg["action"]["href"]
header = msg["action"]["header"]
chunk_size = int(header.pop("chunk_size"))
presigned_urls: List[str] = list(header.values())
# Send a "started" progress event to allow other workers to start.
# Otherwise they're delayed until first "progress" event is reported,
# i.e. after the first 5GB by default (!)
write_msg(
{
"event": "progress",
"oid": oid,
"bytesSoFar": 1,
"bytesSinceLast": 0,
}
)
parts = []
with open(filepath, "rb") as file:
for i, presigned_url in enumerate(presigned_urls):
with SliceFileObj(
file,
seek_from=i * chunk_size,
read_limit=chunk_size,
) as data:
r = get_session().put(presigned_url, data=data)
hf_raise_for_status(r)
parts.append(
{
"etag": r.headers.get("etag"),
"partNumber": i + 1,
}
)
# In order to support progress reporting while data is uploading / downloading,
# the transfer process should post messages to stdout
write_msg(
{
"event": "progress",
"oid": oid,
"bytesSoFar": (i + 1) * chunk_size,
"bytesSinceLast": chunk_size,
}
)
# Not precise but that's ok.
r = get_session().post(
completion_url,
json={
"oid": oid,
"parts": parts,
},
)
hf_raise_for_status(r)
write_msg({"event": "complete", "oid": oid})