A Crossplane function that lets you compose resources and run operational tasks using Python.
To compose resources, provide a Python script that defines a compose function with this signature:
from crossplane.function.proto.v1 import run_function_pb2 as fnv1 def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse)function-python passes the compose function a request and a response. It pre-populates the response with the results of any previous functions in the pipeline. The compose function should modify the response (rsp), for example to add composed resources.
The RunFunctionRequest and RunFunctionResponse types provided by this SDK are generated from a proto3 protocol buffer schema. Their fields behave similarly to built-in Python types like lists and dictionaries, but there are some differences. Read the generated code documentation to familiarize yourself with the the differences.
Your script has access to function-sdk-python as the crossplane.function module. For example you can import crossplane.function.resource. It also has access to the full Python standard library - use it with care.
apiVersion: apiextensions.crossplane.io/v1 kind: Composition metadata: name: compose-a-resource-with-python spec: compositeTypeRef: apiVersion: example.crossplane.io/v1 kind: XR mode: Pipeline pipeline: - step: compose-a-resource-with-python functionRef: name: function-python input: apiVersion: python.fn.crossplane.io/v1beta1 kind: Script script: | from crossplane.function.proto.v1 import run_function_pb2 as fnv1 def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse): rsp.desired.resources["bucket"].resource.update({ "apiVersion": "s3.aws.upbound.io/v1beta2", "kind": "Bucket", "spec": { "forProvider": { "region": req.observed.composite.resource["spec"]["region"] } }, }) rsp.desired.resources["bucket"].ready = Truefunction-python also supports Crossplane Operations, which are one-time operational tasks that run to completion (like Kubernetes Jobs). For operations, provide a Python script that defines an operate function with this signature:
from crossplane.function.proto.v1 import run_function_pb2 as fnv1 def operate(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse): # Your operational logic here # Set output for operation monitoring rsp.output["result"] = "success" rsp.output["message"] = "Operation completed successfully"apiVersion: ops.crossplane.io/v1alpha1 kind: Operation metadata: name: check-cert-expiry spec: template: spec: pipeline: - step: check-certificate functionRef: name: function-python input: apiVersion: python.fn.crossplane.io/v1beta1 kind: Script script: | from crossplane.function.proto.v1 import run_function_pb2 as fnv1 import datetime def operate(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse): # Example: Check certificate expiration # In real use, you'd retrieve and validate actual certificates # Set operation output rsp.output["check_time"] = datetime.datetime.now().isoformat() rsp.output["status"] = "healthy" rsp.output["message"] = "Certificate expiry check completed"For more complex operations, see the operation examples.
function-python is best for very simple cases. If writing Python inline of YAML becomes unwieldy, consider building a Python function using function-template-python.
![@renovate[bot]](https://avatars.githubusercontent.com/in/2740?s=64&v=4){kind=small}
