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]

Close the global asyncio event loop at interpreter shutdown.

This function is registered with atexit.register() so that the event loop is properly cleaned up when the Python interpreter exits. If the loop is already closed or was never created, 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 global asyncio event loop, creating it if necessary.

If the global event loop is None or has been closed, a new event loop is created and set as the current event loop for the thread.

Returns:: The global asyncio event loop.
Return type:: asyncio.AbstractEventLoop

async ipfs_dict_chain.IPFS._test_connection() → bool[source]

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

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