Skip to content

plugins.aea-cli-ipfs.aea_cli_ipfs.ipfs_client

Lightweight IPFS HTTP API client.

Replaces the ipfshttpclient package. Talks directly to the IPFS daemon's HTTP API (/api/v0/*) using urllib.

IPFSError Objects

class IPFSError(Exception)

Base IPFS client error.

CommunicationError Objects

class CommunicationError(IPFSError)

Could not communicate with the IPFS daemon.

__init__

def __init__(*args: Any, **kwargs: Any) -> None

Initialize, accepting optional original for compat.

TimeoutError Objects

class TimeoutError(CommunicationError)

Request to the IPFS daemon timed out.

StatusError Objects

class StatusError(CommunicationError)

IPFS daemon returned an unexpected HTTP status.

ErrorResponse Objects

class ErrorResponse(StatusError)

IPFS daemon returned a JSON error message.

_PinSection Objects

class _PinSection()

Pin management commands.

ls

def ls(type: str = "all") -> Dict

List pinned objects.

add

def add(cid: str, recursive: bool = True) -> Dict

Pin an object.

rm

def rm(cid: str, recursive: bool = True) -> Dict

Unpin an object.

_NameSection Objects

class _NameSection()

IPNS name commands.

publish

def publish(ipfs_path: str, **kwargs: Any) -> Dict

Publish an IPNS name.

_RepoSection Objects

class _RepoSection()

Repository commands.

gc

def gc(quiet: bool = False) -> List[Dict]

Run garbage collection.

IPFSHTTPClient Objects

class IPFSHTTPClient()

Lightweight IPFS HTTP API client.

Replaces ipfshttpclient.Client. Communicates with the IPFS daemon via its HTTP API at /api/v0/*.

__init__

def __init__(addr: str, base: str = "api/v0") -> None

Initialize client.

Arguments:

  • addr: multiaddr string (e.g. /dns/host/tcp/443/https).
  • base: API base path.

id

def id(peer: Optional[str] = None) -> Dict

Get node identity info.

add

def add(file_or_dir: str,
        pin: bool = True,
        recursive: bool = True,
        wrap_with_directory: bool = True) -> Union[Dict, List[Dict]]

Add file or directory to IPFS.

Arguments:

  • file_or_dir: path to file or directory.
  • pin: whether to pin the content.
  • recursive: whether to add recursively.
  • wrap_with_directory: whether to wrap with directory.

Returns:

list of dicts with 'Name' and 'Hash' keys.

add_bytes

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

Add bytes to IPFS.

Arguments:

  • data: bytes to add.
  • kwargs: additional keyword arguments.

Returns:

hash string.

get

def get(cid: str, target: str = ".") -> None

Download a file or directory from IPFS.

The IPFS /api/v0/get endpoint returns a tar archive.

Arguments:

  • cid: IPFS CID to download.
  • target: local directory to extract into.