pycapnp

Cap'n'proto 邮件列表 文档

要求

• 支持 C++14 的编译器
  • gcc 6.1+（5+ 可能可用）
  • clang 6（3.4+ 可能可用）
  • Visual Studio 2017+
• cmake（捆绑的 capnproto 需要）
  • ninja（macOS + Linux）
  • Visual Studio 2017+
• capnproto-1.0（如果链接到系统库，>=0.8.0 也可工作）
  • 如果使用捆绑的 capnproto，则不需要
• Python 开发头文件（即 Python.h）
  • python.org 的发行版包含这些头文件，但在 Linux 发行版上通常位于单独的包中

32 位 Linux 要求 capnproto 使用 -fPIC 编译。除非您自己编译 capnproto，否则这通常已正确设置。对于 cmake，这也称为 -DCMAKE_POSITION_INDEPENDENT_CODE=1。

pycapnp 有额外的开发依赖项，包括 cython 和 pytest。所有依赖项详见 requirements.txt。

构建和安装

使用 pip install pycapnp 进行安装。您可以设置 CC 环境变量来控制使用哪个编译器，例如 CC=gcc-8.2 pip install pycapnp。

或者，您可以像这样克隆仓库：

git clone https://github.com/capnproto/pycapnp.git
cd pycapnp
pip install .

默认情况下，安装脚本会自动使用本地已安装的 Cap'n Proto。 如果未安装 Cap'n Proto，脚本会捆绑并构建匹配的 Cap'n Proto 库。

要强制捆绑 Cap'n Proto 库：

pip install . -C force-bundled-libcapnp=True

如果您希望使用最新的上游 C++ Cap'n Proto 进行安装：

pip install . \
    -C force-bundled-libcapnp=True \
    -C libcapnp-url="https://github.com/capnproto/capnproto/archive/master.tar.gz"

为强制使用系统中已安装的 Cap'n Proto：

pip install . -C force-system-libcapnp=True

打包系统并不那么智能，因此在更改版本时可能需要清理打包的构建文件：

python setup.py clean

存根文件生成

虽然 pycapnp 不直接支持存根文件生成，但已有工具可帮助生成 pycapnp 存根文件以辅助开发（如果您刚接触 pycapnp，这会非常有用！）。更多详情请参见 #289。

Python Capnp 存根生成器

Python 版本

支持 Python 3.9 及更高版本。

开发

已弃用 Git flow，请使用 master 分支。

要进行测试，请使用 pipenv（或手动安装 requirements.txt 并运行 pytest）。

pip install pipenv
pipenv install
pipenv run pytest

二进制包

构建 Python wheel 分发版

pip wheel .

发布到 PyPI

每次推送（包括标签推送）时，Build GitHub Actions 工作流（.github/workflows/wheels.yml）都会构建 Wheels 和 sdist。辅助脚本 scripts/release-pypi.sh 会为指定标签（或显式运行 ID）下载这些制品，并通过 twine 将其上传到 PyPI。

典型的发布流程：

git tag v2.2.1
git push origin v2.2.1
# wait for the "Build" workflow run to finish successfully on GitHub

# Download artifacts and upload to PyPI (creates dist_221/ by default).
scripts/release-pypi.sh v2.2.1

# Or, target a specific Actions run id:
scripts/release-pypi.sh 1234567890

# Dry run: upload to TestPyPI (https://test.pypi.org) instead of real PyPI.
# Useful for validating the release flow end-to-end before pushing to
# production. Requires a TestPyPI account + API token configured in
# ~/.pypirc under a [testpypi] section. See
# https://packaging.python.org/en/latest/guides/using-testpypi/ .
scripts/release-pypi.sh v2.2.1 --test

发布机器的要求：

• gh 命令行工具，已完成身份验证（gh auth login）
• python3（脚本会创建 .venv-release/ 并在其中安装 twine）
• twine 可使用的 PyPI 凭据，例如 TWINE_USERNAME=__token__ 和 TWINE_PASSWORD=<api-token>，或者已配置的 ~/.pypirc

脚本功能：

1. 解析标签对应的最新成功的 wheels.yml 运行（或使用提供的运行 ID）。
2. 下载 cibw-* 工件，并将所有 *.whl / *.tar.gz 文件提取到输出目录（标签默认目录为 dist_<digits>，运行 ID 默认目录为 dist_run_<id>；可传递第二个参数覆盖目录，使用 --force 可复用非空目录）。
3. 运行 twine check，打印文件列表，并在运行 twine upload 前提示确认。

文档/示例

这里有一些基本文档 here。

请务必查看 examples。这些示例通常会保持更新，以展示库的推荐用法。

示例目录中有一个示例很好地展示了 pycapnp 的功能。现转载如下：

import os
import capnp

import addressbook_capnp

def writeAddressBook(file):
    addresses = addressbook_capnp.AddressBook.new_message()
    people = addresses.init('people', 2)

    alice = people[0]
    alice.id = 123
    alice.name = 'Alice'
    alice.email = 'alice@example.com'
    alicePhones = alice.init('phones', 1)
    alicePhones[0].number = "555-1212"
    alicePhones[0].type = 'mobile'
    alice.employment.school = "MIT"

    bob = people[1]
    bob.id = 456
    bob.name = 'Bob'
    bob.email = 'bob@example.com'
    bobPhones = bob.init('phones', 2)
    bobPhones[0].number = "555-4567"
    bobPhones[0].type = 'home'
    bobPhones[1].number = "555-7654"
    bobPhones[1].type = 'work'
    bob.employment.unemployed = None

    addresses.write(file)


def printAddressBook(file):
    addresses = addressbook_capnp.AddressBook.read(file)

    for person in addresses.people:
        print(person.name, ':', person.email)
        for phone in person.phones:
            print(phone.type, ':', phone.number)

        which = person.employment.which()
        print(which)

        if which == 'unemployed':
            print('unemployed')
        elif which == 'employer':
            print('employer:', person.employment.employer)
        elif which == 'school':
            print('student at:', person.employment.school)
        elif which == 'selfEmployed':
            print('self employed')
        print()


if __name__ == '__main__':
    f = open('example', 'w')
    writeAddressBook(f)

    f = open('example', 'r')
    printAddressBook(f)

此外，pycapnp 还新增了 RPC 功能，包括流水线操作和基于 promise 风格的 API。有关更详细的演示，请参考示例目录中的计算器示例：

import asyncio
import capnp
import socket

import test_capability_capnp


class Server(test_capability_capnp.TestInterface.Server):

    def __init__(self, val=1):
        self.val = val

    async def foo(self, i, j, **kwargs):
        return str(i * 5 + self.val)


async def client(read_end):
    client = capnp.TwoPartyClient(read_end)

    cap = client.bootstrap()
    cap = cap.cast_as(test_capability_capnp.TestInterface)

    remote = cap.foo(i=5)
    response = await remote

    assert response.x == '125'

async def main():
    client_end, server_end = socket.socketpair(socket.AF_UNIX)
    # This is a toy example using socketpair.
    # In real situations, you can use any socket.

    client_end = await capnp.AsyncIoStream.create_connection(sock=client_end)
    server_end = await capnp.AsyncIoStream.create_connection(sock=server_end)

    _ = capnp.TwoPartyServer(server_end, bootstrap=Server(100))
    await client(client_end)


if __name__ == '__main__':
    asyncio.run(capnp.run(main()))