Skip to content

plugins.aea-cli-ipfs.aea_cli_ipfs.ipfs_utils

Ipfs utils for ipfs cli command.

resolve_addr

def resolve_addr(addr: str) -> Tuple[str, ...]

Multiaddr resolver.

Arguments:

  • addr: multiaddr string.

Returns:

http URL

addr_to_url

def addr_to_url(addr: str) -> str

Convert address to url.

is_remote_addr

def is_remote_addr(host: str) -> bool

Check if addr is remote or local.

IPFSDaemon Objects

class IPFSDaemon()

Set up the IPFS daemon.

Raises:

  • Exception: if IPFS is not installed.

__init__

def __init__(node_url: str = "http://127.0.0.1:5001", is_remote: bool = False)

Initialise IPFS daemon.

is_started_externally

def is_started_externally() -> bool

Check daemon was started externally.

is_started_internally

def is_started_internally() -> bool

Check daemon was started internally.

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 if it was started internally.

__enter__

def __enter__() -> None

Run the ipfs daemon.

__exit__

def __exit__(exc_type, exc_val, exc_tb) -> None

Terminate the ipfs daemon.

IPFSTool Objects

class IPFSTool()

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

__init__

def __init__(addr: Optional[str] = None, base: str = DEFAULT_IPFS_URI_BASE)

Init tool.

Arguments:

  • addr: multiaddr string for IPFS client.
  • base: API base for IPFS client.

addr

@property
def addr() -> str

Node address

is_a_package

def is_a_package(package_hash: str) -> bool

Checks if a package with package_hash is pinned or not

all_pins

def all_pins(recursive_only: bool = True) -> Set[str]

Returns a list of all pins.

add_bytes

def add_bytes(data: bytes, **kwargs) -> str

Add bytes data to ipfs.

Arguments:

  • data: bytes
  • kwargs: options passed to request library

Returns:

hash

add

def add(dir_path: str,
        pin: bool = True,
        recursive: bool = True,
        wrap_with_directory: 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
  • recursive: bool, publish dierctory recursively or not
  • wrap_with_directory: bool, wrap object with directory or not

Returns:

dir name published, hash, list of items processed

pin

def pin(hash_id: str) -> Dict

Pin content with hash_id

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.

remove_unpinned_files

def remove_unpinned_files() -> None

Remove dir added by it's hash.

download

def download(hash_id: str,
             target_dir: Union[str, Path],
             fix_path: bool = True,
             attempts: int = 5) -> str

Download dir by its hash.

Arguments:

  • hash_id: str. hash of file or package to download
  • target_dir: Union[str, Path]. directory to place downloaded
  • fix_path: bool. default True. on download don't wrap result in to hash_id directory.
  • attempts: int. default 5. How often to attempt the download.

Returns:

downloaded path

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.

check_ipfs_node_running

def check_ipfs_node_running() -> None

Check ipfs node running.