No project description provided
Project description
Automated PyPI Deployment sample
For used github action see pypi-release.yml
For automatically creating tags see pypi_packaging.py
Test
Python SDK for CloudEvents
Status
This SDK is still considered a work in progress, therefore things might (and will) break with every update.
This SDK current supports the following versions of CloudEvents:
- v1.0
- v0.3
Python SDK
Package cloudevents provides primitives to work with CloudEvents specification: https://github.com/cloudevents/spec.
Parsing upstream structured Event from HTTP request:
import io
from cloudevents.sdk.event import v1
from cloudevents.sdk import marshaller
m = marshaller.NewDefaultHTTPMarshaller()
event = m.FromRequest(
v1.Event(),
{"content-type": "application/cloudevents+json"},
io.StringIO(
"""
{
"specversion": "1.0",
"datacontenttype": "application/json",
"type": "word.found.name",
"id": "96fb5f0b-001e-0108-6dfe-da6e2806f124",
"time": "2018-10-23T12:28:22.4579346Z",
"source": "<source-url>"
}
"""
),
lambda x: x.read(),
)
Parsing upstream binary Event from HTTP request:
import io
from cloudevents.sdk.event import v1
from cloudevents.sdk import marshaller
m = marshaller.NewDefaultHTTPMarshaller()
event = m.FromRequest(
v1.Event(),
{
"ce-specversion": "1.0",
"content-type": "application/json",
"ce-type": "word.found.name",
"ce-id": "96fb5f0b-001e-0108-6dfe-da6e2806f124",
"ce-time": "2018-10-23T12:28:22.4579346Z",
"ce-source": "<source-url>",
},
io.BytesIO(b"this is where your CloudEvent data"),
lambda x: x.read(),
)
Creating a minimal CloudEvent in version 0.1:
from cloudevents.sdk.event import v1
event = (
v1.Event()
.SetContentType("application/json")
.SetData('{"name":"john"}')
.SetEventID("my-id")
.SetSource("from-galaxy-far-far-away")
.SetEventTime("tomorrow")
.SetEventType("cloudevent.greet.you")
)
Creating HTTP request from CloudEvent:
from cloudevents.sdk import converters
from cloudevents.sdk import marshaller
from cloudevents.sdk.converters import structured
from cloudevents.sdk.event import v1
event = (
v1.Event()
.SetContentType("application/json")
.SetData('{"name":"john"}')
.SetEventID("my-id")
.SetSource("from-galaxy-far-far-away")
.SetEventTime("tomorrow")
.SetEventType("cloudevent.greet.you")
)
m = marshaller.NewHTTPMarshaller([structured.NewJSONHTTPCloudEventConverter()])
headers, body = m.ToRequest(event, converters.TypeStructured, lambda x: x)
HOWTOs with various Python HTTP frameworks
In this topic you'd find various example how to integrate an SDK with various HTTP frameworks.
Python requests
One of popular framework is requests
.
CloudEvent to request
The code below shows how integrate both libraries in order to convert a CloudEvent into an HTTP request:
def run_binary(event, url):
binary_headers, binary_data = http_marshaller.ToRequest(
event, converters.TypeBinary, json.dumps)
print("binary CloudEvent")
for k, v in binary_headers.items():
print("{0}: {1}\r\n".format(k, v))
print(binary_data.getvalue())
response = requests.post(url,
headers=binary_headers,
data=binary_data.getvalue())
response.raise_for_status()
def run_structured(event, url):
structured_headers, structured_data = http_marshaller.ToRequest(
event, converters.TypeStructured, json.dumps
)
print("structured CloudEvent")
print(structured_data.getvalue())
response = requests.post(url,
headers=structured_headers,
data=structured_data.getvalue())
response.raise_for_status()
Complete example of turning a CloudEvent into a request you can find here.
Request to CloudEvent
The code below shows how integrate both libraries in order to create a CloudEvent from an HTTP request:
response = requests.get(url)
response.raise_for_status()
headers = response.headers
data = io.BytesIO(response.content)
event = v1.Event()
http_marshaller = marshaller.NewDefaultHTTPMarshaller()
event = http_marshaller.FromRequest(
event, headers, data, json.load)
Complete example of turning a CloudEvent into a request you can find here.
SDK versioning
The goal of this package is to provide support for all released versions of CloudEvents, ideally while maintaining the same API. It will use semantic versioning with following rules:
- MAJOR version increments when backwards incompatible changes is introduced.
- MINOR version increments when backwards compatible feature is introduced INCLUDING support for new CloudEvents version.
- PATCH version increments when a backwards compatible bug fix is introduced.
Community
- There are bi-weekly calls immediately following the Serverless/CloudEvents call at 9am PT (US Pacific). Which means they will typically start at 10am PT, but if the other call ends early then the SDK call will start early as well. See the CloudEvents meeting minutes to determine which week will have the call.
- Slack: #cloudeventssdk channel under CNCF's Slack workspace.
- Email: https://lists.cncf.io/g/cncf-cloudevents-sdk
- Contact for additional information: Denis Makogon (
@denysmakogon
on slack).
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for cumason123-helloworld-1.0.5.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 31112e92a6a6c768b49b0870967c5ca0b21cc6648046f395ed8d7ca75e0d3508 |
|
MD5 | fd2eaabf8fa7410c13c70cd06ddf1048 |
|
BLAKE2b-256 | 1d7a191e6960f7ac499d421e2d034b7de844f79a105c5d8b2770806f42684be5 |
Hashes for cumason123_helloworld-1.0.5-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 8fc37e35fff37d4e3c32edba02bf1ad0c6ea0c23b876d594dbc9fb8e452160c8 |
|
MD5 | 222e7b26ecd0c9d85c926f45f816e6b5 |
|
BLAKE2b-256 | a779bcc286e850e3655d0963bd46315bc3cd73e11a3dc667ff18aa4ac26cc629 |