Event Gateway Python SDK
Project description
Event Gateway Python SDK
Python library for interacting with the Event Gateway.
Contents
Background
This is the Python SDK for interacting with the Event Gateway, the hub for connecting events to serverless functions.
Installation
pip install event-gateway-sdk
Usage
Use the emit
command to emit a CloudEvent payload to your Event Gateway. The event will be received by any function that is subscribed to your event.
from eventgateway import EventGateway
eg = EventGateway()
cloudEvent = {
"eventType": "user.created",
"cloudEventsVersion": "0.1",
"source": "/services/users",
"data": {
"userId": "foo",
"userName": "bar"
}
}
eg.emit(cloudEvent=configData, path="/user/send-mail-user")
The emit()
function takes three arguments:
- an
event
which is a valid CloudEvent, - a
path
which is the path associated to the function (default:/
) - a
headers
object that represents the headers sent to the gateway (default:{"Content-type": "application/json"}
)
The function returns a request object. If your event has a sync
subscription attached, the fetch
response will have the status code and body from the subscription. If not, the response will return a 202 Accepted
status code with an empty body.
Constructor
In the example above, we created an Event Gateway client using the application URL from the hosted Event Gateway provided by Serverless, Inc.
You can also use the Event Gateway SDK with your own, self-hosted Event Gateway. Constructor details are listed below.
Parameters
url
-string
- optional, Events API URL, default:http://localhost
space
-string
- optional, space name, default:default
adminPort
-int
- optional, port for configuration, default:4001
clientPort
-int
- optional, port for calling the event gateway, default:4000
Example
from eventgateway import EventGateway
eg = EventGateway(url="https://mytenant-myapp.slsgateway.com", space="user")
Available Functions
checkConnection
Used to check the connectivity to the Event Gateway (using the /v1/status
endpoint).
Example
from eventgateway import EventGateway
eg = EventGateway()
if eg.checkConnection():
print("Connection succesfull")
else:
print("Issue while connecting")
printConfig
Utility to print the current configuration.
Example
from eventgateway import EventGateway
eg = EventGateway()
eg.printConfig()
createEventType
Function to create an event type.
Example
from eventgateway import EventGateway
eventtype = {
"name": "http.request"
}
eg = EventGateway()
eg.createEventType(eventtype)
createFunction
Function to create a function trigger.
Example
from eventgateway import EventGateway
function = {
"functionId": "new-user",
"type": "http",
"provider":{
"url": "http://myapp.com/user/new"
}
}
eg = EventGateway()
eg.createFunction(function)
createSubscription
Function to subscribe to a function.
Example
from eventgateway import EventGateway
subscription = {
"functionId": "new-user",
"event": "http",
"method": "POST",
"path": "/user/new",
"eventType": "http.request",
"type": "async"
}
eg = EventGateway()
eg.createSubscription(subscription)
Contribute
TBD
Project details
Release history Release notifications | RSS feed
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 event_gateway_sdk-1.0.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | cbf4c6f0e0cf70a7a84b8edc5552264868b471a4ce6557dab5e37f9aff36f00f |
|
MD5 | e1468bacb1512493694c2297bc82b295 |
|
BLAKE2b-256 | 2dc55bf10948b7a6a8d809de191fa59031a0f435b2a7ff41ed8a14570ff185b4 |