Web3区块链【免费下载链接】web3.pyA python interface for interacting with the Ethereum blockchain and ecosystem.项目地址https://gitcode.com/gh_mirrors/we/web3.py点击查看免费下载web3.py 在公开 APIWeb3/AsyncWeb3对象与所连接的节点之间存在多层抽象Provider 负责与区块链的底层通信Middleware 提供请求/响应的拦截钩子Manager 则承担线程安全与异步原语。本文以官方文档 docs/internals.rst 为核心骨架结合仓库源码与测试逐层拆解请求生命周期、请求缓存Request Caching、HTTP 异常重试以及WebSocketProvider/AsyncIPCProvider持久连接下的请求-响应匹配与订阅处理帮助高级用户理解并定制这些底层 API。⚠️ 适用提示本文涉及的是面向高级用户的内层 API。如果你不清楚自己在做什么建议停留在公开 API如w3.eth、w3.contract层面。请求生命周期一次 RPC 调用的完整旅程每一个 web3 RPC 调用都会依次穿过以下三层*********** ************ | Request | | Response | *********** ************ | ^ v | ----------------------------- | Manager | ----------------------------- | ^ v | ----------------------------- | Middleware | ----------------------------- | ^ v | ----------------------------- | Provider | -----------------------------可以用“洋葱”来理解这个关系Provider 在最中心请求从最外层的 Manager 发起逐层穿过每一层 Middleware最终抵达中心的 ProviderProvider 处理完后响应再从洋葱中心逐层向外传出最终由 Manager 返回给调用者。从源码看这一流程的实现位于 RequestManager_make_request()调用provider.request_func()后者在 BaseProvider.request_func() 中把整个MiddlewareOnion与self.make_request组合combine_middleware再逐层包装出最终的请求函数同步路径经request_blocking()返回异步路径经coro_request()返回最终都会调用 formatted_response() 做 JSON-RPC 结果校验与格式化。异步场景下请求与响应分离send()/recv()详见下文“持久连接”一节。利用这套分层可以实现的常见场景包括把某些 RPC 请求重定向到不同的 Provider例如所有读操作发给远程节点所有写操作发给本地自控节点透明拦截eth_sendTransaction发送的交易在本地签名后改走eth_sendRawTransaction发送修改 RPC 响应格式例如把响应中的整数值统一转换为十六进制校验 RPC 请求的输入参数。Providers直接与区块链交互的层Provider 负责所有与区块链的直接交互。大多数情况下这意味着通过 HTTP 或 IPC socket 与以太坊节点的 JSON-RPC 服务通信但 Provider 并不强制要求基于 RPC例如测试场景可以基于内存 EVMin-memory EVM实现一个 Provider 来满足请求仓库中的 eth-tester 相关实现即属此类。编写自定义 Provider自定义 Provider 需要实现两个必需方法并设置该 Provider 使用的 MiddlewareBaseProvider.make_request(method, params)每个 Provider 类必须实现此方法。它应当返回一个 JSON 对象成功时带result键失败时带error键。method被调用的 JSON-RPC 方法名字符串如eth_sendTransactionparams该 JSON-RPC 方法的参数列表或其他可迭代对象。基类中的默认实现直接抛出NotImplementedError见 web3/providers/base.py。BaseProvider.is_connected(show_tracebackFalse)根据 Provider 是否应被视为“已连接”返回True/False。例如 IPC socket 类 Providersocket 打开返回True关闭返回False若传入show_tracebackTrue则在不应视为已连接时抛出ProviderConnectionError并给出原因。参考实现JSONBaseProvider.is_connected()web3/providers/base.py会发一次web3_clientVersion请求来探测连接。BaseProvider.middleware应为可迭代的 Middleware 集合。可以通过给provider.middleware赋值来设置新的 Middleware 列表列表开头的第一个 Middleware 最先处理请求。此外JSONBaseProvider还提供了encode_rpc_request/decode_rpc_response见 web3/providers/base.py负责把请求封装为 JSON-RPC 2.0 格式自动生成递增的id以及解析原始字节响应自定义 HTTP/IPC 类 Provider 可复用这些工具。Provider 配置请求缓存Request Caching⚠️ 重要在启用请求缓存之前请先熟悉其校验逻辑。该特性为了尽力保证数据有效性常常需要在底层发起额外的请求可能为你的场景引入不必要的开销。可以设置request_cache_validation_threshold为None来关闭校验缓存所有允许缓存的请求也可以按需调整以匹配你的性能诉求。请求缓存通过 Provider 实例上的以下配置项启用与调整cache_allowed_requests: bool False总开关默认关闭cacheable_requests: Optional[Set[RPCEndpoint]]允许被缓存的 RPC 端点集合request_cache_validation_threshold: Optional[Union[RequestCacheValidationThreshold, int]]缓存校验阈值。对不依赖区块数据的请求如eth_chainId把cache_allowed_requests设为True即可安全地缓存所有响应。但对依赖区块数据的请求如eth_getBlockByNumber不能总是缓存——区块数据会变化例如链重组reorg期间或尚未达成最终性finality时。request_cache_validation_threshold即为这类请求配置一个安全的缓存阈值默认情况下该选项被配置为对你所连接链的 chain id 而言“安全”的内部值连接以太坊主网时取finalized已最终确定区块号连接其他链时取一个从当前时刻算起、对该链最终性机制“安全”的时间间隔秒。对不在内部列表中的链包括所有测试网默认值为 1 小时。内部配置的链及其默认阈值如下源码见 CHAIN_VALIDATION_THRESHOLD_DEFAULTS链默认校验阈值ETHRequestCacheValidationThreshold.FINALIZEDfinalized 区块ARB17 天ZKSYNC1 小时OETH3 分钟MATIC30 分钟ZKEVM1 小时BASE7 天SCR1 小时GNO5 分钟AVAX2 分钟BNB2 分钟FTM1 分钟以以太坊主网为例请求所依赖的区块号小于等于finalized区块号时响应会被缓存超过finalized区块号则不缓存。对其他链当请求所依赖数据的区块时间戳早于等于该链配置的时间间隔时缓存响应。可以通过三种方式修改该行为见 BaseProvider 的构造参数设为RequestCacheValidationThreshold.SAFE以safe区块作为阈值仅以太坊主网设为自定义秒数任何链包括以太坊主网设为None关闭校验、缓存所有请求对非测试网链不推荐。RequestCacheValidationThreshold枚举主网finalized与safe两个值从web3.utils模块导入见 web3/utils/caching.py。另外注意cacheable_requests用于限定允许缓存的端点集合默认是一个内部维护的“安全可缓存”列表排除了像eth_call这种响应可变、不宜缓存的端点。默认的可缓存请求列表如下加粗者为受request_cache_validation_threshold校验的请求eth_chainIdweb3_clientVersionnet_versioneth_getBlockByNumbereth_getRawTransactionByBlockNumberAndIndexeth_getBlockTransactionCountByNumbereth_getUncleByBlockNumberAndIndexeth_getUncleCountByBlockNumbereth_getBlockByHasheth_getTransactionByHasheth_getTransactionByBlockNumberAndIndexeth_getTransactionByBlockHashAndIndexeth_getBlockTransactionCountByHasheth_getRawTransactionByBlockHashAndIndexeth_getUncleByBlockHashAndIndexeth_getUncleCountByBlockHash从实现看该默认列表由 INTERNAL_VALIDATION_MAP 的键构成CACHEABLE_REQUESTS并按端点类型分为三类校验器参数含区块号validate_from_block_id_in_params、结果含区块信息validate_from_blocknum_in_result、参数含区块哈希validate_from_blockhash_in_params异步场景有对应实现request_caching_validation.py。值得注意的是这些校验器在必要时会额外发起 RPC 请求来获取区块信息例如验证一笔交易是否已越过安全阈值高频率请求场景下可能造成明显开销此外校验期间会临时关闭缓存以防递归见 is_beyond_validation_threshold 的cache_allowed_requests False处理。因此理解何时开缓存、如何配置校验是避免不必要开销的关键。配置示例from web3 import Web3, HTTPProvider from web3.utils import RequestCacheValidationThreshold w3 Web3(HTTPProvider( endpoint_uri..., # 可选开启请求缓存默认 False cache_allowed_requestsTrue, # 可选允许缓存的端点集合默认是内部“安全可缓存”列表见上 cacheable_requests{eth_chainId, eth_getBlockByNumber}, # 可选默认根据 chain id 取值见上 request_cache_validation_threshold60 * 60, # 1 小时 # request_cache_validation_thresholdRequestCacheValidationThreshold.SAFE, # 仅以太坊主网 ))Provider 配置HTTP Provider 请求重试HTTPProvider与AsyncHTTPProvider默认会对某些请求在异常时进行重试。通过 Provider 实例上的exception_retry_configuration属性配置其值为 ExceptionRetryConfiguration一个 pydanticBaseModel实例。重试机制采用指数退避策略以backoff_factor决定的初始值为起点每次重试间隔翻倍最多尝试retries次。配置项说明errorsProvider 应重试的异常元组。默认值HTTPProvider为(ConnectionError, requests.HTTPError, requests.Timeout)AsyncHTTPProvider为(aiohttp.ClientError, asyncio.TimeoutError)。两种 Provider 的DEFAULT_EXCEPTIONS定义见下。retries重试次数默认 5。backoff_factor初始延迟倍数每次重试翻倍默认 0.125。method_allowlist允许重试的方法列表默认是内部“安全可重试”方法清单REQUEST_RETRY_ALLOWLIST见 web3/providers/rpc/utils.py包含admin、net、txpool、testing、evm前缀方法以及eth_call、eth_sendRawTransaction等读/写方法。示例展示默认选项及如何覆盖from web3 import Web3, HTTPProvider from web3.providers.rpc.utils import ( REQUEST_RETRY_ALLOWLIST, ExceptionRetryConfiguration, ) w3 Web3(HTTPProvider( endpoint_uri..., exception_retry_configurationExceptionRetryConfiguration( errorsDEFAULT_EXCEPTIONS, # 重试次数 retries5, # 初始延迟倍数每次重试翻倍 backoff_factor0.125, # 内部默认的可重试方法清单 method_allowlistREQUEST_RETRY_ALLOWLIST, ), ))不同 HTTP Provider 的DEFAULT_EXCEPTIONS定义为HTTPProvider(ConnectionError, requests.HTTPError, requests.Timeout)AsyncHTTPProvider(ConnectionError, aiohttp.ClientError, asyncio.TimeoutError)把retry_configuration即exception_retry_configuration设为None将关闭该 Provider 实例的异常重试from web3 import Web3, HTTPProvider w3 Web3(HTTPProvider(endpoint_uri..., retry_configurationNone)从实现看重试逻辑位于 HTTPProvider._make_request() 与AsyncHTTPProvider的对应方法仅当配置非空且check_if_retry_on_failure()web3/providers/rpc/utils.py判定方法在允许列表中时执行重试每次捕获errors中的异常后按backoff_factor * 2**i计算退避延迟见 async_rpc.py。仓库在 tests/core/providers/test_http_request_retry.py 中对重试次数、异常类型与配置等价性均有覆盖。Managers请求/响应生命周期的守门人Manager 是请求/响应生命周期的守门人gatekeeper。大部分功能可以在 Middleware 层实现因此一般无需更换 Manager。RequestManager在初始化时会装配默认 Middleware 洋葱get_default_middlewaregas_price_strategy、ens_name_to_address、attrdict、validation、gas_estimate并提供request_blocking/coro_request两个同步/异步请求入口。持久连接 Provider 的请求处理RequestProcessorRequestProcessor 负责为PersistentConnectionProvider存储并同步异步请求与其响应。WebSocketProvider 与AsyncIPCProvider是两个持久连接 Provider。由于要通过同一个 socket 发送请求并接收对应响应PersistentConnectionProvider必须把请求的id与 socket 返回的响应的id匹配起来。任何不按 JSON-RPC 2.0 规范携带 id 的 Provider 都无法与PersistentConnectionProvider一起工作。从实现看RequestProcessor内部维护了三个独立数据结构request_processor.py_request_information_cache请求信息缓存容量默认 500、_request_response_cache一对一响应缓存容量 500、_subscription_response_queue一对多订阅队列FIFO容量默认 500类型为TaskReliantQueue。监听响应PersistentConnectionProvider的实现类在 socket 连接建立时会启动一个消息监听后台任务_message_listener_task见 persistent.py 的connect()。该任务负责监听 socket 连接上到来的所有消息并存入 Provider 实例内部的RequestProcessor。RequestProcessor根据消息是否带 JSON-RPCid值把消息存到不同的缓存一对一缓存或一对多订阅队列。一对一请求一对一请求指期望只收到一个响应的请求例如通过eth模块 API 获取最新区块号 async def ws_one_to_one_example(): ... async with AsyncWeb3(WebSocketProvider(fws://127.0.0.1:8546)) as w3: ... # 发起请求并在同一行内收到单个响应 ... latest_block_num await w3.eth.block_number asyncio.run(ws_one_to_one_example())持久 socket 连接下我们只能调用send()再通过其他方式一般是recv()或遍历 socket 消息异步接收响应。正因收发是异步的为了把响应匹配回原始请求必须把请求信息存到某处即与收到的响应id匹配的那个请求。存储的请求信息随后用于处理响应把它送入 web3.py 库内部的响应格式化器response formatters与中间件处理管线。存储请求信息的载体是RequestProcessor内部的RequestInformation缓存类其保存的字段如下源码见 RequestInformationmethod方法名如eth_subscribeparams发起调用时的参数如(newPendingTransactions, True)response_formatters用于处理响应的格式化器middleware_response_processors发起请求时实例上存在的、处理响应的中间件处理器按顺序追加响应到来时依序经过这些逻辑subscription_id若请求是eth_subscribe收到订阅调用的响应即订阅id时不是把这条信息从缓存弹出而是把订阅 id 一并保存到请求信息中以便正确后续处理所有携带该订阅id的订阅消息一对一请求场景该值恒为None。一对一响应响应对象中含 JSON-RPCid存放在内部SimpleCache实例中与任何一对多响应隔离。PersistentConnectionProvider在内部查找响应时期望监听任务把响应存入该缓存由于缓存键的生成使用了请求id它会查找与响应id匹配请求id的缓存键找到则处理并返回给用户找不到则操作超时并抛出TimeExhausted异常。该超时可通过实例化PersistentConnectionProvider时的response_timeout关键字参数配置对应源码中的request_timeout默认 30 秒见 persistent.py。相关缓存键生成逻辑见 generate_cache_key。一对多请求订阅一对多请求指一次初始请求后期望收到多条响应的请求目前唯一的例子是eth_subscribe。初始的eth_subscribe请求本身只期待一个响应——订阅id值但成功后还期待收到很多eth_subscription消息。因此原始请求被视为一对一请求以便在同一行把订阅id返回给用户该调用后续产生的多条响应可以按以下几种方式之一处理。方式一订阅管理器 API推荐订阅管理器 API 是AsyncWeb3类上的公开 API当连接的是PersistentConnectionProvider实例时可用允许用户订阅一个订阅并异步处理其多条响应。只要每次订阅调用都传入 handlersubscription_manager实例就负责处理 socket 连接上到来的多条响应用户用完订阅后也可以用它退订。 async def new_heads_handler( ... handler_context: NewHeadsSubscriptionContext, ... ) - None: ... result handler_context.result ... print(fNew block header: {result}\n) ... if result[number] 1234567: ... await handler_context.subscription.unsubscribe() async def ws_subscription_example(): ... async with AsyncWeb3(WebSocketProvider(fws://127.0.0.1:8546)) as w3: ... # 订阅新区块头并拿到 subscription_id。 ... # 这是一次一对一调用附带大量响应的触发器 ... subscription_id await w3.eth.subscribe(newHeads, handlernew_heads_handler) ... ... # 用订阅管理器异步处理订阅消息。 ... # 会一直运行到订阅管理器中不再有订阅为止 ... # 若把 run_forever 置为 True 则无限运行。 ... await w3.subscription_manager.handle_subscriptions(run_foreverFalse) asyncio.run(ws_subscription_example())订阅管理器还可以一次性订阅多个订阅。EthSubscription系列类可从web3.utils.subscriptions导入提供了友好的订阅管理 API。由于每个连接与 Provider 实例都有各自的监听任务与订阅管理器实例你可以一次订阅多个订阅并通过 handler 处理多条响应。handler 中包含async_w3发起该订阅的AsyncWeb3实例subscriptionhandler 所依附的订阅实例resultsocket 连接上到来的、属于该订阅的响应。订阅还接受handler_context参数可在订阅时向 handler 传递额外信息例如传入一个事件对象用于在日志事件到来时解析它。 from web3 import ( ... AsyncWeb3, ... WebSocketProvider, ... AsyncIPCProvider, ... ) from web3.utils.subscriptions import ( ... EthSubscription, ... NewHeadsSubscription, ... NewHeadsSubscriptionContext, ... PendingTxSubscription, ... PendingTxSubscriptionContext, ... LogsSubscription, ... LogsSubscriptionContext, ... ) async def new_heads_handler( ... handler_context: NewHeadsSubscriptionContext, ... ) - None: ... header handler_context.result ... print(fNew block header: {header}\n) ... if header[number] 1234567: ... await handler_context.subscription.unsubscribe() async def pending_txs_handler( ... handler_context: PendingTxSubscriptionContext, ... ) - None: ... ... async def log_handler( ... handler_context: LogsSubscriptionContext, ... ) - None: ... log_receipt handler_context.result ... # 订阅日志时我们把事件传入了 handler_context现在可直接取用 ... event_data handler_context.transfer_event.process_log(log_receipt) ... print(fLog event data: {event_data}\n) async def sub_manager(): ... local_w3 await AsyncWeb3(AsyncIPCProvider(LOCAL_IPC)) ... ... # 通过订阅管理器 handler 一次订阅多个订阅 ... weth_contract local_w3.eth.contract( ... addresslocal_w3.to_checksum_address(0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2), ... abiWETH_ABI, ... ) ... transfer_event weth_contract.events.Transfer() ... await local_w3.subscription_manager.subscribe( ... [ ... NewHeadsSubscription(labelnew-heads-mainnet, handlernew_heads_handler), ... PendingTxSubscription( ... labelpending-tx-mainnet, # 可选 label ... full_transactionsTrue, ... handlerpending_tx_handler, ... ), ... LogsSubscription( ... labelWETH transfers, # 可选 label ... addressweth_contract.address, ... topics[transfer_event.topic], ... handlerlog_handler, ... # 可选的 handler_context 参数帮助解析响应 ... handler_context{transfer_event: transfer_event}, ... ), ... ] ... ) ... ... public_w3 await AsyncWeb3(WebSocketProvider(PUBLIC_PROVIDER_WS)) ... # 通过 eth_subscribe 订阅带 handler 与可选 label ... await public_w3.eth.subscribe(public_newHeads, handlerpending_tx_handler, labelnew-heads-public-ws) # 处理所有订阅直到两个订阅管理器中都不再有订阅。 ... # 若任一管理器的 run_forever 为 True则无限运行。 await asyncio.gather( ... public_w3.subscription_manager.handle_subscriptions(), ... local_w3.subscription_manager.handle_subscriptions(), ... ) ... ... # 关闭连接 ... await local_w3.provider.disconnect() ... await public_w3.provider.disconnect() asyncio.run(sub_manager())说明订阅系列类如 NewHeadsSubscription、PendingTxSubscription、LogsSubscription、SyncingSubscription的构造参数在源码中均有类型注解与默认值例如PendingTxSubscription(full_transactionsFalse)、LogsSubscription(addressNone, topicsNone)且各订阅都支持可选的label、handler_context与parallelize参数w3.eth.subscribe()会按订阅类型自动创建对应的类型化订阅见 async_eth.py 的 subscribe() 与 _create_type_aware_subscription。方式二process_subscriptions()异步迭代器PersistentConnection类web3/providers/persistent/persistent_connection.py是操作活动持久 socket 连接的公开 API其process_subscriptions()方法按异步迭代器模式接收eth_subscription响应。你可以用它在消息到来时监听并处理原始消息。 async def ws_subscription_example(): ... async with AsyncWeb3(WebSocketProvider(fws://127.0.0.1:8546)) as w3: ... # 订阅新区块头并拿到 subscription_id。 ... # 这是一次一对一调用附带大量响应的触发器 ... subscription_id await w3.eth.subscribe(newHeads) ... ... # 利用 w3.socketPersistentConnection 公开 API的 ... # process_subscriptions() 方法监听 socket 上的大量响应 ... async for response in w3.socket.process_subscriptions(): ... # 这里只接收一对多响应避免把一对一请求的响应 ... # 意外地带进这个代码块 ... ... print(f{response}\n) ... ... if some_condition: ... # 退订新区块头这是另一次一对一请求 ... is_unsubscribed await w3.eth.unsubscribe(subscription_id) ... if is_unsubscribed: ... break asyncio.run(ws_subscription_example())一对多响应响应对象中不含 JSON-RPCid存放在内部asyncio.Queue实例中与一对一响应隔离。PersistentConnectionProvider在内部查找一对多响应时期望监听任务把这些消息存入该队列。由于消息顺序很重要该队列是 FIFO 队列PersistentConnection的process_subscriptions()方法按 FIFO 从队列弹出消息并暴露为异步迭代器模式。从源码看RequestProcessor.cache_raw_response()request_processor.py会把带 handler 的订阅消息放入_handler_subscription_queue无 handler 的放入_subscription_response_queue_persistent_message_stream()manager.py则从队列弹出并经_process_response()格式化后 yield。如果 socket 的消息流没有被其他任务中断队列一般会与 socket 上到来的消息保持同步监听者把消息放入队列process_subscriptions()从队列弹出该消息并把循环控制权交回监听者如此往复直到 socket 连接关闭或用户退订。如果消息流稍有滞后或 Provider 订阅了订阅却未消费消息内部队列可能填满至最大容量随后触发等待中的asyncio.Event_listen_event直到 Provider 重新开始消费队列消息。因此一旦发起订阅就应尽快通过process_subscriptions()开始消费队列消息。小结web3.py 的三层架构把“通信”Provider、“拦截/改造”Middleware与“调度/同步”Manager解耦请求自 Manager 发起逐层穿过 Middleware 洋葱到达 Provider响应再原路返回。对高级使用者而言最有价值的可定制点集中在 Provider 层——通过make_request/is_connected编写自定义 Provider、通过cache_allowed_requests/cacheable_requests/request_cache_validation_threshold精细控制请求缓存、通过ExceptionRetryConfiguration调整 HTTP 重试策略而在持久连接WebSocket/IPC场景下RequestProcessor以请求/响应id匹配为核心配合RequestInformation缓存、FIFO 订阅队列与订阅管理器 API实现了透明的一对一与一对多订阅消息处理。深入理解这些内部机制能帮助你在不触碰公开 API 的前提下安全地为特定场景定制 web3.py 的行为。赞分享Web3区块链【免费下载链接】web3.pyA python interface for interacting with the Ethereum blockchain and ecosystem.项目地址https://gitcode.com/gh_mirrors/we/web3.py点击查看免费下载相关推荐web3.py Provider 连接指南HTTP、IPC、WebSocket 与持久化连接全解析web3.py Provider 连接指南HTTP、IPC、WebSocket 与持久化连接全解析 导读 以太坊应用开发的第一步是让 web3.py httWeb3区块链web3.py 持久连接 Provider 请求缓存修复WebSocketProvider 与 AsyncIPCProvider 缓存失效问题解析web3.py 持久连接 Provider 请求缓存修复WebSocketProvider 与 AsyncIPCProvider 缓存失效问题解析 导读 本文Web3区块链Guzzle请求处理机制深入理解Handlers与MiddlewareGuzzle请求处理机制深入理解Handlers与Middleware 概述 Guzzle作为PHP中最流行的HTTP客户端之一其核心功能建立在handle后端上一篇Repowise 在 CI 中的实践用四个增量门禁守护 Pull Request 的补丁质量下一篇LunaTranslator游戏翻译工具实时视觉小说翻译的完整解决方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Web3区块链【免费下载链接】web3.pyA python interface for interacting with the Ethereum blockchain and ecosystem.项目地址https://gitcode.com/gh_mirrors/we/web3.py点击查看免费下载web3.py 在公开 APIWeb3/AsyncWeb3对象与所连接的节点之间存在多层抽象Provider 负责与区块链的底层通信Middleware 提供请求/响应的拦截钩子Manager 则承担线程安全与异步原语。本文以官方文档 docs/internals.rst 为核心骨架结合仓库源码与测试逐层拆解请求生命周期、请求缓存Request Caching、HTTP 异常重试以及WebSocketProvider/AsyncIPCProvider持久连接下的请求-响应匹配与订阅处理帮助高级用户理解并定制这些底层 API。⚠️ 适用提示本文涉及的是面向高级用户的内层 API。如果你不清楚自己在做什么建议停留在公开 API如w3.eth、w3.contract层面。请求生命周期一次 RPC 调用的完整旅程每一个 web3 RPC 调用都会依次穿过以下三层*********** ************ | Request | | Response | *********** ************ | ^ v | ----------------------------- | Manager | ----------------------------- | ^ v | ----------------------------- | Middleware | ----------------------------- | ^ v | ----------------------------- | Provider | -----------------------------可以用“洋葱”来理解这个关系Provider 在最中心请求从最外层的 Manager 发起逐层穿过每一层 Middleware最终抵达中心的 ProviderProvider 处理完后响应再从洋葱中心逐层向外传出最终由 Manager 返回给调用者。从源码看这一流程的实现位于 RequestManager_make_request()调用provider.request_func()后者在 BaseProvider.request_func() 中把整个MiddlewareOnion与self.make_request组合combine_middleware再逐层包装出最终的请求函数同步路径经request_blocking()返回异步路径经coro_request()返回最终都会调用 formatted_response() 做 JSON-RPC 结果校验与格式化。异步场景下请求与响应分离send()/recv()详见下文“持久连接”一节。利用这套分层可以实现的常见场景包括把某些 RPC 请求重定向到不同的 Provider例如所有读操作发给远程节点所有写操作发给本地自控节点透明拦截eth_sendTransaction发送的交易在本地签名后改走eth_sendRawTransaction发送修改 RPC 响应格式例如把响应中的整数值统一转换为十六进制校验 RPC 请求的输入参数。Providers直接与区块链交互的层Provider 负责所有与区块链的直接交互。大多数情况下这意味着通过 HTTP 或 IPC socket 与以太坊节点的 JSON-RPC 服务通信但 Provider 并不强制要求基于 RPC例如测试场景可以基于内存 EVMin-memory EVM实现一个 Provider 来满足请求仓库中的 eth-tester 相关实现即属此类。编写自定义 Provider自定义 Provider 需要实现两个必需方法并设置该 Provider 使用的 MiddlewareBaseProvider.make_request(method, params)每个 Provider 类必须实现此方法。它应当返回一个 JSON 对象成功时带result键失败时带error键。method被调用的 JSON-RPC 方法名字符串如eth_sendTransactionparams该 JSON-RPC 方法的参数列表或其他可迭代对象。基类中的默认实现直接抛出NotImplementedError见 web3/providers/base.py。BaseProvider.is_connected(show_tracebackFalse)根据 Provider 是否应被视为“已连接”返回True/False。例如 IPC socket 类 Providersocket 打开返回True关闭返回False若传入show_tracebackTrue则在不应视为已连接时抛出ProviderConnectionError并给出原因。参考实现JSONBaseProvider.is_connected()web3/providers/base.py会发一次web3_clientVersion请求来探测连接。BaseProvider.middleware应为可迭代的 Middleware 集合。可以通过给provider.middleware赋值来设置新的 Middleware 列表列表开头的第一个 Middleware 最先处理请求。此外JSONBaseProvider还提供了encode_rpc_request/decode_rpc_response见 web3/providers/base.py负责把请求封装为 JSON-RPC 2.0 格式自动生成递增的id以及解析原始字节响应自定义 HTTP/IPC 类 Provider 可复用这些工具。Provider 配置请求缓存Request Caching⚠️ 重要在启用请求缓存之前请先熟悉其校验逻辑。该特性为了尽力保证数据有效性常常需要在底层发起额外的请求可能为你的场景引入不必要的开销。可以设置request_cache_validation_threshold为None来关闭校验缓存所有允许缓存的请求也可以按需调整以匹配你的性能诉求。请求缓存通过 Provider 实例上的以下配置项启用与调整cache_allowed_requests: bool False总开关默认关闭cacheable_requests: Optional[Set[RPCEndpoint]]允许被缓存的 RPC 端点集合request_cache_validation_threshold: Optional[Union[RequestCacheValidationThreshold, int]]缓存校验阈值。对不依赖区块数据的请求如eth_chainId把cache_allowed_requests设为True即可安全地缓存所有响应。但对依赖区块数据的请求如eth_getBlockByNumber不能总是缓存——区块数据会变化例如链重组reorg期间或尚未达成最终性finality时。request_cache_validation_threshold即为这类请求配置一个安全的缓存阈值默认情况下该选项被配置为对你所连接链的 chain id 而言“安全”的内部值连接以太坊主网时取finalized已最终确定区块号连接其他链时取一个从当前时刻算起、对该链最终性机制“安全”的时间间隔秒。对不在内部列表中的链包括所有测试网默认值为 1 小时。内部配置的链及其默认阈值如下源码见 CHAIN_VALIDATION_THRESHOLD_DEFAULTS链默认校验阈值ETHRequestCacheValidationThreshold.FINALIZEDfinalized 区块ARB17 天ZKSYNC1 小时OETH3 分钟MATIC30 分钟ZKEVM1 小时BASE7 天SCR1 小时GNO5 分钟AVAX2 分钟BNB2 分钟FTM1 分钟以以太坊主网为例请求所依赖的区块号小于等于finalized区块号时响应会被缓存超过finalized区块号则不缓存。对其他链当请求所依赖数据的区块时间戳早于等于该链配置的时间间隔时缓存响应。可以通过三种方式修改该行为见 BaseProvider 的构造参数设为RequestCacheValidationThreshold.SAFE以safe区块作为阈值仅以太坊主网设为自定义秒数任何链包括以太坊主网设为None关闭校验、缓存所有请求对非测试网链不推荐。RequestCacheValidationThreshold枚举主网finalized与safe两个值从web3.utils模块导入见 web3/utils/caching.py。另外注意cacheable_requests用于限定允许缓存的端点集合默认是一个内部维护的“安全可缓存”列表排除了像eth_call这种响应可变、不宜缓存的端点。默认的可缓存请求列表如下加粗者为受request_cache_validation_threshold校验的请求eth_chainIdweb3_clientVersionnet_versioneth_getBlockByNumbereth_getRawTransactionByBlockNumberAndIndexeth_getBlockTransactionCountByNumbereth_getUncleByBlockNumberAndIndexeth_getUncleCountByBlockNumbereth_getBlockByHasheth_getTransactionByHasheth_getTransactionByBlockNumberAndIndexeth_getTransactionByBlockHashAndIndexeth_getBlockTransactionCountByHasheth_getRawTransactionByBlockHashAndIndexeth_getUncleByBlockHashAndIndexeth_getUncleCountByBlockHash从实现看该默认列表由 INTERNAL_VALIDATION_MAP 的键构成CACHEABLE_REQUESTS并按端点类型分为三类校验器参数含区块号validate_from_block_id_in_params、结果含区块信息validate_from_blocknum_in_result、参数含区块哈希validate_from_blockhash_in_params异步场景有对应实现request_caching_validation.py。值得注意的是这些校验器在必要时会额外发起 RPC 请求来获取区块信息例如验证一笔交易是否已越过安全阈值高频率请求场景下可能造成明显开销此外校验期间会临时关闭缓存以防递归见 is_beyond_validation_threshold 的cache_allowed_requests False处理。因此理解何时开缓存、如何配置校验是避免不必要开销的关键。配置示例from web3 import Web3, HTTPProvider from web3.utils import RequestCacheValidationThreshold w3 Web3(HTTPProvider( endpoint_uri..., # 可选开启请求缓存默认 False cache_allowed_requestsTrue, # 可选允许缓存的端点集合默认是内部“安全可缓存”列表见上 cacheable_requests{eth_chainId, eth_getBlockByNumber}, # 可选默认根据 chain id 取值见上 request_cache_validation_threshold60 * 60, # 1 小时 # request_cache_validation_thresholdRequestCacheValidationThreshold.SAFE, # 仅以太坊主网 ))Provider 配置HTTP Provider 请求重试HTTPProvider与AsyncHTTPProvider默认会对某些请求在异常时进行重试。通过 Provider 实例上的exception_retry_configuration属性配置其值为 ExceptionRetryConfiguration一个 pydanticBaseModel实例。重试机制采用指数退避策略以backoff_factor决定的初始值为起点每次重试间隔翻倍最多尝试retries次。配置项说明errorsProvider 应重试的异常元组。默认值HTTPProvider为(ConnectionError, requests.HTTPError, requests.Timeout)AsyncHTTPProvider为(aiohttp.ClientError, asyncio.TimeoutError)。两种 Provider 的DEFAULT_EXCEPTIONS定义见下。retries重试次数默认 5。backoff_factor初始延迟倍数每次重试翻倍默认 0.125。method_allowlist允许重试的方法列表默认是内部“安全可重试”方法清单REQUEST_RETRY_ALLOWLIST见 web3/providers/rpc/utils.py包含admin、net、txpool、testing、evm前缀方法以及eth_call、eth_sendRawTransaction等读/写方法。示例展示默认选项及如何覆盖from web3 import Web3, HTTPProvider from web3.providers.rpc.utils import ( REQUEST_RETRY_ALLOWLIST, ExceptionRetryConfiguration, ) w3 Web3(HTTPProvider( endpoint_uri..., exception_retry_configurationExceptionRetryConfiguration( errorsDEFAULT_EXCEPTIONS, # 重试次数 retries5, # 初始延迟倍数每次重试翻倍 backoff_factor0.125, # 内部默认的可重试方法清单 method_allowlistREQUEST_RETRY_ALLOWLIST, ), ))不同 HTTP Provider 的DEFAULT_EXCEPTIONS定义为HTTPProvider(ConnectionError, requests.HTTPError, requests.Timeout)AsyncHTTPProvider(ConnectionError, aiohttp.ClientError, asyncio.TimeoutError)把retry_configuration即exception_retry_configuration设为None将关闭该 Provider 实例的异常重试from web3 import Web3, HTTPProvider w3 Web3(HTTPProvider(endpoint_uri..., retry_configurationNone)从实现看重试逻辑位于 HTTPProvider._make_request() 与AsyncHTTPProvider的对应方法仅当配置非空且check_if_retry_on_failure()web3/providers/rpc/utils.py判定方法在允许列表中时执行重试每次捕获errors中的异常后按backoff_factor * 2**i计算退避延迟见 async_rpc.py。仓库在 tests/core/providers/test_http_request_retry.py 中对重试次数、异常类型与配置等价性均有覆盖。Managers请求/响应生命周期的守门人Manager 是请求/响应生命周期的守门人gatekeeper。大部分功能可以在 Middleware 层实现因此一般无需更换 Manager。RequestManager在初始化时会装配默认 Middleware 洋葱get_default_middlewaregas_price_strategy、ens_name_to_address、attrdict、validation、gas_estimate并提供request_blocking/coro_request两个同步/异步请求入口。持久连接 Provider 的请求处理RequestProcessorRequestProcessor 负责为PersistentConnectionProvider存储并同步异步请求与其响应。WebSocketProvider 与AsyncIPCProvider是两个持久连接 Provider。由于要通过同一个 socket 发送请求并接收对应响应PersistentConnectionProvider必须把请求的id与 socket 返回的响应的id匹配起来。任何不按 JSON-RPC 2.0 规范携带 id 的 Provider 都无法与PersistentConnectionProvider一起工作。从实现看RequestProcessor内部维护了三个独立数据结构request_processor.py_request_information_cache请求信息缓存容量默认 500、_request_response_cache一对一响应缓存容量 500、_subscription_response_queue一对多订阅队列FIFO容量默认 500类型为TaskReliantQueue。监听响应PersistentConnectionProvider的实现类在 socket 连接建立时会启动一个消息监听后台任务_message_listener_task见 persistent.py 的connect()。该任务负责监听 socket 连接上到来的所有消息并存入 Provider 实例内部的RequestProcessor。RequestProcessor根据消息是否带 JSON-RPCid值把消息存到不同的缓存一对一缓存或一对多订阅队列。一对一请求一对一请求指期望只收到一个响应的请求例如通过eth模块 API 获取最新区块号 async def ws_one_to_one_example(): ... async with AsyncWeb3(WebSocketProvider(fws://127.0.0.1:8546)) as w3: ... # 发起请求并在同一行内收到单个响应 ... latest_block_num await w3.eth.block_number asyncio.run(ws_one_to_one_example())持久 socket 连接下我们只能调用send()再通过其他方式一般是recv()或遍历 socket 消息异步接收响应。正因收发是异步的为了把响应匹配回原始请求必须把请求信息存到某处即与收到的响应id匹配的那个请求。存储的请求信息随后用于处理响应把它送入 web3.py 库内部的响应格式化器response formatters与中间件处理管线。存储请求信息的载体是RequestProcessor内部的RequestInformation缓存类其保存的字段如下源码见 RequestInformationmethod方法名如eth_subscribeparams发起调用时的参数如(newPendingTransactions, True)response_formatters用于处理响应的格式化器middleware_response_processors发起请求时实例上存在的、处理响应的中间件处理器按顺序追加响应到来时依序经过这些逻辑subscription_id若请求是eth_subscribe收到订阅调用的响应即订阅id时不是把这条信息从缓存弹出而是把订阅 id 一并保存到请求信息中以便正确后续处理所有携带该订阅id的订阅消息一对一请求场景该值恒为None。一对一响应响应对象中含 JSON-RPCid存放在内部SimpleCache实例中与任何一对多响应隔离。PersistentConnectionProvider在内部查找响应时期望监听任务把响应存入该缓存由于缓存键的生成使用了请求id它会查找与响应id匹配请求id的缓存键找到则处理并返回给用户找不到则操作超时并抛出TimeExhausted异常。该超时可通过实例化PersistentConnectionProvider时的response_timeout关键字参数配置对应源码中的request_timeout默认 30 秒见 persistent.py。相关缓存键生成逻辑见 generate_cache_key。一对多请求订阅一对多请求指一次初始请求后期望收到多条响应的请求目前唯一的例子是eth_subscribe。初始的eth_subscribe请求本身只期待一个响应——订阅id值但成功后还期待收到很多eth_subscription消息。因此原始请求被视为一对一请求以便在同一行把订阅id返回给用户该调用后续产生的多条响应可以按以下几种方式之一处理。方式一订阅管理器 API推荐订阅管理器 API 是AsyncWeb3类上的公开 API当连接的是PersistentConnectionProvider实例时可用允许用户订阅一个订阅并异步处理其多条响应。只要每次订阅调用都传入 handlersubscription_manager实例就负责处理 socket 连接上到来的多条响应用户用完订阅后也可以用它退订。 async def new_heads_handler( ... handler_context: NewHeadsSubscriptionContext, ... ) - None: ... result handler_context.result ... print(fNew block header: {result}\n) ... if result[number] 1234567: ... await handler_context.subscription.unsubscribe() async def ws_subscription_example(): ... async with AsyncWeb3(WebSocketProvider(fws://127.0.0.1:8546)) as w3: ... # 订阅新区块头并拿到 subscription_id。 ... # 这是一次一对一调用附带大量响应的触发器 ... subscription_id await w3.eth.subscribe(newHeads, handlernew_heads_handler) ... ... # 用订阅管理器异步处理订阅消息。 ... # 会一直运行到订阅管理器中不再有订阅为止 ... # 若把 run_forever 置为 True 则无限运行。 ... await w3.subscription_manager.handle_subscriptions(run_foreverFalse) asyncio.run(ws_subscription_example())订阅管理器还可以一次性订阅多个订阅。EthSubscription系列类可从web3.utils.subscriptions导入提供了友好的订阅管理 API。由于每个连接与 Provider 实例都有各自的监听任务与订阅管理器实例你可以一次订阅多个订阅并通过 handler 处理多条响应。handler 中包含async_w3发起该订阅的AsyncWeb3实例subscriptionhandler 所依附的订阅实例resultsocket 连接上到来的、属于该订阅的响应。订阅还接受handler_context参数可在订阅时向 handler 传递额外信息例如传入一个事件对象用于在日志事件到来时解析它。 from web3 import ( ... AsyncWeb3, ... WebSocketProvider, ... AsyncIPCProvider, ... ) from web3.utils.subscriptions import ( ... EthSubscription, ... NewHeadsSubscription, ... NewHeadsSubscriptionContext, ... PendingTxSubscription, ... PendingTxSubscriptionContext, ... LogsSubscription, ... LogsSubscriptionContext, ... ) async def new_heads_handler( ... handler_context: NewHeadsSubscriptionContext, ... ) - None: ... header handler_context.result ... print(fNew block header: {header}\n) ... if header[number] 1234567: ... await handler_context.subscription.unsubscribe() async def pending_txs_handler( ... handler_context: PendingTxSubscriptionContext, ... ) - None: ... ... async def log_handler( ... handler_context: LogsSubscriptionContext, ... ) - None: ... log_receipt handler_context.result ... # 订阅日志时我们把事件传入了 handler_context现在可直接取用 ... event_data handler_context.transfer_event.process_log(log_receipt) ... print(fLog event data: {event_data}\n) async def sub_manager(): ... local_w3 await AsyncWeb3(AsyncIPCProvider(LOCAL_IPC)) ... ... # 通过订阅管理器 handler 一次订阅多个订阅 ... weth_contract local_w3.eth.contract( ... addresslocal_w3.to_checksum_address(0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2), ... abiWETH_ABI, ... ) ... transfer_event weth_contract.events.Transfer() ... await local_w3.subscription_manager.subscribe( ... [ ... NewHeadsSubscription(labelnew-heads-mainnet, handlernew_heads_handler), ... PendingTxSubscription( ... labelpending-tx-mainnet, # 可选 label ... full_transactionsTrue, ... handlerpending_tx_handler, ... ), ... LogsSubscription( ... labelWETH transfers, # 可选 label ... addressweth_contract.address, ... topics[transfer_event.topic], ... handlerlog_handler, ... # 可选的 handler_context 参数帮助解析响应 ... handler_context{transfer_event: transfer_event}, ... ), ... ] ... ) ... ... public_w3 await AsyncWeb3(WebSocketProvider(PUBLIC_PROVIDER_WS)) ... # 通过 eth_subscribe 订阅带 handler 与可选 label ... await public_w3.eth.subscribe(public_newHeads, handlerpending_tx_handler, labelnew-heads-public-ws) # 处理所有订阅直到两个订阅管理器中都不再有订阅。 ... # 若任一管理器的 run_forever 为 True则无限运行。 await asyncio.gather( ... public_w3.subscription_manager.handle_subscriptions(), ... local_w3.subscription_manager.handle_subscriptions(), ... ) ... ... # 关闭连接 ... await local_w3.provider.disconnect() ... await public_w3.provider.disconnect() asyncio.run(sub_manager())说明订阅系列类如 NewHeadsSubscription、PendingTxSubscription、LogsSubscription、SyncingSubscription的构造参数在源码中均有类型注解与默认值例如PendingTxSubscription(full_transactionsFalse)、LogsSubscription(addressNone, topicsNone)且各订阅都支持可选的label、handler_context与parallelize参数w3.eth.subscribe()会按订阅类型自动创建对应的类型化订阅见 async_eth.py 的 subscribe() 与 _create_type_aware_subscription。方式二process_subscriptions()异步迭代器PersistentConnection类web3/providers/persistent/persistent_connection.py是操作活动持久 socket 连接的公开 API其process_subscriptions()方法按异步迭代器模式接收eth_subscription响应。你可以用它在消息到来时监听并处理原始消息。 async def ws_subscription_example(): ... async with AsyncWeb3(WebSocketProvider(fws://127.0.0.1:8546)) as w3: ... # 订阅新区块头并拿到 subscription_id。 ... # 这是一次一对一调用附带大量响应的触发器 ... subscription_id await w3.eth.subscribe(newHeads) ... ... # 利用 w3.socketPersistentConnection 公开 API的 ... # process_subscriptions() 方法监听 socket 上的大量响应 ... async for response in w3.socket.process_subscriptions(): ... # 这里只接收一对多响应避免把一对一请求的响应 ... # 意外地带进这个代码块 ... ... print(f{response}\n) ... ... if some_condition: ... # 退订新区块头这是另一次一对一请求 ... is_unsubscribed await w3.eth.unsubscribe(subscription_id) ... if is_unsubscribed: ... break asyncio.run(ws_subscription_example())一对多响应响应对象中不含 JSON-RPCid存放在内部asyncio.Queue实例中与一对一响应隔离。PersistentConnectionProvider在内部查找一对多响应时期望监听任务把这些消息存入该队列。由于消息顺序很重要该队列是 FIFO 队列PersistentConnection的process_subscriptions()方法按 FIFO 从队列弹出消息并暴露为异步迭代器模式。从源码看RequestProcessor.cache_raw_response()request_processor.py会把带 handler 的订阅消息放入_handler_subscription_queue无 handler 的放入_subscription_response_queue_persistent_message_stream()manager.py则从队列弹出并经_process_response()格式化后 yield。如果 socket 的消息流没有被其他任务中断队列一般会与 socket 上到来的消息保持同步监听者把消息放入队列process_subscriptions()从队列弹出该消息并把循环控制权交回监听者如此往复直到 socket 连接关闭或用户退订。如果消息流稍有滞后或 Provider 订阅了订阅却未消费消息内部队列可能填满至最大容量随后触发等待中的asyncio.Event_listen_event直到 Provider 重新开始消费队列消息。因此一旦发起订阅就应尽快通过process_subscriptions()开始消费队列消息。小结web3.py 的三层架构把“通信”Provider、“拦截/改造”Middleware与“调度/同步”Manager解耦请求自 Manager 发起逐层穿过 Middleware 洋葱到达 Provider响应再原路返回。对高级使用者而言最有价值的可定制点集中在 Provider 层——通过make_request/is_connected编写自定义 Provider、通过cache_allowed_requests/cacheable_requests/request_cache_validation_threshold精细控制请求缓存、通过ExceptionRetryConfiguration调整 HTTP 重试策略而在持久连接WebSocket/IPC场景下RequestProcessor以请求/响应id匹配为核心配合RequestInformation缓存、FIFO 订阅队列与订阅管理器 API实现了透明的一对一与一对多订阅消息处理。深入理解这些内部机制能帮助你在不触碰公开 API 的前提下安全地为特定场景定制 web3.py 的行为。赞分享Web3区块链【免费下载链接】web3.pyA python interface for interacting with the Ethereum blockchain and ecosystem.项目地址https://gitcode.com/gh_mirrors/we/web3.py点击查看免费下载相关推荐web3.py Provider 连接指南HTTP、IPC、WebSocket 与持久化连接全解析web3.py Provider 连接指南HTTP、IPC、WebSocket 与持久化连接全解析 导读 以太坊应用开发的第一步是让 web3.py httWeb3区块链web3.py 持久连接 Provider 请求缓存修复WebSocketProvider 与 AsyncIPCProvider 缓存失效问题解析web3.py 持久连接 Provider 请求缓存修复WebSocketProvider 与 AsyncIPCProvider 缓存失效问题解析 导读 本文Web3区块链Guzzle请求处理机制深入理解Handlers与MiddlewareGuzzle请求处理机制深入理解Handlers与Middleware 概述 Guzzle作为PHP中最流行的HTTP客户端之一其核心功能建立在handle后端上一篇Repowise 在 CI 中的实践用四个增量门禁守护 Pull Request 的补丁质量下一篇LunaTranslator游戏翻译工具实时视觉小说翻译的完整解决方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
![[从入门到入土!!!] Ubuntu20.04 系统开发环境配置（系统配置，显卡驱动，CUDA，ROS，VSCode，Anaconda等）](http://pic.xiahunao.cn/yaotu/[从入门到入土!!!] Ubuntu20.04 系统开发环境配置（系统配置，显卡驱动，CUDA，ROS，VSCode，Anaconda等）)
