Better Protobuf / gRPC 支持Python

本项目旨在通过利用现代语言功能，生成可读的、可理解的、符合Python语法的代码，为在现代Python环境中使用Protobuf / gRPC提供改进的体验。它不支持旧功能或环境（例如Protobuf 2）。以下功能得到支持

• Protobuf 3 & gRPC 代码生成
  • 内置二进制和JSON序列化
• Python 3.6+ 利用
  • 枚举
  • 数据类
  • async/await
  • 时区感知的 datetime 和 timedelta 对象
  • 相对导入
  • Mypy 类型检查

本项目深受以下项目的启发，并借鉴了其功能：

• https://github.com/protocolbuffers/protobuf/tree/master/python
• https://github.com/eigenein/protobuf/
• https://github.com/vmagamedov/grpclib

动机

本项目存在的原因是我对官方Google protoc插件在Python中的状态不满意。

• 没有 async 支持（需要额外的 grpclib 插件）
• 没有类型支持或代码补全/智能提示（需要额外的 mypy 插件）
• 不会生成 __init__.py 模块文件
• 输出不可导入
  • 除非你修改 sys.path，否则导入路径会在Python 3中中断
• 名称冲突时的错误（例如 codecs 包）
• 生成的代码不符合Python语法
  • 完全不可读的运行时代码生成
  • 很多代码看起来像是直接将 C++ 或 Java 代码 1:1 转换到 Python
  • 大写函数名，例如 HasField() 和 SerializeToString()
  • 使用 SerializeToString() 而不是内置的 __bytes__()
  • 特殊包装类型不使用 Python 的 None
  • 时间戳/持续时间类型不使用 Python 的内置 datetime 模块

本项目从头开始重新实现，专注于使用现代 Python 语言的风格，以解决上述问题。虽然由于方法名称和调用模式的变化，它可能不是 1:1 的直接替换，但网络格式是相同的。

安装 & 入门

首先，安装该包。请注意，[compiler] 功能标志告诉它只安装 protoc 插件所需的额外依赖项。

# Install both the library and compiler
$ pip install "betterproto[compiler]"

# Install just the library (to use the generated code output)
$ pip install betterproto

现在，假设你已经安装了编译器，并且有一个 proto 文件，例如 example.proto

syntax = "proto3";

package hello;

// Greeting represents a message you can tell a user.
message Greeting {
  string message = 1;
}

你可以运行以下命令

$ protoc -I . --python_betterproto_out=. example.proto

这会生成 hello.py，如下所示

# Generated by the protocol buffer compiler.  DO NOT EDIT!
# sources: hello.proto
# plugin: python-betterproto
from dataclasses import dataclass

import betterproto


@dataclass
class Hello(betterproto.Message):
    """Greeting represents a message you can tell a user."""

    message: str = betterproto.string_field(1)

现在你可以使用它了！

>>> from hello import Hello
>>> test = Hello()
>>> test
Hello(message='')

>>> test.message = "Hey!"
>>> test
Hello(message="Hey!")

>>> serialized = bytes(test)
>>> serialized
b'\n\x04Hey!'

>>> another = Hello().parse(serialized)
>>> another
Hello(message="Hey!")

>>> another.to_dict()
{"message": "Hey!"}
>>> another.to_json(indent=2)
'{\n  "message": "Hey!"\n}'

异步 gRPC 支持

生成的 Protobuf Message 类与 grpclib 兼容，因此如果你喜欢，可以自由使用它。但话说回来，本项目还包括异步 gRPC 代码生成器的支持，具有更好的静态类型检查和代码补全支持。默认启用。

给定一个示例

syntax = "proto3";

package echo;

message EchoRequest {
  string value = 1;
  // Number of extra times to echo
  uint32 extra_times = 2;
}

message EchoResponse {
  repeated string values = 1;
}

message EchoStreamResponse  {
  string value = 1;
}

service Echo {
  rpc Echo(EchoRequest) returns (EchoResponse);
  rpc EchoStream(EchoRequest) returns (stream EchoStreamResponse);
}

你可以这样使用它（首先在交互式 shell 中启用异步）

>>> import echo
>>> from grpclib.client import Channel

>>> channel = Channel(host="127.0.0.1", port=1234)
>>> service = echo.EchoStub(channel)
>>> await service.echo(value="hello", extra_times=1)
EchoResponse(values=["hello", "hello"])

>>> async for response in service.echo_stream(value="hello", extra_times=1)
        print(response)

EchoStreamResponse(value="hello")
EchoStreamResponse(value="hello")

JSON

支持将消息序列化和反序列化为 JSON 和 Python 字典，使用以下方法

• 字典：Message().to_dict()，Message().from_dict(...)
• JSON：Message().to_json()，Message().from_json(...)

为了兼容性，默认将字段名称转换为 camelCase。你可以通过传递一个大小写值来控制此行为，例如

>>> MyMessage().to_dict(casing=betterproto.Casing.SNAKE)

确定消息是否已发送

有时确定消息是否已通过网络发送很有用。这就是 Google 包装类型如何让你知道值是否未设置、默认（零值）或设置为其他值的方式。

使用 betterproto.serialized_on_wire(message) 来确定它是否已发送。这与官方 Google 生成的 Python 代码略有不同，并且它位于生成的 Message 类之外，以防止名称冲突。请注意，它 仅 支持 Proto 3，因此 仅 可以用于检查 Message 字段是否已设置。你不能检查标量是否已通过网络发送。

# Old way (official Google Protobuf package)
>>> mymessage.HasField('myfield')

# New way (this project)
>>> betterproto.serialized_on_wire(mymessage.myfield)

One-of 支持

Protobuf 支持在 oneof 子句中分组字段。在给定时间，组中的一个字段可能被设置。例如，给定以下 proto

syntax = "proto3";

message Test {
  oneof foo {
    bool on = 1;
    int32 count = 2;
    string name = 3;
  }
}

你可以使用 betterproto.which_one_of(message, group_name) 来确定哪个字段被设置了。它返回一个包含字段名称和值的元组，如果未设置，则返回一个空字符串和 None。

>>> test = Test()
>>> betterproto.which_one_of(test, "foo")
["", None]

>>> test.on = True
>>> betterproto.which_one_of(test, "foo")
["on", True]

# Setting one member of the group resets the others.
>>> test.count = 57
>>> betterproto.which_one_of(test, "foo")
["count", 57]
>>> test.on
False

# Default (zero) values also work.
>>> test.name = ""
>>> betterproto.which_one_of(test, "foo")
["name", ""]
>>> test.count
0
>>> test.on
False

这又与官方 Google 代码生成器略有不同

# Old way (official Google protobuf package)
>>> message.WhichOneof("group")
"foo"

# New way (this project)
>>> betterproto.which_one_of(message, "group")
["foo", "foo's value"]

知名 Google 类型

Google 提供了多个知名的消息类型，如时间戳、持续时间以及用于提供可选零值支持的几个包装器。每个类型都有特殊的 JSON 表示形式，并且与普通消息的处理方式略有不同。这些在 Python 中的映射如下

对于包装器类型，Python 类型对应于包装的类型，例如 google.protobuf.BoolValue 变为 Optional[bool]，而 google.protobuf.Int32Value 变为 Optional[int]。所有可选值默认为 None，因此不要忘记检查该可能的状态。给定

syntax = "proto3";

import "google/protobuf/duration.proto";
import "google/protobuf/timestamp.proto";
import "google/protobuf/wrappers.proto";

message Test {
  google.protobuf.BoolValue maybe = 1;
  google.protobuf.Timestamp ts = 2;
  google.protobuf.Duration duration = 3;
}

你可以做类似的事情

>>> t = Test().from_dict({"maybe": True, "ts": "2019-01-01T12:00:00Z", "duration": "1.200s"})
>>> t
Test(maybe=True, ts=datetime.datetime(2019, 1, 1, 12, 0, tzinfo=datetime.timezone.utc), duration=datetime.timedelta(seconds=1, microseconds=200000))

>>> t.ts - t.duration
datetime.datetime(2019, 1, 1, 11, 59, 58, 800000, tzinfo=datetime.timezone.utc)

>>> t.ts.isoformat()
'2019-01-01T12:00:00+00:00'

>>> t.maybe = None
>>> t.to_dict()
{'ts': '2019-01-01T12:00:00Z', 'duration': '1.200s'}

开发

首先，请确保您已安装 Python 3.6+ 和 pipenv，以及适用于您平台的官方 Protobuf 编译器。然后

# Get set up with the virtual env & dependencies
$ pipenv install --dev

# Link the local package
$ pipenv shell
$ pip install -e .

代码风格

本项目强制执行 black Python 代码格式化。

在提交更改之前运行

pipenv run black .

为了避免后续的合并冲突，非 black 格式的 Python 代码将在 CI 中失败。

测试

有两种类型的测试

1. 标准测试
2. 自定义测试

标准测试

添加标准测试用例很容易。

• 创建一个新的目录 betterproto/tests/inputs/<name>
  • 添加 <name>.proto，其中包含名为 Test 的消息
  • 添加 <name>.json，其中包含一些测试数据

在运行测试时，它将自动被选中。

• 另请参阅： 标准测试开发指南

自定义测试

自定义测试位于 tests/test_*.py 中，并使用 pytest 运行。

运行

以下是运行测试的方法。

# Generate assets from sample .proto files
$ pipenv run generate

# Run all tests
$ pipenv run test

待办事项

• 固定长度字段
  • 打包固定长度
• 交错有符号字段（sint32，sint64）
• 对于嵌套类型，不要对零值进行编码
• 枚举
• 重复消息字段
• 映射
  • 消息字段的映射
• 支持未知字段的透传
• 对嵌套类型的引用
• proto 文件中的导入
• 已知 Google 类型
  • 作为请求输入支持
  • 作为响应输出支持
    • 自动包装/解包响应
• OneOf 支持
  • 基本有线支持
  • 检查组中设置了哪个
  • 设置一个会取消其他设置
• 不是完全天真的 JSON。
  • 64 位整型作为字符串
  • 映射
  • 列表
  • 字节作为 base64
  • Any 支持
  • 枚举字符串
  • 已知类型支持（时间戳、持续时间、包装器）
  • 支持不同的大小写（原版、驼峰式等）
• 异步服务存根
  • 单-单
  • 服务器端流响应
  • 客户端流请求
• 重命名消息和字段以符合 Python 命名标准
• 与语言关键字冲突的重命名
• Python 包
• 自动化运行测试
• 清理！

许可证

版权所有 © 2019 Daniel G. Taylor

http://dgt.mit-license.org/