为 redis-py 新增 Redis 命令支持从命令规范到双协议测试的完整实战指南【免费下载链接】redis-pyRedis Python client项目地址: https://gitcode.com/GitHub_Trending/re/redis-py本篇指南面向 redis-pyRedis Python client的贡献者与二次开发者系统讲解如何根据一份命令规范command specification为 redis-py 新增一个 Redis 命令的完整工程流程。你将掌握命令在 redis/commands 目录中的组织方式、同步/异步 API 的overload双签名实现规范、RESP2/RESP3 双协议下的响应回调设计以及基于 pytest 的集成测试与验证方法最终交付一个同时兼容 Standalone、Cluster、Sentinel 三类客户端且通过双协议测试的命令实现。一、写在动手之前流程总览与前置准备新增命令支持并非写一个方法这么简单redis-py 要求命令实现同时满足以下约束命令方法必须挂在对应的 trait 对象上供 StandaloneRedis、ClusterRedisCluster、SentinelSentinel客户端通过继承复用同步redis.client.Redis与异步redis.asyncio.client.Redis两条 API 必须同时具备并通过方法重载overload提供正确的类型提示参数与返回值必须正确处理 bytes 表示decode_responsesFalse场景响应形状必须兼容 RESP2 与 RESP3 两种线协议通过response_callbacks完成第二层解析。正式动手前需要完成两项准备阅读仓库协作约定关联流程文档要求先遵循.agents/instructions.md中的指引。需要注意截至本指南撰写时仓库.agents/目录下实际包含 add-new-command、sync-claude-md 两个命令目录以及 sync_async_type_hints_overload_guide.md 类型提示指南请以仓库实际存在的文件为准。通读命令开发指南specs/redis_commands_guide.md 是新增命令的纲领性文档其中包含命令 API 规范、文件结构、二进制表示、协议兼容性矩阵和参数定义约定下文会持续引用它。二、理解命令在 redis-py 中的组织方式2.1 目录结构按 specs/redis_commands_guide.md 的说明命令相关代码集中在redis/commands/基目录redis/ ├── commands/ # Base directory │ ├── module/ # Module commands directory │ │ ├── commands.py # Module commands public API │ │ └── file.py # Custom helpers │ │... # Other modules │ ├── __init__.py # Module exports │ ├── cluster.py # Cluster commands public API │ ├── core.py # Core commands public API │ ├── helpers.py # Helpers for commands modules │ ├── redismodules.py # Trait for all Redis modules │ └── sentinel.py # Sentinel commands public API对应到当前仓库可以看到完整的实际实现redis/commands/core.py核心命令公共 API定义了CoreCommands等 trait 类包含 string、list、set、hash、zset、stream、geo、pubsub、script、server 管理等全部核心命令redis/commands/cluster.py集群命令公共 API如cluster_addslots、cluster_nodes等其中部分核心命令如exists、delete、mget会在集群场景下重写为按 slot 分片执行redis/commands/sentinel.pySentinel 命令公共 APIsentinel_masters、sentinel_failover等redis/commands/redismodules.py所有 Redis 模块的 trait 入口通过json()、ft()、ts()、bf()等工厂方法返回模块命令对象redis/commands/helpers.py命令模块的公共助手函数。定位命令归属新增命令前先判断它属于哪一类——string/list/set/hash/zset/stream 等核心命令放入core.py集群管理类命令放入cluster.pySentinel 相关放入sentinel.py若命令属于某个 Redis 模块如 RediSearch、RedisJSON、TimeSeries、Bloom 系、VectorSet则放入 redis/commands/ 下对应的模块子目录search/、json/、timeseries/、bf/、vectorset/。2.2 公共 API 与底层执行链所有命令实现的最终落点都是execute_command(*args, **kwargs)由CommandProtocol定义。无论同步还是异步客户端命令方法只负责把用户友好的参数翻译成 Redis 友好的线协议参数列表然后交给execute_command完成编码、发送与响应解析。这也是参数定义一节中提到的在命令方法内部做转换的原因——比如scan把可选的COUNT选项拼接进pieces列表。三、第一步撰写/获取命令规范新增命令的起点是一份命令规范文件。规范的来源有两种官方文档已收录Redis 官方命令文档是所有命令相关信息的权威来源可直接引用官方文档尚未收录Redis 是活跃演进的项目新命令可能尚未出现在官方文档中。此时需要基于 command-specification-template.md 模板自行创建规范。模板定义了规范的五个核心板块这是新增命令工作的需求规格书板块内容示例Supported version命令所需的最低 Redis 版本Redis 6.2.0Command description命令的功能描述一句话说明命令做什么Command API采用官方文档格式的命令签名$COMMAND_NAME $key $member [NX\|XX] [CH] [INCR]Redis-CLI examples相关的 redis-cli 操作示例实际可执行的示例命令Test plan集成测试计划逐条列出测试场景与断言3.1 为什么 Test plan 要逐条写清楚规范中的 Test plan 直接决定后续测试代码的编写。模板给出的示例是- Test only with required arguments, assert that single value returned - Test with required arguments and optional XX modifier, ensure that 1 returned - ...每条测试计划应包含输入参数的组合与期望的返回值两部分这样实现者可以直接将其翻译为assert语句。四、第二步阅读理解规范拿到规范后按顺序完成以下阅读理解工作通读整个规范尤其是 Command Description先确认命令类型归属string / list / set / hash / zset / stream / module 等这决定了它应该被放进哪个 trait 对象、以及测试文件应该建在哪里梳理 Command API区分必需参数与可选参数optional arguments判断 Redis 命令参数类型如何映射为 Python 类型数字、字符串、字节、字典、列表等确定返回值类型及所有可能的响应类型单个值、数组、字典、None 等查看 Redis-CLI 示例示例既是理解命令语义的最佳途径也是编写集成测试断言的基础复核 Test plan确认测试场景是否覆盖了必需参数、可选修饰符、边界条件等。4.1 命令参数如何映射为 Python 类型以 redis/commands/core.py 中scan命令第 6061 行起的真实实现为例可以看到典型的参数映射与转换逻辑overload def scan( self: SyncClientProtocol, cursor: int 0, match: PatternT | None None, count: int | None None, _type: str | None None, **kwargs, ) - ScanResponse: ... overload def scan( self: AsyncClientProtocol, cursor: int 0, match: PatternT | None None, count: int | None None, _type: str | None None, **kwargs, ) - Awaitable[ScanResponse]: ... def scan( self, cursor: int 0, match: PatternT | None None, count: int | None None, _type: str | None None, **kwargs, ) - ScanResponse | Awaitable[ScanResponse]: Incrementally return lists of key names. Also return a cursor indicating the scan position. match allows for filtering the keys by pattern count provides a hint to Redis about the number of keys to return per batch. _type filters the returned values by a particular Redis type. Stock Redis instances allow for the following types: HASH, LIST, SET, STREAM, STRING, ZSET Additionally, Redis modules can expose other types as well. pieces: list[EncodableT] [cursor] if match is not None: # 将用户友好的 match 参数转换为 Redis 线协议中的 MATCH 选项 pieces.extend([bMATCH, match]) ...注意两个要点隐藏复杂度Redis 原生命令要求COUNT 后面跟数量这种固定语法但公共 API 直接以count: int | None None暴露由方法内部拼装pieces用户无需关心线协议细节。这也是 specs/redis_commands_guide.md 中Arguments definition一节强调的将 Redis 命令的复杂 API 结构转换为用户友好的形式。参数类型尽量宽泛对于输入参数优先使用最通用的抽象接口。规范中的指导原则是如果参数只会被迭代应标注为Iterable[...]而非list[...]以接受 list、tuple、set、generator 等任意可迭代输入只有实现确实需要索引、切片、排序保证或原地修改时才固定为list[...]。例如# Preferred: accepts any iterable of strings. def querylabels(self, label: str | None None, filters: Iterable[str] | None None): ... # Avoid unless a list is explicitly required (indexing, ordering, mutation): def querylabels(self, label: str | None None, filters: list[str] | None None): ...这一原则只适用于输入参数返回值类型提示则应反映命令实际产出的具体类型见下文协议兼容性部分。五、第三步按序实现命令5.1 定位任务涉及的代码文件先确认需要修改或新建的文件阅读相关联的既有实现如最接近的新命令同类命令作为参照。新增一个核心命令通常涉及redis/commands/core.py或对应的模块commands.py新增命令方法redis/typing.py若引入了新的响应结构需要补充类型别名redis/_parsers/response_callbacks.py或redis/commands/$module/注册响应回调测试文件tests/test_commands.py同步与tests/test_asyncio/test_commands.py异步。5.2 在 trait 对象中实现命令将命令方法添加至匹配的 trait 类例如核心命令放入 redis/commands/core.py 的CoreCommands。实现的关键约束是同步/异步 API 都要有通过overload实现双重签名参数与响应类型考虑 bytes 表示客户端可配置decode_responsesFalse此时响应为 bytes同时允许传 bytes 参数替代字符串通过response_callbacks保证 RESP2/RESP3 兼容。同步/异步 overload 三件套模板按照 .agents/sync_async_type_hints_overload_guide.md 的规范同时支持同步与异步客户端的命令使用如下模板from typing import Awaitable, overload from redis.typing import AsyncClientProtocol, SyncClientProtocol class SomeCommands: overload def method(self: SyncClientProtocol, arg: str | None None) - SyncReturnType: ... overload def method( self: AsyncClientProtocol, arg: str | None None ) - Awaitable[SyncReturnType]: ... def method( self, arg: str | None None ) - SyncReturnType | Awaitable[SyncReturnType]: # implementation ...要点两个overload签名必须完整镜像实现的输入签名——参数名、顺序、默认值、位置/关键字形态、*args、**kwargs全部保持一致使用现代注解风格X | Y和T | None而非Union[...]/Optional[...]实现体的返回注解为SyncType | Awaitable[SyncType]当同步侧本身是联合类型时用括号分组后再接异步分支如(SentinelMastersResponse | bool) | Awaitable[SentinelMastersResponse | bool]运行时装饰器deprecation、experimental 等必须放在真实实现上而不是overload存根上若同步与异步客户端返回相同的具体类型只需保留单个方法不必添加冗余 overload。判别依据redis/typing.py中定义了基于Literal标记的 Protocolclass SyncClientProtocol(Protocol): Marker for sync clients. _is_async_client: Literal[False] class AsyncClientProtocol(Protocol): Marker for async clients. _is_async_client: Literal[True]采用 Protocol 判别而非Redis[bytes]/AsyncRedis[str]是为了避免因decode_responses设置而倍增 overload 数量。类型别名体系输入参数类型应优先复用 redis/typing.py 中预定义的类型别名别名含义定义KeyT主 Redis 键空间bytes \| str \| memoryviewFieldThash 表、stream、geo 命令中的字段EncodableTEncodableT任意可编码值bytes \| bytearray \| memoryview \| str \| int \| floatExpiryT相对过期时长int \| timedeltaAbsExpiryT绝对过期时间int \| datetimePatternT匹配键/字段的模式bytes \| str \| memoryviewKeysT键或键的迭代KeyT \| Iterable[KeyT]如果命令引入了新的响应结构如新的 dict 形状、元组组合应在redis/typing.py中补充命名类型别名并确保公共命令的返回类型提示包含这些新结构。5.3 协议兼容性RESP2 与 RESP3 的双层解析redis-py 8.0 起客户端默认在线上使用 RESP3 协议同时legacy_responsesTrue保留与 RESP2 兼容的 Python 响应形状。线协议与 Python 响应形状是独立选择的specs/redis_commands_guide.md 给出了完整的组合矩阵Redis()线上 RESP3 传统 RESP2 兼容的 Python 响应形状Redis(protocol2)线上 RESP2 传统 RESP2 Python 响应形状Redis(protocol3)线上 RESP3 传统 RESP3 Python 响应形状Redis(legacy_responsesFalse)启用统一响应形状unified shapes必须与协议无关除非命令存在官方文档记载的服务器端差异否则 RESP2 与 RESP3 下形状应一致。新增或修改命令时需要为以下每一种受支持模式确认期望的响应形状RESP2 legacyprotocol2、legacy_responsesTrueRESP3 legacyprotocol3、legacy_responsesTruedefault legacyprotocol未设置、legacy_responsesTrueRESP3 线协议 RESP2 兼容 Python 形状RESP2 unifiedprotocol2、legacy_responsesFalseRESP3 unifiedprotocol3、legacy_responsesFalseRESP 线协议解析由 redis/_parsers/resp2.py 与 redis/_parsers/resp3.py 完成而协议/响应形状兼容性则通过第二层解析——响应回调response callbacks实现核心命令回调定义在 redis/_parsers/response_callbacks.py模块专属回调映射表位于 redis/commands 下各模块目录。回调注册规则根据 .agents/sync_async_type_hints_overload_guide.md 的说明核心命令在 redis/_parsers/helpers.py 的三层回调字典中注册_RedisCallbacksRESP2 与 RESP3 共享的基础回调_RedisCallbacksRESP2RESP2 专属的覆盖/新增回调_RedisCallbacksRESP3RESP3 专属的覆盖/新增回调。模块命令在模块的__init__.py中注册_MODULE_CALLBACKS共享回调_RESP2_MODULE_CALLBACKS/_RESP3_MODULE_CALLBACKS各协议专属回调。回调查找顺序RESP2 下先查_RedisCallbacksRESP2未命中再回退_RedisCallbacksRESP3 同理。若没有注册任何回调则返回原始响应形状取决于decode_responses设置。协议相关返回类型差异是常见陷阱。指南中列出的典型差异包括命令RESP2 返回RESP3 返回原因acl_catlist[str]list[bytes \| str]RESP2 有str_if_bytesRESP3 无回调geohashlist[str]list[bytes \| str]同上hgetalldictpairs_to_dictdictidentity解析逻辑不同zincrby/zscorefloatfloat_or_nonefloatrawRESP2 解析、RESP3 直接返回类型提示策略对存在协议差异的命令采用最宽松的联合类型——若 RESP2 返回str而 RESP3 返回bytes | str则公共注解用bytes | str保证任何协议下都类型安全。兼容性原则legacy 回调应保留既有行为除非改动有意修复 bugunified 回调应提供推荐的、与协议无关的公共形状。5.4 边实现边验证每修改一个文件就立刻检查语法是否正确无缩进/括号错误import 是否正确新增的类型别名、回调、Protocol 等是否已导入类型是否正确定义overload与实现的签名是否完全镜像RESP2 与 RESP3 的响应 schema 是否保持一致或有意识地按文档差异处理。六、第四步实现测试计划实现完成后进入测试环节定位测试文件命令测试位于tests/test_*command_type*.py异步命令位于tests/test_asyncio/test_*command_type*.py。例如核心字符串命令对应 tests/test_commands.py 与 tests/test_asyncio/test_commands.py。因此第一步就识别命令类型至关重要它直接决定测试文件位置。若不存在匹配的测试文件则新建一个。每个测试用例一个独立测试方法把规范 Test plan 中的每一条翻译为一个测试方法每个方法只验证一个场景。版本约束若规范中声明了版本要求通过 tests/conftest.py 中定义的自定义注解skip_if_server_version_lt()与skip_if_server_version_gte()控制跳过。带有版本约束的命令通常只做集成测试。覆盖边界情况除规范 Test plan 列出的场景外补充空输入、None 返回、bytes/str 双形态等边界断言。七、第五步运行测试并保证双协议通过运行新增测试用例时使用 pytest 并显式指定协议# 以 RESP2 协议运行 pytest tests/test_commands.py -k test_your_new_command --protocol 2 # 以 RESP3 协议运行 pytest tests/test_commands.py -k test_your_new_command --protocol 3 # 异步版本同理 pytest tests/test_asyncio/test_commands.py -k test_your_new_command --protocol 2 pytest tests/test_asyncio/test_commands.py -k test_your_new_command --protocol 3--protocol是 tests/conftest.py第 123-128 行注册的 pytest 选项默认值来自default_protocol用于指定线协议版本同文件还注册了--legacy-responses选项可强制覆盖测试客户端的legacy_responses参数true/false/default三种取值。如需同时验证统一响应形状可组合使用pytest tests/test_commands.py -k test_your_new_command --protocol 3 --legacy-responses false验收标准同一组测试用例必须在 RESP2 与 RESP3 下全部通过。若任一协议下失败回到实现阶段修复重点检查响应回调注册与协议差异处理直到双协议绿。八、第六步最终验证清单提交前逐项确认计划中的全部任务已完成所有测试已创建且通过RESP2 与 RESP3 双协议代码符合项目约定overload 三件套、类型别名复用、现代注解风格、回调分层注册必要的文档已新增或更新如涉及新模块或新结构同步更新相关文档。九、输出报告完成后按要求输出工作总结包含Completed Tasks已完成任务清单创建的文件带路径修改的文件带路径。Tests Added创建的测试文件实现的测试用例测试结果双协议通过情况。十、常见问题速查Q新命令返回结构复杂类型提示怎么写先在 redis/typing.py 中定义命名类型别名再让 sync 与 async 两个 overload 以及实现体都引用该别名避免三处重复书写长联合类型。QRESP2 下测试通过、RESP3 下失败通常是什么原因最常见的是未注册 RESP3 专属回调或反之导致不同协议下返回原始形状不一致。检查 redis/_parsers/helpers.py 与 redis/_parsers/response_callbacks.py 中_RedisCallbacksRESP3的注册情况并对照协议返回类型差异表确认是否属于有意的协议差异。Q命令需要按服务器版本跳过测试怎么处理使用 tests/conftest.py 中的skip_if_server_version_lt(x.y.z)装饰器并在测试方法上声明。规范中的 Supported version 板块就是这里的依据。Q模块命令如 RediSearch、RedisJSON与核心命令的实现有何不同模块命令放入 redis/commands 下对应模块子目录回调注册在模块自身的__init__.py_MODULE_CALLBACKS/_RESP2_MODULE_CALLBACKS/_RESP3_MODULE_CALLBACKS并通过 redis/commands/redismodules.py 的工厂方法ft()、json()、ts()等暴露其余 overload、类型别名、双协议测试流程与核心命令完全一致。通过以上流程你可以系统地为 redis-py 贡献新命令支持并确保其与项目既有的同步/异步双 API、RESP2/RESP3 双协议架构保持一致。进一步的实现细节可继续研读 specs/redis_commands_guide.md 与 .agents/sync_async_type_hints_overload_guide.md 两份项目内部规范文档。【免费下载链接】redis-pyRedis Python client项目地址: https://gitcode.com/GitHub_Trending/re/redis-py创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/14 16:16:54 · 阅读 1.2k 为 redis-py 新增 Redis 命令支持：从命令规范到双协议测试的完整实战指南
为 redis-py 新增 Redis 命令支持从命令规范到双协议测试的完整实战指南【免费下载链接】redis-pyRedis Python client项目地址: https://gitcode.com/GitHub_Trending/re/redis-py本篇指南面向 redis-pyRedis Python client的贡献者与二次开发者系统讲解如何根据一份命令规范command specification为 redis-py 新增一个 Redis 命令的完整工程流程。你将掌握命令在 redis/commands 目录中的组织方式、同步/异步 API 的overload双签名实现规范、RESP2/RESP3 双协议下的响应回调设计以及基于 pytest 的集成测试与验证方法最终交付一个同时兼容 Standalone、Cluster、Sentinel 三类客户端且通过双协议测试的命令实现。一、写在动手之前流程总览与前置准备新增命令支持并非写一个方法这么简单redis-py 要求命令实现同时满足以下约束命令方法必须挂在对应的 trait 对象上供 StandaloneRedis、ClusterRedisCluster、SentinelSentinel客户端通过继承复用同步redis.client.Redis与异步redis.asyncio.client.Redis两条 API 必须同时具备并通过方法重载overload提供正确的类型提示参数与返回值必须正确处理 bytes 表示decode_responsesFalse场景响应形状必须兼容 RESP2 与 RESP3 两种线协议通过response_callbacks完成第二层解析。正式动手前需要完成两项准备阅读仓库协作约定关联流程文档要求先遵循.agents/instructions.md中的指引。需要注意截至本指南撰写时仓库.agents/目录下实际包含 add-new-command、sync-claude-md 两个命令目录以及 sync_async_type_hints_overload_guide.md 类型提示指南请以仓库实际存在的文件为准。通读命令开发指南specs/redis_commands_guide.md 是新增命令的纲领性文档其中包含命令 API 规范、文件结构、二进制表示、协议兼容性矩阵和参数定义约定下文会持续引用它。二、理解命令在 redis-py 中的组织方式2.1 目录结构按 specs/redis_commands_guide.md 的说明命令相关代码集中在redis/commands/基目录redis/ ├── commands/ # Base directory │ ├── module/ # Module commands directory │ │ ├── commands.py # Module commands public API │ │ └── file.py # Custom helpers │ │... # Other modules │ ├── __init__.py # Module exports │ ├── cluster.py # Cluster commands public API │ ├── core.py # Core commands public API │ ├── helpers.py # Helpers for commands modules │ ├── redismodules.py # Trait for all Redis modules │ └── sentinel.py # Sentinel commands public API对应到当前仓库可以看到完整的实际实现redis/commands/core.py核心命令公共 API定义了CoreCommands等 trait 类包含 string、list、set、hash、zset、stream、geo、pubsub、script、server 管理等全部核心命令redis/commands/cluster.py集群命令公共 API如cluster_addslots、cluster_nodes等其中部分核心命令如exists、delete、mget会在集群场景下重写为按 slot 分片执行redis/commands/sentinel.pySentinel 命令公共 APIsentinel_masters、sentinel_failover等redis/commands/redismodules.py所有 Redis 模块的 trait 入口通过json()、ft()、ts()、bf()等工厂方法返回模块命令对象redis/commands/helpers.py命令模块的公共助手函数。定位命令归属新增命令前先判断它属于哪一类——string/list/set/hash/zset/stream 等核心命令放入core.py集群管理类命令放入cluster.pySentinel 相关放入sentinel.py若命令属于某个 Redis 模块如 RediSearch、RedisJSON、TimeSeries、Bloom 系、VectorSet则放入 redis/commands/ 下对应的模块子目录search/、json/、timeseries/、bf/、vectorset/。2.2 公共 API 与底层执行链所有命令实现的最终落点都是execute_command(*args, **kwargs)由CommandProtocol定义。无论同步还是异步客户端命令方法只负责把用户友好的参数翻译成 Redis 友好的线协议参数列表然后交给execute_command完成编码、发送与响应解析。这也是参数定义一节中提到的在命令方法内部做转换的原因——比如scan把可选的COUNT选项拼接进pieces列表。三、第一步撰写/获取命令规范新增命令的起点是一份命令规范文件。规范的来源有两种官方文档已收录Redis 官方命令文档是所有命令相关信息的权威来源可直接引用官方文档尚未收录Redis 是活跃演进的项目新命令可能尚未出现在官方文档中。此时需要基于 command-specification-template.md 模板自行创建规范。模板定义了规范的五个核心板块这是新增命令工作的需求规格书板块内容示例Supported version命令所需的最低 Redis 版本Redis 6.2.0Command description命令的功能描述一句话说明命令做什么Command API采用官方文档格式的命令签名$COMMAND_NAME $key $member [NX\|XX] [CH] [INCR]Redis-CLI examples相关的 redis-cli 操作示例实际可执行的示例命令Test plan集成测试计划逐条列出测试场景与断言3.1 为什么 Test plan 要逐条写清楚规范中的 Test plan 直接决定后续测试代码的编写。模板给出的示例是- Test only with required arguments, assert that single value returned - Test with required arguments and optional XX modifier, ensure that 1 returned - ...每条测试计划应包含输入参数的组合与期望的返回值两部分这样实现者可以直接将其翻译为assert语句。四、第二步阅读理解规范拿到规范后按顺序完成以下阅读理解工作通读整个规范尤其是 Command Description先确认命令类型归属string / list / set / hash / zset / stream / module 等这决定了它应该被放进哪个 trait 对象、以及测试文件应该建在哪里梳理 Command API区分必需参数与可选参数optional arguments判断 Redis 命令参数类型如何映射为 Python 类型数字、字符串、字节、字典、列表等确定返回值类型及所有可能的响应类型单个值、数组、字典、None 等查看 Redis-CLI 示例示例既是理解命令语义的最佳途径也是编写集成测试断言的基础复核 Test plan确认测试场景是否覆盖了必需参数、可选修饰符、边界条件等。4.1 命令参数如何映射为 Python 类型以 redis/commands/core.py 中scan命令第 6061 行起的真实实现为例可以看到典型的参数映射与转换逻辑overload def scan( self: SyncClientProtocol, cursor: int 0, match: PatternT | None None, count: int | None None, _type: str | None None, **kwargs, ) - ScanResponse: ... overload def scan( self: AsyncClientProtocol, cursor: int 0, match: PatternT | None None, count: int | None None, _type: str | None None, **kwargs, ) - Awaitable[ScanResponse]: ... def scan( self, cursor: int 0, match: PatternT | None None, count: int | None None, _type: str | None None, **kwargs, ) - ScanResponse | Awaitable[ScanResponse]: Incrementally return lists of key names. Also return a cursor indicating the scan position. match allows for filtering the keys by pattern count provides a hint to Redis about the number of keys to return per batch. _type filters the returned values by a particular Redis type. Stock Redis instances allow for the following types: HASH, LIST, SET, STREAM, STRING, ZSET Additionally, Redis modules can expose other types as well. pieces: list[EncodableT] [cursor] if match is not None: # 将用户友好的 match 参数转换为 Redis 线协议中的 MATCH 选项 pieces.extend([bMATCH, match]) ...注意两个要点隐藏复杂度Redis 原生命令要求COUNT 后面跟数量这种固定语法但公共 API 直接以count: int | None None暴露由方法内部拼装pieces用户无需关心线协议细节。这也是 specs/redis_commands_guide.md 中Arguments definition一节强调的将 Redis 命令的复杂 API 结构转换为用户友好的形式。参数类型尽量宽泛对于输入参数优先使用最通用的抽象接口。规范中的指导原则是如果参数只会被迭代应标注为Iterable[...]而非list[...]以接受 list、tuple、set、generator 等任意可迭代输入只有实现确实需要索引、切片、排序保证或原地修改时才固定为list[...]。例如# Preferred: accepts any iterable of strings. def querylabels(self, label: str | None None, filters: Iterable[str] | None None): ... # Avoid unless a list is explicitly required (indexing, ordering, mutation): def querylabels(self, label: str | None None, filters: list[str] | None None): ...这一原则只适用于输入参数返回值类型提示则应反映命令实际产出的具体类型见下文协议兼容性部分。五、第三步按序实现命令5.1 定位任务涉及的代码文件先确认需要修改或新建的文件阅读相关联的既有实现如最接近的新命令同类命令作为参照。新增一个核心命令通常涉及redis/commands/core.py或对应的模块commands.py新增命令方法redis/typing.py若引入了新的响应结构需要补充类型别名redis/_parsers/response_callbacks.py或redis/commands/$module/注册响应回调测试文件tests/test_commands.py同步与tests/test_asyncio/test_commands.py异步。5.2 在 trait 对象中实现命令将命令方法添加至匹配的 trait 类例如核心命令放入 redis/commands/core.py 的CoreCommands。实现的关键约束是同步/异步 API 都要有通过overload实现双重签名参数与响应类型考虑 bytes 表示客户端可配置decode_responsesFalse此时响应为 bytes同时允许传 bytes 参数替代字符串通过response_callbacks保证 RESP2/RESP3 兼容。同步/异步 overload 三件套模板按照 .agents/sync_async_type_hints_overload_guide.md 的规范同时支持同步与异步客户端的命令使用如下模板from typing import Awaitable, overload from redis.typing import AsyncClientProtocol, SyncClientProtocol class SomeCommands: overload def method(self: SyncClientProtocol, arg: str | None None) - SyncReturnType: ... overload def method( self: AsyncClientProtocol, arg: str | None None ) - Awaitable[SyncReturnType]: ... def method( self, arg: str | None None ) - SyncReturnType | Awaitable[SyncReturnType]: # implementation ...要点两个overload签名必须完整镜像实现的输入签名——参数名、顺序、默认值、位置/关键字形态、*args、**kwargs全部保持一致使用现代注解风格X | Y和T | None而非Union[...]/Optional[...]实现体的返回注解为SyncType | Awaitable[SyncType]当同步侧本身是联合类型时用括号分组后再接异步分支如(SentinelMastersResponse | bool) | Awaitable[SentinelMastersResponse | bool]运行时装饰器deprecation、experimental 等必须放在真实实现上而不是overload存根上若同步与异步客户端返回相同的具体类型只需保留单个方法不必添加冗余 overload。判别依据redis/typing.py中定义了基于Literal标记的 Protocolclass SyncClientProtocol(Protocol): Marker for sync clients. _is_async_client: Literal[False] class AsyncClientProtocol(Protocol): Marker for async clients. _is_async_client: Literal[True]采用 Protocol 判别而非Redis[bytes]/AsyncRedis[str]是为了避免因decode_responses设置而倍增 overload 数量。类型别名体系输入参数类型应优先复用 redis/typing.py 中预定义的类型别名别名含义定义KeyT主 Redis 键空间bytes \| str \| memoryviewFieldThash 表、stream、geo 命令中的字段EncodableTEncodableT任意可编码值bytes \| bytearray \| memoryview \| str \| int \| floatExpiryT相对过期时长int \| timedeltaAbsExpiryT绝对过期时间int \| datetimePatternT匹配键/字段的模式bytes \| str \| memoryviewKeysT键或键的迭代KeyT \| Iterable[KeyT]如果命令引入了新的响应结构如新的 dict 形状、元组组合应在redis/typing.py中补充命名类型别名并确保公共命令的返回类型提示包含这些新结构。5.3 协议兼容性RESP2 与 RESP3 的双层解析redis-py 8.0 起客户端默认在线上使用 RESP3 协议同时legacy_responsesTrue保留与 RESP2 兼容的 Python 响应形状。线协议与 Python 响应形状是独立选择的specs/redis_commands_guide.md 给出了完整的组合矩阵Redis()线上 RESP3 传统 RESP2 兼容的 Python 响应形状Redis(protocol2)线上 RESP2 传统 RESP2 Python 响应形状Redis(protocol3)线上 RESP3 传统 RESP3 Python 响应形状Redis(legacy_responsesFalse)启用统一响应形状unified shapes必须与协议无关除非命令存在官方文档记载的服务器端差异否则 RESP2 与 RESP3 下形状应一致。新增或修改命令时需要为以下每一种受支持模式确认期望的响应形状RESP2 legacyprotocol2、legacy_responsesTrueRESP3 legacyprotocol3、legacy_responsesTruedefault legacyprotocol未设置、legacy_responsesTrueRESP3 线协议 RESP2 兼容 Python 形状RESP2 unifiedprotocol2、legacy_responsesFalseRESP3 unifiedprotocol3、legacy_responsesFalseRESP 线协议解析由 redis/_parsers/resp2.py 与 redis/_parsers/resp3.py 完成而协议/响应形状兼容性则通过第二层解析——响应回调response callbacks实现核心命令回调定义在 redis/_parsers/response_callbacks.py模块专属回调映射表位于 redis/commands 下各模块目录。回调注册规则根据 .agents/sync_async_type_hints_overload_guide.md 的说明核心命令在 redis/_parsers/helpers.py 的三层回调字典中注册_RedisCallbacksRESP2 与 RESP3 共享的基础回调_RedisCallbacksRESP2RESP2 专属的覆盖/新增回调_RedisCallbacksRESP3RESP3 专属的覆盖/新增回调。模块命令在模块的__init__.py中注册_MODULE_CALLBACKS共享回调_RESP2_MODULE_CALLBACKS/_RESP3_MODULE_CALLBACKS各协议专属回调。回调查找顺序RESP2 下先查_RedisCallbacksRESP2未命中再回退_RedisCallbacksRESP3 同理。若没有注册任何回调则返回原始响应形状取决于decode_responses设置。协议相关返回类型差异是常见陷阱。指南中列出的典型差异包括命令RESP2 返回RESP3 返回原因acl_catlist[str]list[bytes \| str]RESP2 有str_if_bytesRESP3 无回调geohashlist[str]list[bytes \| str]同上hgetalldictpairs_to_dictdictidentity解析逻辑不同zincrby/zscorefloatfloat_or_nonefloatrawRESP2 解析、RESP3 直接返回类型提示策略对存在协议差异的命令采用最宽松的联合类型——若 RESP2 返回str而 RESP3 返回bytes | str则公共注解用bytes | str保证任何协议下都类型安全。兼容性原则legacy 回调应保留既有行为除非改动有意修复 bugunified 回调应提供推荐的、与协议无关的公共形状。5.4 边实现边验证每修改一个文件就立刻检查语法是否正确无缩进/括号错误import 是否正确新增的类型别名、回调、Protocol 等是否已导入类型是否正确定义overload与实现的签名是否完全镜像RESP2 与 RESP3 的响应 schema 是否保持一致或有意识地按文档差异处理。六、第四步实现测试计划实现完成后进入测试环节定位测试文件命令测试位于tests/test_*command_type*.py异步命令位于tests/test_asyncio/test_*command_type*.py。例如核心字符串命令对应 tests/test_commands.py 与 tests/test_asyncio/test_commands.py。因此第一步就识别命令类型至关重要它直接决定测试文件位置。若不存在匹配的测试文件则新建一个。每个测试用例一个独立测试方法把规范 Test plan 中的每一条翻译为一个测试方法每个方法只验证一个场景。版本约束若规范中声明了版本要求通过 tests/conftest.py 中定义的自定义注解skip_if_server_version_lt()与skip_if_server_version_gte()控制跳过。带有版本约束的命令通常只做集成测试。覆盖边界情况除规范 Test plan 列出的场景外补充空输入、None 返回、bytes/str 双形态等边界断言。七、第五步运行测试并保证双协议通过运行新增测试用例时使用 pytest 并显式指定协议# 以 RESP2 协议运行 pytest tests/test_commands.py -k test_your_new_command --protocol 2 # 以 RESP3 协议运行 pytest tests/test_commands.py -k test_your_new_command --protocol 3 # 异步版本同理 pytest tests/test_asyncio/test_commands.py -k test_your_new_command --protocol 2 pytest tests/test_asyncio/test_commands.py -k test_your_new_command --protocol 3--protocol是 tests/conftest.py第 123-128 行注册的 pytest 选项默认值来自default_protocol用于指定线协议版本同文件还注册了--legacy-responses选项可强制覆盖测试客户端的legacy_responses参数true/false/default三种取值。如需同时验证统一响应形状可组合使用pytest tests/test_commands.py -k test_your_new_command --protocol 3 --legacy-responses false验收标准同一组测试用例必须在 RESP2 与 RESP3 下全部通过。若任一协议下失败回到实现阶段修复重点检查响应回调注册与协议差异处理直到双协议绿。八、第六步最终验证清单提交前逐项确认计划中的全部任务已完成所有测试已创建且通过RESP2 与 RESP3 双协议代码符合项目约定overload 三件套、类型别名复用、现代注解风格、回调分层注册必要的文档已新增或更新如涉及新模块或新结构同步更新相关文档。九、输出报告完成后按要求输出工作总结包含Completed Tasks已完成任务清单创建的文件带路径修改的文件带路径。Tests Added创建的测试文件实现的测试用例测试结果双协议通过情况。十、常见问题速查Q新命令返回结构复杂类型提示怎么写先在 redis/typing.py 中定义命名类型别名再让 sync 与 async 两个 overload 以及实现体都引用该别名避免三处重复书写长联合类型。QRESP2 下测试通过、RESP3 下失败通常是什么原因最常见的是未注册 RESP3 专属回调或反之导致不同协议下返回原始形状不一致。检查 redis/_parsers/helpers.py 与 redis/_parsers/response_callbacks.py 中_RedisCallbacksRESP3的注册情况并对照协议返回类型差异表确认是否属于有意的协议差异。Q命令需要按服务器版本跳过测试怎么处理使用 tests/conftest.py 中的skip_if_server_version_lt(x.y.z)装饰器并在测试方法上声明。规范中的 Supported version 板块就是这里的依据。Q模块命令如 RediSearch、RedisJSON与核心命令的实现有何不同模块命令放入 redis/commands 下对应模块子目录回调注册在模块自身的__init__.py_MODULE_CALLBACKS/_RESP2_MODULE_CALLBACKS/_RESP3_MODULE_CALLBACKS并通过 redis/commands/redismodules.py 的工厂方法ft()、json()、ts()等暴露其余 overload、类型别名、双协议测试流程与核心命令完全一致。通过以上流程你可以系统地为 redis-py 贡献新命令支持并确保其与项目既有的同步/异步双 API、RESP2/RESP3 双协议架构保持一致。进一步的实现细节可继续研读 specs/redis_commands_guide.md 与 .agents/sync_async_type_hints_overload_guide.md 两份项目内部规范文档。【免费下载链接】redis-pyRedis Python client项目地址: https://gitcode.com/GitHub_Trending/re/redis-py创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考