plugins.aea-cli-ipfs.aea_cli_ipfs.ipfs_utils

Ipfs utils for ipfs cli command.

IPFSDaemon Objects

class IPFSDaemon()

Set up the IPFS daemon.

Raises:

• Exception: if IPFS is not installed.

__init__

def __init__() -> None

Initialise IPFS daemon.

is_started

def is_started() -> bool

Check daemon was started.

start

def start() -> None

Run the ipfs daemon.

stop

def stop() -> None

Terminate the ipfs daemon.

BaseIPFSToolException Objects

class BaseIPFSToolException(Exception)

Base ipfs tool exception.

RemoveError Objects

class RemoveError(BaseIPFSToolException)

Exception on remove.

PublishError Objects

class PublishError(BaseIPFSToolException)

Exception on publish.

NodeError Objects

class NodeError(BaseIPFSToolException)

Exception for node connection check.

DownloadError Objects

class DownloadError(BaseIPFSToolException)

Exception on download failed.

IPFSTool Objects

class IPFSTool()

IPFS tool to add, publish, remove, download directories.

__init__

def __init__(client_options: Optional[Dict] = None)

Init tool.

Arguments:

• client_options: dict, options for ipfshttpclient instance.

add

def add(dir_path: str, pin: bool = True) -> Tuple[str, str, List]

Add directory to ipfs.

It wraps into directory.

Arguments:

• dir_path: str, path to dir to publish
• pin: bool, pin object or not

Returns:

dir name published, hash, list of items processed

remove

def remove(hash_id: str) -> Dict

Remove dir added by it's hash.

Arguments:

• hash_id: str. hash of dir to remove

Returns:

dict with unlinked items.

download

def download(hash_id: str, target_dir: str, fix_path: bool = True) -> None

Download dir by it's hash.

Arguments:

• hash_id: str. hash of file to download
• target_dir: str. directory to place downloaded
• fix_path: bool. default True. on download don't wrap result in to hash_id directory.

publish

def publish(hash_id: str) -> Dict

Publish directory by it's hash id.

Arguments:

• hash_id: hash of the directory to publish.

Returns:

dict of names it was publish for.

chec_ipfs_node_running

def chec_ipfs_node_running() -> None

Check ipfs node running.