Python第三方庫IPFS-API使用詳解：構建去中心化應用的完整指南

星際文件系統（IPFS，InterPlanetary File System）是一種革命性的點對點超媒體協議，旨在創建持久且分布式的網絡傳輸方式。與傳統HTTP協議基于位置尋址不同，IPFS使用內容尋址來唯一標識文件，這意味著文件不再通過服務器位置而是通過其內容哈希來訪問。

根據2023年數據，IPFS網絡已經存儲了超過150億個唯一內容，每天處理超過數千萬次的API請求。Python作為數據科學和Web開發的主流語言，通過ipfs-api庫提供了與IPFS網絡交互的完整能力。

本文將深入探討ipfs-api庫的使用方法，從基礎概念到高級應用，幫助開發者充分利用IPFS構建去中心化應用。

1 IPFS核心概念與架構

1.1 內容尋址 vs 位置尋址

傳統Web使用位置尋址：

https://example.com/path/to/file.txt

IPFS使用內容尋址：

/ipfs/QmXoypizjW3WknFiJnKLwHCnL72vedxjQkDDP1mXWo6uco

內容標識符(CID)的計算基于文件內容的加密哈希，確保了：

· 唯一性：相同內容總是產生相同CID
· 完整性：任何內容修改都會改變CID
· 永久性：內容不受位置變化影響

1.2 IPFS節點架構

1.3 IPFS HTTP API

IPFS守護進程提供HTTP API接口，默認在端口5001上監聽。ipfs-api庫本質上是一個與這個HTTP API交互的Python客戶端。

2 環境安裝與配置

2.1 安裝IPFS節點

首先需要安裝IPFS命令行工具：

# 在Ubuntu/Debian上安裝
wget https://dist.ipfs.tech/kubo/v0.22.0/kubo_v0.22.0_linux-amd64.tar.gz
tar -xvzf kubo_v0.22.0_linux-amd64.tar.gz
cd kubo
sudo ./install.sh# 初始化IPFS節點
ipfs init# 啟動IPFS守護進程
ipfs daemon

2.2 安裝Python ipfs-api庫

# 安裝ipfs-api庫
pip install ipfs-api# 或者安裝開發版本
pip install git+https://github.com/ipfs/py-ipfs-api.git

2.3 驗證安裝

import ipfsapitry:# 連接到本地IPFS節點api = ipfsapi.connect('127.0.0.1', 5001)print("IPFS節點連接成功!")print(f"節點ID: {api.id()['ID']}")print(f"IPFS版本: {api.version()['Version']}")
except Exception as e:print(f"連接失敗: {e}")

3 基礎用法詳解

3.1 連接IPFS節點

import ipfsapi
import json
from typing import Dict, Anyclass IPFSClient:"""IPFS客戶端封裝類"""def __init__(self, host: str = '127.0.0.1', port: int = 5001):"""初始化IPFS客戶端Args:host: IPFS守護進程主機地址port: IPFS API端口號"""self.host = hostself.port = portself.client = Nonedef connect(self, timeout: int = 30) -> bool:"""連接到IPFS節點Args:timeout: 連接超時時間(秒)Returns:bool: 連接是否成功"""try:self.client = ipfsapi.Client(host=self.host, port=self.port)# 測試連接self.client.id()print(f"成功連接到IPFS節點 {self.host}:{self.port}")return Trueexcept ipfsapi.exceptions.ConnectionError as e:print(f"連接錯誤: {e}")return Falseexcept ipfsapi.exceptions.TimeoutError as e:print(f"連接超時: {e}")return Falseexcept Exception as e:print(f"未知錯誤: {e}")return Falsedef get_node_info(self) -> Dict[str, Any]:"""獲取節點信息Returns:Dict: 節點信息字典"""if not self.client:raise ConnectionError("未連接到IPFS節點")return self.client.id()

3.2 文件操作

3.2.1 添加文件到IPFS

class IPFSFileOperations(IPFSClient):"""IPFS文件操作類"""def add_file(self, file_path: str, pin: bool = True) -> Dict[str, Any]:"""添加文件到IPFSArgs:file_path: 文件路徑pin: 是否固定文件Returns:Dict: 文件添加結果"""try:result = self.client.add(file_path, pin=pin)print(f"文件添加成功: {result['Hash']}")return resultexcept ipfsapi.exceptions.Error as e:print(f"文件添加失敗: {e}")raisedef add_bytes_data(self, data: bytes, file_name: str = "data.bin") -> Dict[str, Any]:"""添加字節數據到IPFSArgs:data: 字節數據file_name: 文件名Returns:Dict: 添加結果"""try:result = self.client.add_bytes(data)print(f"字節數據添加成功: {result}")return {'Hash': result, 'Name': file_name}except Exception as e:print(f"字節數據添加失敗: {e}")raisedef add_json_data(self, data: Dict[str, Any], file_name: str = "data.json") -> Dict[str, Any]:"""添加JSON數據到IPFSArgs:data: JSON數據file_name: 文件名Returns:Dict: 添加結果"""try:# 將JSON轉換為字符串json_str = json.dumps(data, ensure_ascii=False)# 添加到IPFSresult = self.client.add_str(json_str)print(f"JSON數據添加成功: {result}")return {'Hash': result, 'Name': file_name}except Exception as e:print(f"JSON數據添加失敗: {e}")raisedef add_directory(self, directory_path: str, recursive: bool = True) -> List[Dict[str, Any]]:"""添加目錄到IPFSArgs:directory_path: 目錄路徑recursive: 是否遞歸添加子目錄Returns:List: 添加結果列表"""try:result = self.client.add(directory_path, recursive=recursive, pin=True)print(f"目錄添加成功，共添加 {len(result)} 個文件")return resultexcept Exception as e:print(f"目錄添加失敗: {e}")raise

3.2.2 從IPFS獲取文件

class IPFSFileRetrieval(IPFSFileOperations):"""IPFS文件檢索類"""def get_file(self, cid: str, output_path: str = None) -> bytes:"""根據CID獲取文件內容Args:cid: 內容標識符output_path: 輸出文件路徑(可選)Returns:bytes: 文件內容"""try:content = self.client.cat(cid)if output_path:with open(output_path, 'wb') as f:f.write(content)print(f"文件已保存到: {output_path}")return contentexcept ipfsapi.exceptions.Error as e:print(f"文件獲取失敗: {e}")raisedef get_json_data(self, cid: str) -> Dict[str, Any]:"""獲取JSON數據Args:cid: JSON數據的CIDReturns:Dict: JSON數據"""try:content = self.client.cat(cid)json_data = json.loads(content.decode('utf-8'))return json_dataexcept Exception as e:print(f"JSON數據獲取失敗: {e}")raisedef download_file(self, cid: str, output_path: str) -> bool:"""下載文件到本地路徑Args:cid: 內容標識符output_path: 輸出路徑Returns:bool: 下載是否成功"""try:self.client.get(cid, output_path)print(f"文件下載成功: {output_path}")return Trueexcept Exception as e:print(f"文件下載失敗: {e}")return Falsedef list_directory(self, cid: str) -> List[Dict[str, Any]]:"""列出目錄內容Args:cid: 目錄的CIDReturns:List: 目錄內容列表"""try:contents = self.client.ls(cid)return contents['Objects'][0]['Links']except Exception as e:print(f"目錄列表獲取失敗: {e}")raise

3.3 目錄操作高級示例

class IPFSDirectoryManager(IPFSFileRetrieval):"""IPFS目錄管理類"""def create_virtual_directory(self, files_data: Dict[str, bytes]) -> str:"""創建虛擬目錄并添加多個文件Args:files_data: 文件名到內容的映射Returns:str: 目錄CID"""try:# 使用MFS(可變文件系統)創建目錄dir_path = "/virtual_dir_" + str(int(time.time()))# 創建目錄self.client.files_mkdir(dir_path)# 添加文件到目錄for file_name, content in files_data.items():file_path = f"{dir_path}/{file_name}"if isinstance(content, str):content = content.encode('utf-8')# 寫入文件self.client.files_write(file_path, io.BytesIO(content), create=True, truncate=True)# 獲取目錄CIDdir_cid = self.client.files_stat(dir_path)['Hash']print(f"虛擬目錄創建成功: {dir_cid}")return dir_cidexcept Exception as e:print(f"虛擬目錄創建失敗: {e}")raisedef download_entire_directory(self, cid: str, output_dir: str) -> bool:"""下載整個目錄結構Args:cid: 目錄CIDoutput_dir: 輸出目錄Returns:bool: 下載是否成功"""try:# 確保輸出目錄存在os.makedirs(output_dir, exist_ok=True)# 獲取目錄內容contents = self.list_directory(cid)for item in contents:item_path = os.path.join(output_dir, item['Name'])if item['Type'] == 1:  # 目錄os.makedirs(item_path, exist_ok=True)# 遞歸下載子目錄self.download_entire_directory(item['Hash'], item_path)else:  # 文件self.download_file(item['Hash'], item_path)print(f"目錄下載完成: {output_dir}")return Trueexcept Exception as e:print(f"目錄下載失敗: {e}")return False

4 高級功能與技巧

4.1 Pin管理機制

Pin是IPFS中確保內容持久化的關鍵機制：

class IPFSPinManager(IPFSDirectoryManager):"""IPFS Pin管理類"""def pin_content(self, cid: str) -> bool:"""固定內容Args:cid: 內容標識符Returns:bool: 固定是否成功"""try:self.client.pin_add(cid)print(f"內容已固定: {cid}")return Trueexcept Exception as e:print(f"內容固定失敗: {e}")return Falsedef unpin_content(self, cid: str) -> bool:"""取消固定內容Args:cid: 內容標識符Returns:bool: 取消固定是否成功"""try:self.client.pin_rm(cid)print(f"內容已取消固定: {cid}")return Trueexcept Exception as e:print(f"取消固定失敗: {e}")return Falsedef list_pinned_content(self) -> List[Dict[str, Any]]:"""列出所有固定的內容Returns:List: 固定的內容列表"""try:pinned = self.client.pin_ls()return pinnedexcept Exception as e:print(f"獲取固定列表失敗: {e}")return []def check_pin_status(self, cid: str) -> Dict[str, Any]:"""檢查內容的固定狀態Args:cid: 內容標識符Returns:Dict: 固定狀態信息"""try:# 獲取所有固定內容pinned = self.client.pin_ls()if cid in pinned:return {'pinned': True,'type': pinned[cid]['Type']}else:return {'pinned': False}except Exception as e:print(f"檢查固定狀態失敗: {e}")return {'pinned': False, 'error': str(e)}

4.2 IPNS（星際命名系統）

IPNS允許為可變內容創建固定引用：

class IPNSManager(IPFSPinManager):"""IPNS管理類"""def publish_to_ipns(self, cid: str, key: str = 'self') -> Dict[str, Any]:"""發布內容到IPNSArgs:cid: 內容標識符key: IPNS密鑰名稱Returns:Dict: 發布結果"""try:result = self.client.name_publish(cid, key=key)print(f"內容已發布到IPNS: {result['Name']}")return resultexcept Exception as e:print(f"IPNS發布失敗: {e}")raisedef resolve_ipns(self, ipns_name: str) -> str:"""解析IPNS名稱到CIDArgs:ipns_name: IPNS名稱Returns:str: 解析得到的CID"""try:result = self.client.name_resolve(ipns_name)return result['Path'].split('/')[-1]  # 提取CIDexcept Exception as e:print(f"IPNS解析失敗: {e}")raisedef create_ipns_key(self, key_name: str) -> Dict[str, Any]:"""創建新的IPNS密鑰Args:key_name: 密鑰名稱Returns:Dict: 密鑰信息"""try:result = self.client.key_gen(key_name, 'rsa')print(f"IPNS密鑰創建成功: {result['Name']}")return resultexcept Exception as e:print(f"IPNS密鑰創建失敗: {e}")raisedef list_ipns_keys(self) -> List[Dict[str, Any]]:"""列出所有IPNS密鑰Returns:List: 密鑰列表"""try:result = self.client.key_list()return result['Keys']except Exception as e:print(f"獲取IPNS密鑰列表失敗: {e}")return []

4.3 PubSub（發布-訂閱系統）

IPFS PubSub支持實時消息傳遞：

import threading
import timeclass IPFSPubSubManager(IPNSManager):"""IPFS發布-訂閱管理類"""def __init__(self, host: str = '127.0.0.1', port: int = 5001):super().__init__(host, port)self.subscriptions = {}def publish_message(self, topic: str, message: str) -> bool:"""發布消息到主題Args:topic: 主題名稱message: 消息內容Returns:bool: 發布是否成功"""try:self.client.pubsub_pub(topic, message)print(f"消息已發布到主題 '{topic}': {message}")return Trueexcept Exception as e:print(f"消息發布失敗: {e}")return Falsedef subscribe_to_topic(self, topic: str, callback: callable) -> threading.Thread:"""訂閱主題Args:topic: 主題名稱callback: 消息處理回調函數Returns:threading.Thread: 訂閱線程"""def subscription_worker():try:# 創建訂閱sub = self.client.pubsub_sub(topic)self.subscriptions[topic] = subprint(f"已訂閱主題: {topic}")# 處理消息for message in sub:try:# 解析消息msg_data = json.loads(message['data'].decode('utf-8'))callback(msg_data)except Exception as e:print(f"消息處理錯誤: {e}")except Exception as e:print(f"訂閱錯誤: {e}")# 啟動訂閱線程thread = threading.Thread(target=subscription_worker, daemon=True)thread.start()return threaddef unsubscribe_from_topic(self, topic: str) -> bool:"""取消訂閱主題Args:topic: 主題名稱Returns:bool: 取消訂閱是否成功"""try:if topic in self.subscriptions:self.subscriptions[topic].close()del self.subscriptions[topic]print(f"已取消訂閱主題: {topic}")return Truereturn Falseexcept Exception as e:print(f"取消訂閱失敗: {e}")return Falsedef list_peers(self, topic: str = None) -> List[str]:"""列出PubSub對等節點Args:topic: 主題名稱(可選)Returns:List: 對等節點列表"""try:if topic:peers = self.client.pubsub_peers(topic)else:peers = self.client.pubsub_peers()return peersexcept Exception as e:print(f"獲取對等節點列表失敗: {e}")return []

4.4 DHT（分布式哈希表）操作

class IPFSDHTManager(IPFSPubSubManager):"""IPFS DHT管理類"""def find_peer(self, peer_id: str) -> Dict[str, Any]:"""查找對等節點信息Args:peer_id: 對等節點IDReturns:Dict: 對等節點信息"""try:result = self.client.dht_findpeer(peer_id)return resultexcept Exception as e:print(f"查找對等節點失敗: {e}")raisedef find_providers(self, cid: str, max_results: int = 10) -> List[Dict[str, Any]]:"""查找內容提供者Args:cid: 內容標識符max_results: 最大結果數量Returns:List: 提供者列表"""try:providers = []for provider in self.client.dht_findprovs(cid):if len(providers) >= max_results:breakproviders.append(provider)return providersexcept Exception as e:print(f"查找內容提供者失敗: {e}")return []def query_dht(self, key: str) -> List[Dict[str, Any]]:"""查詢DHTArgs:key: DHT鍵Returns:List: 查詢結果"""try:result = self.client.dht_get(key)return resultexcept Exception as e:print(f"DHT查詢失敗: {e}")return []def put_dht_record(self, key: str, value: str) -> bool:"""向DHT存儲記錄Args:key: DHT鍵value: 存儲值Returns:bool: 存儲是否成功"""try:self.client.dht_put(key, value)print(f"DHT記錄存儲成功: {key}")return Trueexcept Exception as e:print(f"DHT記錄存儲失敗: {e}")return False

5 完整示例應用

5.1 去中心化文件共享系統

class DecentralizedFileSharingSystem(IPFSDHTManager):"""去中心化文件共享系統"""def __init__(self, host: str = '127.0.0.1', port: int = 5001):super().__init__(host, port)self.shared_files = {}self.connect()def share_file(self, file_path: str, description: str = "") -> Dict[str, Any]:"""共享文件Args:file_path: 文件路徑description: 文件描述Returns:Dict: 共享信息"""try:# 添加文件到IPFSresult = self.add_file(file_path)cid = result['Hash']# 固定文件self.pin_content(cid)# 創建文件元數據metadata = {'cid': cid,'name': os.path.basename(file_path),'description': description,'size': os.path.getsize(file_path),'timestamp': time.time(),'mime_type': self._guess_mime_type(file_path)}# 存儲元數據metadata_cid = self.add_json_data(metadata)['Hash']# 發布到IPNSipns_result = self.publish_to_ipns(metadata_cid)# 記錄共享文件self.shared_files[cid] = {'metadata': metadata,'metadata_cid': metadata_cid,'ipns_name': ipns_result['Name']}print(f"文件共享成功!")print(f"CID: {cid}")print(f"IPNS: {ipns_result['Name']}")return {'cid': cid,'ipns': ipns_result['Name'],'metadata': metadata}except Exception as e:print(f"文件共享失敗: {e}")raisedef discover_shared_files(self, ipns_name: str = None) -> List[Dict[str, Any]]:"""發現共享文件Args:ipns_name: 特定的IPNS名稱(可選)Returns:List: 共享文件列表"""try:discovered_files = []if ipns_name:# 解析特定IPNSmetadata_cid = self.resolve_ipns(ipns_name)metadata = self.get_json_data(metadata_cid)discovered_files.append(metadata)else:# 這里可以實現更復雜的發現機制# 例如使用PubSub或DHT查詢print("發現功能需要實現更復雜的查詢邏輯")return discovered_filesexcept Exception as e:print(f"文件發現失敗: {e}")return []def download_shared_file(self, cid: str, output_path: str) -> bool:"""下載共享文件Args:cid: 文件CIDoutput_path: 輸出路徑Returns:bool: 下載是否成功"""try:return self.download_file(cid, output_path)except Exception as e:print(f"文件下載失敗: {e}")return Falsedef _guess_mime_type(self, file_path: str) -> str:"""猜測文件MIME類型Args:file_path: 文件路徑Returns:str: MIME類型"""import mimetypesmime_type, _ = mimetypes.guess_type(file_path)return mime_type or 'application/octet-stream'def create_shared_directory(self, directory_path: str, description: str = "") -> Dict[str, Any]:"""共享整個目錄Args:directory_path: 目錄路徑description: 目錄描述Returns:Dict: 共享信息"""try:# 添加目錄到IPFSresult = self.add_directory(directory_path)# 獲取根目錄CID（通常是最后一個結果）root_cid = result[-1]['Hash']# 固定整個目錄self.pin_content(root_cid)# 創建目錄元數據metadata = {'cid': root_cid,'name': os.path.basename(directory_path),'description': description,'timestamp': time.time(),'file_count': len(result),'type': 'directory'}# 存儲元數據metadata_cid = self.add_json_data(metadata)['Hash']# 發布到IPNSipns_result = self.publish_to_ipns(metadata_cid)print(f"目錄共享成功!")print(f"根目錄CID: {root_cid}")print(f"IPNS: {ipns_result['Name']}")return {'cid': root_cid,'ipns': ipns_result['Name'],'metadata': metadata}except Exception as e:print(f"目錄共享失敗: {e}")raise# 使用示例
def demo_file_sharing():"""演示文件共享系統"""try:# 創建文件共享系統實例fs = DecentralizedFileSharingSystem()# 共享文件file_info = fs.share_file("example.txt", "這是一個示例文件")# 等待一段時間讓內容傳播time.sleep(2)# 發現文件（這里簡化實現）discovered = fs.discover_shared_files(file_info['ipns'])print(f"發現文件: {discovered}")# 下載文件fs.download_shared_file(file_info['cid'], "downloaded_example.txt")except Exception as e:print(f"演示失敗: {e}")if __name__ == "__main__":demo_file_sharing()

5.2 實時協作編輯器

class RealTimeCollaborativeEditor(IPFSDHTManager):"""實時協作編輯器"""def __init__(self, document_name: str, host: str = '127.0.0.1', port: int = 5001):super().__init__(host, port)self.document_name = document_nameself.topic = f"collab-doc-{document_name}"self.document_state = ""self.callbacks = []self.connect()def on_message(self, message: Dict[str, Any]):"""處理接收到的消息"""try:if message['type'] == 'edit':# 應用編輯操作self._apply_edit(message['position'],message['text'],message['is_delete'])# 通知回調for callback in self.callbacks:callback(message)except Exception as e:print(f"消息處理錯誤: {e}")def _apply_edit(self, position: int, text: str, is_delete: bool):"""應用編輯操作"""if is_delete:# 刪除操作self.document_state = (self.document_state[:position] + self.document_state[position + len(text):])else:# 插入操作self.document_state = (self.document_state[:position] + text + self.document_state[position:])def start_collaboration(self):"""開始協作"""# 訂閱主題self.subscribe_to_topic(self.topic, self.on_message)print(f"已加入協作文檔: {self.document_name}")def send_edit(self, position: int, text: str, is_delete: bool = False):"""發送編輯操作"""message = {'type': 'edit','position': position,'text': text,'is_delete': is_delete,'timestamp': time.time(),'author': self.get_node_info()['ID']}self.publish_message(self.topic, json.dumps(message))def add_callback(self, callback: callable):"""添加消息回調"""self.callbacks.append(callback)def save_document(self) -> str:"""保存文檔到IPFS"""try:result = self.add_bytes_data(self.document_state.encode('utf-8'),f"{self.document_name}.txt")print(f"文檔已保存: {result['Hash']}")return result['Hash']except Exception as e:print(f"文檔保存失敗: {e}")raisedef load_document(self, cid: str):"""從IPFS加載文檔"""try:content = self.get_file(cid)self.document_state = content.decode('utf-8')print(f"文檔已加載: {cid}")except Exception as e:print(f"文檔加載失敗: {e}")raise# 使用示例
def demo_collaborative_editor():"""演示協作編輯器"""try:# 創建兩個編輯器實例（模擬兩個用戶）user1 = RealTimeCollaborativeEditor("test-doc")user2 = RealTimeCollaborativeEditor("test-doc")# 用戶1開始協作user1.start_collaboration()# 用戶2開始協作user2.start_collaboration()# 用戶1發送編輯user1.send_edit(0, "Hello, World!")# 等待消息傳播time.sleep(1)print(f"用戶2的文檔狀態: {user2.document_state}")# 用戶2發送編輯user2.send_edit(12, " This is collaborative editing!")# 等待消息傳播time.sleep(1)print(f"用戶1的文檔狀態: {user1.document_state}")# 保存文檔cid = user1.save_document()print(f"文檔保存為: {cid}")except Exception as e:print(f"協作演示失敗: {e}")if __name__ == "__main__":demo_collaborative_editor()

6 性能優化與最佳實踐

6.1 連接池管理

class IPFSConnectionPool:"""IPFS連接池"""def __init__(self, host: str = '127.0.0.1', port: int = 5001, max_connections: int = 10):self.host = hostself.port = portself.max_connections = max_connectionsself.connection_pool = []self.lock = threading.Lock()def get_connection(self) -> ipfsapi.Client:"""獲取連接"""with self.lock:if self.connection_pool:return self.connection_pool.pop()else:return ipfsapi.Client(host=self.host, port=self.port)def release_connection(self, connection: ipfsapi.Client):"""釋放連接"""with self.lock:if len(self.connection_pool) < self.max_connections:self.connection_pool.append(connection)else:# 超過最大連接數，直接關閉try:# 沒有直接的關閉方法，這里只是從池中移除passexcept:passdef execute_with_connection(self, func: callable, *args, **kwargs) -> Any:"""使用連接執行函數"""connection = self.get_connection()try:return func(connection, *args, **kwargs)finally:self.release_connection(connection)

6.2 批量操作優化

class IPFSBatchOperations(IPFSDHTManager):"""IPFS批量操作優化"""def __init__(self, host: str = '127.0.0.1', port: int = 5001, batch_size: int = 100):super().__init__(host, port)self.batch_size = batch_sizedef batch_add_files(self, file_paths: List[str]) -> List[Dict[str, Any]]:"""批量添加文件Args:file_paths: 文件路徑列表Returns:List: 添加結果列表"""results = []for i in range(0, len(file_paths), self.batch_size):batch = file_paths[i:i + self.batch_size]print(f"處理批次 {i//self.batch_size + 1}: {len(batch)} 個文件")for file_path in batch:try:result = self.add_file(file_path)results.append(result)except Exception as e:print(f"文件添加失敗 {file_path}: {e}")results.append({'error': str(e), 'file': file_path})return resultsdef batch_pin_content(self, cids: List[str]) -> Dict[str, Any]:"""批量固定內容Args:cids: CID列表Returns:Dict: 固定結果"""success_count = 0fail_count = 0errors = []for cid in cids:try:if self.pin_content(cid):success_count += 1else:fail_count += 1errors.append(cid)except Exception as e:fail_count += 1errors.append({'cid': cid, 'error': str(e)})return {'success': success_count,'failed': fail_count,'errors': errors}def batch_download_files(self, cid_path_pairs: List[tuple]) -> Dict[str, Any]:"""批量下載文件Args:cid_path_pairs: (CID, 保存路徑) 元組列表Returns:Dict: 下載結果"""success_count = 0fail_count = 0errors = []for cid, save_path in cid_path_pairs:try:if self.download_file(cid, save_path):success_count += 1else:fail_count += 1errors.append({'cid': cid, 'path': save_path})except Exception as e:fail_count += 1errors.append({'cid': cid, 'path': save_path, 'error': str(e)})return {'success': success_count,'failed': fail_count,'errors': errors}

6.3 錯誤處理與重試機制

class IPFSRetryManager(IPFSBatchOperations):"""IPFS重試管理"""def __init__(self, host: str = '127.0.0.1', port: int = 5001, max_retries: int = 3, retry_delay: float = 1.0):super().__init__(host, port)self.max_retries = max_retriesself.retry_delay = retry_delaydef execute_with_retry(self, func: callable, *args, **kwargs) -> Any:"""帶重試的執行Args:func: 執行函數*args: 函數參數**kwargs: 函數關鍵字參數Returns:Any: 函數執行結果"""last_exception = Nonefor attempt in range(self.max_retries):try:return func(*args, **kwargs)except ipfsapi.exceptions.ConnectionError as e:last_exception = eprint(f"連接錯誤 (嘗試 {attempt + 1}/{self.max_retries}): {e}")except ipfsapi.exceptions.TimeoutError as e:last_exception = eprint(f"超時錯誤 (嘗試 {attempt + 1}/{self.max_retries}): {e}")except Exception as e:last_exception = eprint(f"未知錯誤 (嘗試 {attempt + 1}/{self.max_retries}): {e}")# 等待后重試if attempt < self.max_retries - 1:time.sleep(self.retry_delay * (2 ** attempt))  # 指數退避# 所有重試都失敗raise last_exceptiondef robust_add_file(self, file_path: str, **kwargs) -> Dict[str, Any]:"""健壯的文件添加方法Args:file_path: 文件路徑**kwargs: 其他參數Returns:Dict: 添加結果"""return self.execute_with_retry(super().add_file, file_path, **kwargs)def robust_get_file(self, cid: str, **kwargs) -> bytes:"""健壯的文件獲取方法Args:cid: 內容標識符**kwargs: 其他參數Returns:bytes: 文件內容"""return self.execute_with_retry(super().get_file,cid,**kwargs)def robust_pin_content(self, cid: str, **kwargs) -> bool:"""健壯的內容固定方法Args:cid: 內容標識符**kwargs: 其他參數Returns:bool: 固定是否成功"""return self.execute_with_retry(super().pin_content,cid,**kwargs)

7 完整代碼示例

下面是一個完整的IPFS API使用示例，集成了上述所有功能：

#!/usr/bin/env python3
"""
IPFS API完整示例
演示IPFS API的各種用法和最佳實踐
"""import json
import time
import os
import threading
from typing import Dict, List, Any, Optional
import ipfsapiclass CompleteIPFSExample:"""完整的IPFS API示例類"""def __init__(self, host: str = '127.0.0.1', port: int = 5001):self.host = hostself.port = portself.api = Noneself.connection_pool = []self.max_connections = 5self.lock = threading.Lock()def connect(self) -> bool:"""連接到IPFS節點"""try:self.api = ipfsapi.connect(self.host, self.port)print(f"成功連接到IPFS節點 {self.host}:{self.port}")print(f"節點ID: {self.api.id()['ID']}")return Trueexcept Exception as e:print(f"連接失敗: {e}")return Falsedef get_connection(self) -> ipfsapi.Client:"""獲取連接（簡單連接池）"""with self.lock:if self.connection_pool:return self.connection_pool.pop()return ipfsapi.Client(host=self.host, port=self.port)def release_connection(self, conn: ipfsapi.Client):"""釋放連接"""with self.lock:if len(self.connection_pool) < self.max_connections:self.connection_pool.append(conn)def demonstrate_basic_operations(self):"""演示基本操作"""print("\n=== 基本操作演示 ===")# 添加文件test_content = "Hello IPFS World! 這是測試內容。"with open('test_file.txt', 'w', encoding='utf-8') as f:f.write(test_content)try:# 添加文件add_result = self.api.add('test_file.txt')file_cid = add_result['Hash']print(f"文件添加成功: {file_cid}")# 獲取文件內容content = self.api.cat(file_cid)print(f"文件內容: {content.decode('utf-8')}")# 固定內容pin_result = self.api.pin_add(file_cid)print(f"內容固定成功: {pin_result}")# 獲取節點信息node_info = self.api.id()print(f"節點ID: {node_info['ID']}")print(f"節點地址: {node_info['Addresses']}")except Exception as e:print(f"基本操作演示失敗: {e}")def demonstrate_advanced_operations(self):"""演示高級操作"""print("\n=== 高級操作演示 ===")try:# 創建并發布到IPNStest_data = {'message': 'Hello IPNS!','timestamp': time.time(),'author': 'IPFS Demo'}# 添加JSON數據json_cid = self.api.add_str(json.dumps(test_data))print(f"JSON數據添加成功: {json_cid}")# 發布到IPNSipns_result = self.api.name_publish(json_cid)print(f"IPNS發布成功: {ipns_result['Name']}")# 解析IPNSresolve_result = self.api.name_resolve(ipns_result['Name'])print(f"IPNS解析結果: {resolve_result}")except Exception as e:print(f"高級操作演示失敗: {e}")def demonstrate_directory_operations(self):"""演示目錄操作"""print("\n=== 目錄操作演示 ===")try:# 創建測試目錄和文件test_dir = 'test_directory'os.makedirs(test_dir, exist_ok=True)# 創建多個測試文件for i in range(3):with open(f'{test_dir}/file_{i}.txt', 'w') as f:f.write(f'這是文件 {i} 的內容')# 添加整個目錄add_results = self.api.add(test_dir, recursive=True)print(f"目錄添加成功，共 {len(add_results)} 個文件")# 找到根目錄CID（通常是最后一個）root_cid = Nonefor result in add_results:if result['Name'] == test_dir:root_cid = result['Hash']breakif root_cid:print(f"根目錄CID: {root_cid}")# 列出目錄內容ls_result = self.api.ls(root_cid)print("目錄內容:")for item in ls_result['Objects'][0]['Links']:print(f"  {item['Name']} - {item['Hash']} - {item['Size']} bytes")except Exception as e:print(f"目錄操作演示失敗: {e}")finally:# 清理測試文件import shutilif os.path.exists('test_directory'):shutil.rmtree('test_directory')if os.path.exists('test_file.txt'):os.remove('test_file.txt')def demonstrate_pubsub(self):"""演示PubSub功能"""print("\n=== PubSub演示 ===")topic = 'test-topic'received_messages = []def message_handler(message):"""消息處理函數"""try:msg_data = json.loads(message['data'].decode('utf-8'))received_messages.append(msg_data)print(f"收到消息: {msg_data}")except:passtry:# 訂閱主題sub = self.api.pubsub_sub(topic)# 啟動訂閱線程def subscription_worker():for message in sub:message_handler(message)sub_thread = threading.Thread(target=subscription_worker, daemon=True)sub_thread.start()# 發布消息for i in range(3):message = {'number': i,'text': f'測試消息 {i}','timestamp': time.time()}self.api.pubsub_pub(topic, json.dumps(message))print(f"已發布消息: {message}")time.sleep(0.5)# 等待消息處理time.sleep(1)# 關閉訂閱sub.close()print(f"共收到 {len(received_messages)} 條消息")except Exception as e:print(f"PubSub演示失敗: {e}")def run_all_demos(self):"""運行所有演示"""if not self.connect():print("無法連接到IPFS節點，請確保IPFS守護進程正在運行")returnself.demonstrate_basic_operations()self.demonstrate_advanced_operations()self.demonstrate_directory_operations()self.demonstrate_pubsub()print("\n=== 演示完成 ===")def main():"""主函數"""print("IPFS API 完整示例")print("=" * 50)# 創建示例實例example = CompleteIPFSExample()# 運行所有演示example.run_all_demos()if __name__ == "__main__":main()