当前位置：首页 > news >正文

Python第三方库IPFS-API使用详解：构建去中心化应用的完整指南

news 2025/8/24 8:02:33

星际文件系统（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()