IPFS Module

IPFS client utilities for adding and retrieving JSON data.

class ipfs_dict_chain.IPFS.IPFSCache(ttl: int = 300)[source]

Bases: object

An in-memory cache for IPFS data with TTL-based expiration.

Entries are stored with an expiry timestamp. When an entry is retrieved, it is checked for expiration; expired entries are automatically removed. The TTL can be configured via the ttl parameter (default 300 seconds).

__init__(ttl: int = 300) → None[source]

Initialize an empty IPFS cache.

Parameters:: ttl (int) – Time-to-live for cache entries in seconds. Defaults to 300.

cleanup() → None[source]: Remove all expired entries from the cache.

clear() → None[source]: Remove all entries from the cache.

get(cid: str) → dict | None[source]

Retrieve data from the cache by its Content Identifier (CID).

Expired entries are removed and None is returned.

Parameters:: cid (str) – The Content Identifier (CID) of the data in the cache.
Returns:: The data retrieved from the cache, or None if not found or expired.
Return type:: Optional[Dict]

set(cid: str, data: dict) → None[source]

Store data in the cache with its Content Identifier (CID).

The entry will expire after the configured TTL.

Parameters:

cid (str) – The Content Identifier (CID) of the data.
data (Dict) – The data to be stored in the cache.

exception ipfs_dict_chain.IPFS.IPFSError[source]

Bases: Exception

Custom exception for IPFS-related errors.

async ipfs_dict_chain.IPFS._add_json(data: dict) → str[source]

Add JSON data to IPFS and return its Content Identifier (CID).

Parameters:: data (Dict) – The JSON data to be added to IPFS.
Returns:: The Content Identifier (CID) of the added JSON data.
Return type:: str

ipfs_dict_chain.IPFS._close_loop() → None[source]

Stop and close the background event loop at interpreter shutdown.

This function is registered with atexit.register() so that the loop and its thread are properly cleaned up when the Python interpreter exits. If the loop was never created or is already closed, no action is taken.

async ipfs_dict_chain.IPFS._get_json(cid: str) → dict[source]

Retrieve JSON data from IPFS by its Content Identifier (CID) and cache the result.

Parameters:: cid (str) – The Content Identifier (CID) of the JSON data in IPFS.
Returns:: The JSON data retrieved from IPFS.
Return type:: Dict

ipfs_dict_chain.IPFS._get_loop() → AbstractEventLoop[source]

Return the dedicated background event loop, starting its thread if needed.

The loop runs in its own daemon thread so that coroutines can be dispatched safely from any thread – including a thread that already has a running event loop – via asyncio.run_coroutine_threadsafe().

Returns:: The dedicated background asyncio event loop.
Return type:: asyncio.AbstractEventLoop

ipfs_dict_chain.IPFS._run(coro: Any) → Any[source]

Run a coroutine on the background loop from any thread and return its result.

This is safe to call from within a running event loop as well as from multiple threads concurrently, because the coroutine is scheduled on the dedicated background loop via asyncio.run_coroutine_threadsafe().

Parameters:: coro – The coroutine to run.
Returns:: The result of the coroutine.

async ipfs_dict_chain.IPFS._test_connection(maddr: ~multiaddr.multiaddr.Multiaddr = <Multiaddr /ip4/127.0.0.1/tcp/5001>) → bool[source]

Test the connection to the IPFS daemon using a read-only operation.

Parameters:: maddr (Multiaddr) – The multiaddr of the IPFS daemon to test. Defaults to the global multi_address.
Returns:: True if the connection is successful.
Return type:: bool
Raises:: IPFSError – If the connection test fails.

ipfs_dict_chain.IPFS.add_json(data: dict) → str[source]

Add JSON data to IPFS and return its Content Identifier (CID) using a synchronous wrapper.

Parameters:: data (Dict) – The JSON data to be added to IPFS.
Returns:: The Content Identifier (CID) of the added JSON data.
Return type:: str

ipfs_dict_chain.IPFS.connect(host: str, port: int) → None[source]

Connect to an IPFS daemon.

Parameters:

host (str) – The host of the IPFS daemon.
port (int) – The port of the IPFS daemon.

Raises:

IPFSError – If the connection to the IPFS daemon fails.

async ipfs_dict_chain.IPFS.get_file_content(cid: str) → str[source]

Retrieve the content of a file from IPFS by its Content Identifier (CID).

Parameters:: cid (str) – The Content Identifier (CID) of the file in IPFS.
Returns:: The content of the file.
Return type:: str

ipfs_dict_chain.IPFS.get_json(cid: str) → dict[source]

Retrieve JSON data from IPFS by its Content Identifier (CID) using a synchronous wrapper.

Parameters:: cid (str) – The Content Identifier (CID) of the JSON data in IPFS.
Returns:: The JSON data retrieved from IPFS.
Return type:: Dict