Released ScalePlan CloudEvents for Python

Posted by Yüce Tekol on 2020-09-29

Continue reading

We released spce-python, our first contribution to the CloudEvents ecosystem.

What is spce-python?

spce-python is a Python library which implements CloudEvents 1.0 specification together with JSON and Avro encoders/decoders.

Here are the links to the currently supported specifications:

CloudEvents is a new Cloud Native Computing Foundation specification for describing event data. It was adopted by giants like IBM, Microsoft, SAP and Google. We are also heavily investing on it. Particle DataGateway has top-notch CloudEvents support for receiving and sending data and also inter-node communication.

Install

spce-python has no dependencies beyond the Python standard library. It optionally depends on the avro package for Avro encode/decode functionality.

We officially support only Python 3.6 or above, but prior versions may also work. (Sorry, definitely no Python 2 support).

Installing spce-python is trivial:

  • With the JSON codec:

    pip install spce
  • With JSON and Avro codecs:

    pip install spce[avro]

Usage:

Creating CloudEvents

Create a CloudEvent with required attributes:

from spce import CloudEvent

event = CloudEvent(
    type="OximeterMeasured",
    source="oximeter/123",
    id="1000"
)

The id field is required, it won't be auto-generated if blank.

Create a CloudEvent with optional attributes:

event = CloudEvent(
    type="OximeterMeasured",
    source="oximeter/123",
    id="1000",
    subject="subject1",
    dataschema="https://particlemetrics.com/schema",
    time="2020-09-28T21:33:21Z",
    data='{\"spo2\": 99})',
    datacontenttype="application/json"
)

The time field can be an RFC3336 compatible timestamp string or a datetime.datetime object. If left out, it won't be automatically set. If you need to set the time field to the current time, you can use the datetime.utcnow method:

from datetime import datetime

now = datetime.utcnow()
event = CloudEvent(
    type="OximeterMeasured",
    source="oximeter/123",
    id="1000",
    time=now
)

Check out https://github.com/scaleplandev/spce-python/blob/master/tests/cloudevents_test.py for a few examples that set the time.

Required and optional attributes can be directly accessed:

assert event.type == "OximeterMeasured" 
assert event.time == "2020-09-28T21:33:21Z" 

Create a CloudEvent with extension attributes:

event = CloudEvent(
    type="OximeterMeasured",
    source="oximeter/123",
    id="1000",
    external1="foo/bar"
)

Extension attributes can be accessed using the attribute method:

assert event.attribute("external1") == "foo/bar" 

JSON Encoder/Decoder

Encode an event in JSON:

from spce import Json

encoded_event = Json.encode(event)

Note that blank fields won't be encoded.

Decode an event in JSON:

from spce import Json

text = """
    {
      "type": "OximeterMeasured",
      "source": "oximeter/123",
      "id": "1000",
      "specversion": "1.0",
      "datacontenttype": "application/json",
      "subject": "subject1",
      "dataschema": "https://particlemetrics.com/schema",
      "time": "2020-09-28T21:33:21Z",
      "data": "{\"spo2\": 99})"
    }
"""
decoded_event = Json.decode(text) 

Avro Encoder/Decoder

Encode an event in Avro:

from spce import Avro

encoded_event = Avro.encode(event)

Note that blank fields won't be encoded.

Decode an event in Avro:

from spce import Avro

encoded_event = \
    (b'\n\x08type\x06 OximeterMeasured\x0csource\x06\x18oximeter/123\x04id'
     b'\x06\x081000\x16specversion\x06\x061.0\x1edatacontenttype\x06 application'
     b'/json\x00\x0c\x18{"spo2": 99}')
decoded_event = Avro.decode(encoded_event) 

We designed spce-python to feel natural for Python developers and anyone who is familiar to the CloudEvents specification. Hopefully we achieved that goal.

What's Next?

CloudEvents WG has an official CloudEvents SDK for Python. You may want to check it out if spce-python doesn't fit your needs, or better file an issue on GitHub so we can make spce-python awesome.

Particle DataGateway is probably the best way to get the event data into your system. It's beta quality software, mind you but we are improving it everyday. Contact us to learn more about it and schedule a demo.

We have so much to do with events and CloudEvents in particular. Consider subscribing to our monthly newsletter Awesome Event Systems for the latest news.

Yüce Tekol

Founder, ScalePlan

Conventional systems cannot cope with the vast amount of data flowing from people, IoT devices, games and applications. We research, develop and provide solutions using the best-in-class software and platforms in DevOps, data retrieval, streaming, processing, and storage.