使用 Python 发送 RPC 请求

应用可以使用 gRPC 和 Python 与 Universal Ledger API 进行交互。此方法涉及生成和使用从服务的协议缓冲区定义编译的 Python 客户端库。

本教程面向希望为 Universal Ledger API 实现客户端的开发者，介绍了如何设置合适的开发环境、生成必要的库，以及如何使用 Python 进行 gRPC 调用。

准备工作

如需完成本教程，您需要：

  • 启用了 Universal Ledger API 的 Google Cloud 项目。

  • 一个 IAM 角色，例如 universalledger.googleapis.com/endpointViewer，以便您至少可以查看通用账本端点。

  • 您用户账号的本地身份验证凭据。运行以下命令以配置这些凭据：

    gcloud auth application-default login

如需详细了解这些步骤，请参阅私密预览版初始配置指南。

设置环境

本教程中的说明适用于运行 Ubuntu 25.04 的环境，该环境默认包含 Python 3.13。如果您使用的是其他操作系统，可能需要修改命令。

如需设置开发环境，请完成以下步骤。以下步骤将引导您安装必要的命令行工具。这包括使用 pipx 安装 grpcio-tools（用于构建 gRPC 绑定）。为了隔离此项目的 Python 依赖项，您还将创建并激活 Python 虚拟环境，然后使用 pip 将所需的 Python 库安装到该环境中。

  1. 使用以下命令安装基本依赖项：

    sudo apt update
sudo apt install git pipx python3.13-venv

  2. 使用 pipx 安装 gRPC 工具应用，并确保其二进制目录已添加到 PATH：

    pipx install grpcio-tools
pipx ensurepath

    关闭并重新打开终端，以完成 pipx 的激活。

  3. 创建一个父级目录来保存本教程的文件。

    mkdir gcul_tutorial
cd gcul_tutorial

    除非另有说明，否则其余命令应在此新目录中运行。

  4. 创建 Python 虚拟环境并安装其余库：

    python3 -m venv example_client
cd example_client
source ./bin/activate
pip3 install google-auth googleapis-common-protos grpcio requests
cd ..

    这些库包括：

    • google-auth：用于处理 Google Cloud 身份验证。
    • googleapis-common-protos：Google API 中使用的通用协议缓冲区消息。
    • grpcio：适用于 Python 的 gRPC 库。
    • requests：某些依赖项需要此权限才能发送 HTTP 请求。

生成 Protocol Buffer 库

如需使用 gRPC 与 Universal Ledger API 进行交互，您的 Python 应用需要从服务的 Protocol Buffer (.proto) 定义编译的客户端库。这些定义指定了 API 的服务、方法和消息类型。

本部分介绍了如何从 googleapis 代码库下载这些 .proto 文件，并使用已安装的 grpcio-tools 生成必要的 Python 源代码。

  1. 克隆 googleapis GitHub 代码库，其中发布了 Google API 的 Protocol Buffer 定义。

    git clone https://github.com/googleapis/googleapis.git

  2. 为 Universal Ledger API 构建 Protocol Buffer 定义和 gRPC 绑定。

    python-grpc-tools-protoc --proto_path=googleapis/ \
    --python_out=example_client \
    --pyi_out=example_client \
    --grpc_python_out=example_client \
    google/cloud/universalledger/v1/universalledger.proto
python-grpc-tools-protoc --proto_path=googleapis/ \
    --python_out=example_client \
    --pyi_out=example_client \
    google/cloud/universalledger/v1/query.proto
python-grpc-tools-protoc --proto_path=googleapis/ \
    --python_out=example_client \
    --pyi_out=example_client \
    google/cloud/universalledger/v1/accounts.proto
python-grpc-tools-protoc --proto_path=googleapis/ \
    --python_out=example_client \
    --pyi_out=example_client \
    google/cloud/universalledger/v1/common.proto
python-grpc-tools-protoc --proto_path=googleapis/ \
    --python_out=example_client \
    --pyi_out=example_client \
    google/cloud/universalledger/v1/transactions.proto
python-grpc-tools-protoc --proto_path=googleapis/ \
    --python_out=example_client \
    --pyi_out=example_client \
    google/cloud/universalledger/v1/types.proto
python-grpc-tools-protoc --proto_path=googleapis/ \
    --python_out=example_client \
    --pyi_out=example_client \
    google/cloud/universalledger/v1/status_event.proto

    这应该会在 example_client/google/cloud/universalledger/v1 目录下创建多个文件。

    accounts_pb2.py
accounts_pb2.pyi
common_pb2.py
common_pb2.pyi
query_pb2.py
query_pb2.pyi
status_event_pb2.py
status_event_pb2.pyi
transactions_pb2.py
transactions_pb2.pyi
types_pb2.py
types_pb2.pyi
universalledger_pb2.py
universalledger_pb2.pyi
universalledger_pb2_grpc.py

调用 Universal Ledger API

最后，创建一个 Python 脚本，以使用生成的库调用 Universal Ledger API。将以下代码保存为 example_client/endpoints.py。

在代码中，替换以下占位值：

  • PROJECT_ID：您的 Google Cloud 项目 ID，例如 my-project。
  • REGION：通用账本端点所在的 Google Cloud 区域，例如 us-east5。
"""Example calling the Universal Ledger API."""

import sys

import google.auth
import google.auth.transport.grpc
import google.auth.transport.requests
from google.cloud.universalledger.v1 import universalledger_pb2
from google.cloud.universalledger.v1 import universalledger_pb2_grpc
import grpc

API_ENDPOINT = "universalledger.googleapis.com"
PROJECT = "PROJECT_ID"
REGION = "REGION"
SCOPES = ["https://www.googleapis.com/auth/cloud-platform"]


def main() -> str | None:
  # Get the application default credentials.
  credentials, _ = google.auth.default(scopes=SCOPES)

  # Get an HTTP request function to refresh credentials.
  refresh_request = google.auth.transport.requests.Request()

  # Create a secure channel to the API endpoint.
  with google.auth.transport.grpc.secure_authorized_channel(
      credentials, refresh_request, API_ENDPOINT
  ) as channel:
    # Create the client stub using the generated code.
    stub = universalledger_pb2_grpc.UniversalLedgerStub(channel)

    # Parent location for GCUL network endpoints.
    parent = f"projects/{PROJECT}/locations/{REGION}"

    # Build the request message.
    request = universalledger_pb2.ListEndpointsRequest(parent=parent)

    # Make the gRPC call.
    try:
      metadata = [("x-goog-request-params", f"parent={parent}")]
      response = stub.ListEndpoints(request, metadata=metadata)
    except grpc.RpcError as exc:
      return f"{exc.code().name}: {exc.details()}"

    if not response.endpoints:
      return f"No endpoints found under: {parent}"

    print("Found the following endpoints:")
    for endpoint in response.endpoints:
      print("-", endpoint.name)


if __name__ == "__main__":
  sys.exit(main())

如需运行该脚本，请前往 example_client 目录。 虚拟环境应仍处于设置步骤中的活跃状态。

  1. 更改为脚本目录：

    cd example_client

    shell 提示符应确认虚拟环境处于有效状态（例如，带有 (example_client) 前缀）。

  2. 运行 Python 脚本：

    python3 endpoints.py

您应该会看到如下所示的输出：

Found the following endpoints:
- projects/my-project/locations/us-east5/endpoint/gcul-pilot-testing
- projects/my-project/locations/us-east5/endpoint/gcul-user-testing

后续步骤