Skip to main content

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

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

cumason123-helloworld-1.0.5.tar.gz (4.0 kB view details)

Uploaded Source

Built Distribution

cumason123_helloworld-1.0.5-py3-none-any.whl (3.5 kB view details)

Uploaded Python 3

File details

Details for the file cumason123-helloworld-1.0.5.tar.gz.

File metadata

  • Download URL: cumason123-helloworld-1.0.5.tar.gz
  • Upload date:
  • Size: 4.0 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/3.2.0 pkginfo/1.5.0.1 requests/2.24.0 setuptools/49.2.1 requests-toolbelt/0.9.1 tqdm/4.48.2 CPython/3.8.5

File hashes

Hashes for cumason123-helloworld-1.0.5.tar.gz
Algorithm Hash digest
SHA256 31112e92a6a6c768b49b0870967c5ca0b21cc6648046f395ed8d7ca75e0d3508
MD5 fd2eaabf8fa7410c13c70cd06ddf1048
BLAKE2b-256 1d7a191e6960f7ac499d421e2d034b7de844f79a105c5d8b2770806f42684be5

See more details on using hashes here.

File details

Details for the file cumason123_helloworld-1.0.5-py3-none-any.whl.

File metadata

  • Download URL: cumason123_helloworld-1.0.5-py3-none-any.whl
  • Upload date:
  • Size: 3.5 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/3.2.0 pkginfo/1.5.0.1 requests/2.24.0 setuptools/49.2.1 requests-toolbelt/0.9.1 tqdm/4.48.2 CPython/3.8.5

File hashes

Hashes for cumason123_helloworld-1.0.5-py3-none-any.whl
Algorithm Hash digest
SHA256 8fc37e35fff37d4e3c32edba02bf1ad0c6ea0c23b876d594dbc9fb8e452160c8
MD5 222e7b26ecd0c9d85c926f45f816e6b5
BLAKE2b-256 a779bcc286e850e3655d0963bd46315bc3cd73e11a3dc667ff18aa4ac26cc629

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page