This framework provides a mechanism to write Function-as-a-Service style code in Python for handling HTTP events, including CloudEvents delivered via HTTP.
This framework is primarily intended to work with Knative, but also works to provide a generic server for handling CloudEvents over HTTP (e.g. from Kubernetes or on a local machine).
The framework uses reflection to find a suitable function to wrap; it should not be necessary to import any of the following modules in your own code unless you want (e.g. for type definitions):
framework(this module; on PyPi ashttp-containerize)flaskcloudevents
Instead, simply ensure that you have a single non-_ prefixed function which
uses some combination of the following:
- HTTP request arguments (named
req,request,body,headersor of theflask.Requesttype) - CloudEvent arguments (named
event,payload,data,attributesor of thecloudevents.sdk.event.v1.Eventtype)
Usage:
import logging
from typing import Any
counter = 0
def DoEvent(data: Any, attributes: dict, req: Any):
global counter
counter = counter + 1
logging.info(f"Got data: {data}")
logging.info(f"From {req.origin}, my {counter}th request!")
attributes["type"] = "com.example.reply"
attributes["datacontenttype"] = "text/plain"
return attributes, "It's a demo"pack build pytestapp --buildpack ekanderson/pyfun:0.1.1 --builder paketobuildpacks/builder:baseYou can then invoke it via:
docker run --rm -p 8080:8080 -e 8080 pytestappIt's also expected that you should be able (soon) to run this via kn-plugin-func:
kn func create -l python -t http
kn func run