ipfs_utils.py

import ipfsapi, json
from io import StringIO, BytesIO


class IPFSAPI():
    def __init__(self):
        self.api = ipfsapi.connect('change-me', 5001)

    def files_exists(self, ipfs_path):
        try:
            self.api.files_stat(ipfs_path)
            return True
        except ipfsapi.exceptions.ErrorResponse:
            return False

    def files_dir_map(self, ipfs_path='/'):
        if not ipfs_path.endswith('/'):
            ipfs_path += '/'
        ipfs_path += '.dir_map'
        if not self.files_exists(ipfs_path):
            raise FileNotFoundError(ipfs_path)
        buffer = StringIO(self.api.files_read(ipfs_path).decode())
        buffer.seek(0)
        return json.load(buffer)

    def files_dir_add(self, ipfs_path, mapping):
        dir_map = None
        try:
            dir_map = self.files_dir_map(ipfs_path)
        except FileNotFoundError:
            dir_map = {}
        dir_map.update(mapping)
        buffer = BytesIO(json.dumps(dir_map).encode())
        buffer.seek(0)
        self.api.files_write(ipfs_path, buffer, create=True)

if __name__ == '__main__':
    api = IPFSAPI()
    api.files_dir_add('/', {'test': 'test'})