spce 0.2.1

ScalePlan's CloudEvents implementation

ScalePlan CloudEvents for Python

Unofficial Python implementation for CloudEvents v1.0. Check out the CloudEvents spec.

This package has no dependencies beyond the Python standard library with the base install. Optionally depends on the avro package for Avro encode/decode functionality.

Install

Requirements:

• Python 3.6 or above

Install with JSON codec:

pip install spce

Install with JSON and Avro codecs:

pip install spce[avro]

Usage:

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 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" 

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) 

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

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 = Avro.decode(text) 

License

(c) 2020 Scale Plan Yazılım A.Ş. https://scaleplan.io

Licensed under Apache 2.0. See the LICENSE.