Python SDK for CloudEvents

状态

此SDK仍被视为工作正在进行中，因此更新可能会导致（并且将会）出现问题。

此SDK当前支持以下版本的CloudEvents

• v1.0
• v0.3

Python SDK

包 cloudevents 提供与CloudEvents规范一起工作的原语： https://github.com/cloudevents/spec.

安装

CloudEvents SDK可以使用pip进行安装

pip install cloudevents

发送CloudEvents

以下我们将提供使用流行的 requests 库发送cloudevents的示例。

二进制HTTP CloudEvent

from cloudevents.http import CloudEvent
from cloudevents.conversion import to_binary
import requests

# Create a CloudEvent
# - The CloudEvent "id" is generated if omitted. "specversion" defaults to "1.0".
attributes = {
    "type": "com.example.sampletype1",
    "source": "https://example.com/event-producer",
}
data = {"message": "Hello World!"}
event = CloudEvent(attributes, data)

# Creates the HTTP request representation of the CloudEvent in binary content mode
headers, body = to_binary(event)

# POST
requests.post("<some-url>", data=body, headers=headers)

结构化HTTP CloudEvent

from cloudevents.conversion import to_structured
from cloudevents.http import CloudEvent
import requests

# Create a CloudEvent
# - The CloudEvent "id" is generated if omitted. "specversion" defaults to "1.0".
attributes = {
    "type": "com.example.sampletype2",
    "source": "https://example.com/event-producer",
}
data = {"message": "Hello World!"}
event = CloudEvent(attributes, data)

# Creates the HTTP request representation of the CloudEvent in structured content mode
headers, body = to_structured(event)

# POST
requests.post("<some-url>", data=body, headers=headers)

您可以在示例目录中找到一个将CloudEvent转换为HTTP请求的完整示例。

接收CloudEvents

以下代码演示了如何使用流行的Python Web框架flask消费CloudEvent。

from flask import Flask, request

from cloudevents.http import from_http

app = Flask(__name__)


# create an endpoint at http://localhost:/3000/
@app.route("/", methods=["POST"])
def home():
    # create a CloudEvent
    event = from_http(request.headers, request.get_data())

    # you can access cloudevent fields as seen below
    print(
        f"Found {event['id']} from {event['source']} with type "
        f"{event['type']} and specversion {event['specversion']}"
    )

    return "", 204


if __name__ == "__main__":
    app.run(port=3000)

您可以在样本目录中找到一个将CloudEvent转换为HTTP请求的完整示例。

SDK版本控制

该软件包的目标是支持CloudEvents的所有已发布版本，理想情况下同时保持相同的API。它将使用语义版本控制，并遵循以下规则：

• 主版本号递增时，会引入向后不兼容的更改。
• 次版本号递增时，会引入向后兼容的功能，包括对新CloudEvents版本的支持。
• 修订号递增时，会引入向后兼容的bug修复。

社区

• 在Serverless/CloudEvents电话会议后，每周举行双周会议，时间为美国太平洋时间上午9点。这意味着它们通常在上午10点开始，但如果其他会议提前结束，那么SDK电话会议也会提前开始。请参阅CloudEvents会议纪要以确定哪周将召开会议。
• Slack: 在CNCF的Slack工作区下设有#cloudeventssdk频道。
• Email: https://lists.cncf.io/g/cncf-cloudevents-sdk
• 联系更多信息：Denis Makogon (@denysmakogon on slack)。

每个SDK可能都有其自己的独特流程、工具和指南，相关的共同治理材料可以在CloudEvents docs目录中找到。特别是，在那里您可以找到有关如何管理SDK项目、如何进行PR审查和批准的指南，以及我们的行为准则信息。

如果CloudEvents规范或项目SDK中存在安全问题，请发送电子邮件至cncf-cloudevents-security@lists.cncf.io。

其他SDK资源

• 当前活跃维护者列表
• 如何为项目贡献力量
• SDK的许可证
• SDK的发布流程

维护

我们使用black和isort进行自动格式化。我们设置了一个tox环境来重新格式化代码库。

例如。

pip install tox
tox -e reformat

有关版本升级信息的详细信息，请参阅RELEASING.md