upload_ipfs_with_infura

upload_ipfs_with_infura.py

from typing import Dict, List
import requests
import json
import glob

CHUNK_SIZE = 128
URL = "https://ipfs.infura.io:5001/api/v0/add"
HEADERS = {
    "Authorization": "Basic YOUR-ENCODED-API-KEY"
}
EXT = "png"
FILE_TYPE = f"image/{EXT}"


def upload_ipfs(files: List[str], start: int, end: int) -> List[Dict]:
    files = [
        ('', (files[i].split("/")[-1], open(files[i], 'rb'), FILE_TYPE))
        for i in range(start, end)
    ]

    resp = requests.request(
        "POST", URL, headers=HEADERS, data={}, files=files)

    for f in files:
        f[1][1].close()

    txt = resp.text
    # infura doesn't return a proper json 'list'
    txt = " ".join(txt.split()).replace("} {", "},{")
    txt = f"[{txt}]"
    return json.loads(txt)


def main() -> None:
    result = []
    print("Preparing files to upload.")
    files = [f for f in glob.glob(f"./files/*.{EXT}")]
    l = len(files)
    print(f"Uploading {l} files.")

    for idx in range(0, l, CHUNK_SIZE):
        start = idx
        end = min(idx + CHUNK_SIZE, l)
        print(f"Uploading chunk {start} to {end}")
        result.extend(upload_ipfs(files, start, end))

    print(f"Writing output.")
    with open('cid.json', 'w') as outfile:
        json.dump(result, outfile)
    print("Done.")


if __name__ == '__main__':
    main()