Add support for operations

Signed-off-by: Nic Cope <[email protected]>

Author: negz

README.md

@@ -2,9 +2,11 @@
 
 [![CI](https://github.com/crossplane-contrib/function-python/actions/workflows/ci.yml/badge.svg)](https://github.com/crossplane-contrib/function-python/actions/workflows/ci.yml)
 
-A Crossplane composition function that lets you compose resources using Python.
+A Crossplane composition function that lets you compose resources and run operational tasks using Python.
 
-Provide a Python script that defines a `compose` function with this signature:
+## Composition Functions
+
+For traditional resource composition, provide a Python script that defines a `compose` function with this signature:
 
 ```python
 from crossplane.function.proto.v1 import run_function_pb2 as fnv1
@@ -60,6 +62,61 @@ spec:
             rsp.desired.resources["bucket"].ready = True
 ```
 
+## Operations Functions (Alpha)
+
+`function-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:
+
+```python
+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"
+```
+
+### Operation Example
+
+```yaml
+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"
+```
+
+Operations support several types:
+- **Operation**: One-time tasks
+- **CronOperation**: Scheduled recurring tasks  
+- **WatchOperation**: Reactive tasks triggered by resource changes
+
+For more complex operations, see the [operation examples](example/operation/).
+
+## Usage Notes
+
 `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].

function/fn.py

@@ -17,7 +17,9 @@ def __init__(self):
         self.log = logging.get_logger()
 
     async def RunFunction(
-        self, req: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext
+        self,
+        req: fnv1.RunFunctionRequest,
+        _: grpc.aio.ServicerContext,
     ) -> fnv1.RunFunctionResponse:
         """Run the function."""
         log = self.log.bind(tag=req.meta.tag)
@@ -31,7 +33,25 @@ async def RunFunction(
 
         log.debug("Running script", script=req.input["script"])
         script = load_module("script", req.input["script"])
-        script.compose(req, rsp)
+
+        has_compose = hasattr(script, "compose")
+        has_operate = hasattr(script, "operate")
+
+        match (has_compose, has_operate):
+            case (True, True):
+                msg = "script must define only one function: compose or operate"
+                log.debug(msg)
+                response.fatal(rsp, msg)
+            case (True, False):
+                log.debug("running composition function")
+                script.compose(req, rsp)
+            case (False, True):
+                log.debug("running operation function")
+                script.operate(req, rsp)
+            case (False, False):
+                msg = "script must define a compose or operate function"
+                log.debug(msg)
+                response.fatal(rsp, msg)
 
         return rsp
 

function/main.py

@@ -27,8 +27,10 @@
 @click.option(
     "--insecure",
     is_flag=True,
-    help="Run without mTLS credentials. "
-    "If you supply this flag --tls-certs-dir will be ignored.",
+    help=(
+        "Run without mTLS credentials. "
+        "If you supply this flag --tls-certs-dir will be ignored."
+    ),
 )
 def cli(debug: bool, address: str, tls_certs_dir: str, insecure: bool) -> None:  # noqa:FBT001  # We only expect callers via the CLI.
     """A Crossplane composition function."""

package/crossplane.yaml

@@ -3,4 +3,7 @@ apiVersion: meta.pkg.crossplane.io/v1beta1
 kind: Function
 metadata:
   name: function-python
-spec: {}
+spec:
+  capabilities:
+  - composition
+  - operation

pyproject.toml

@@ -48,6 +48,11 @@ dependencies = ["ipython==9.4.0"]
 [tool.hatch.envs.default.scripts]
 development = "python function/main.py --insecure --debug"
 
+# This special environment is used by hatch fmt.
+[tool.hatch.envs.hatch-static-analysis]
+dependencies = ["ruff==0.12.7"]
+config-path = "none"            # Disable Hatch's default Ruff config.
+
 [tool.hatch.envs.lint]
 type = "virtual"
 detached = true
@@ -67,9 +72,7 @@ unit = "python -m unittest tests/*.py"
 [tool.ruff]
 target-version = "py311"
 exclude = ["function/proto/*"]
-
-[tool.ruff.lint]
-select = [
+lint.select = [
   "A",
   "ARG",
   "ASYNC",
@@ -99,7 +102,7 @@ select = [
   "W",
   "YTT",
 ]
-ignore = ["ISC001"] # Ruff warns this is incompatible with ruff format.
+lint.ignore = ["ISC001"] # Ruff warns this is incompatible with ruff format.
 
 [tool.ruff.lint.per-file-ignores]
 "tests/*" = ["D"] # Don't require docstrings for tests.

tests/test_fn.py

@@ -9,7 +9,7 @@
 
 from function import fn
 
-script = """
+composition_script = """
 from crossplane.function.proto.v1 import run_function_pb2 as fnv1
 
 def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
@@ -24,6 +24,32 @@ def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
     })
 """
 
+operation_script = """
+from crossplane.function.proto.v1 import run_function_pb2 as fnv1
+
+def operate(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
+    # Set output for operation monitoring
+    rsp.output["result"] = "success"
+    rsp.output["message"] = "Operation completed successfully"
+"""
+
+both_functions_script = """
+from crossplane.function.proto.v1 import run_function_pb2 as fnv1
+
+def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
+    pass
+
+def operate(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
+    pass
+"""
+
+no_function_script = """
+from crossplane.function.proto.v1 import run_function_pb2 as fnv1
+
+def some_other_function():
+    pass
+"""
+
 
 class TestFunctionRunner(unittest.IsolatedAsyncioTestCase):
     def setUp(self) -> None:
@@ -41,9 +67,9 @@ class TestCase:
 
         cases = [
             TestCase(
-                reason="The function should return the input as a result.",
+                reason="Function should run composition scripts with compose().",
                 req=fnv1.RunFunctionRequest(
-                    input=resource.dict_to_struct({"script": script})
+                    input=resource.dict_to_struct({"script": composition_script}),
                 ),
                 want=fnv1.RunFunctionResponse(
                     meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)),
@@ -77,6 +103,69 @@ class TestCase:
                 "-want, +got",
             )
 
+    async def test_run_operation(self) -> None:
+        @dataclasses.dataclass
+        class TestCase:
+            reason: str
+            req: fnv1.RunFunctionRequest
+            want: fnv1.RunFunctionResponse
+
+        cases = [
+            TestCase(
+                reason="Function should run operation scripts with operate().",
+                req=fnv1.RunFunctionRequest(
+                    input=resource.dict_to_struct({"script": operation_script}),
+                ),
+                want=fnv1.RunFunctionResponse(
+                    meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)),
+                    desired=fnv1.State(),
+                    context=structpb.Struct(),
+                    output=resource.dict_to_struct(
+                        {
+                            "result": "success",
+                            "message": "Operation completed successfully",
+                        }
+                    ),
+                ),
+            ),
+        ]
+
+        runner = fn.FunctionRunner()
+
+        for case in cases:
+            got = await runner.RunFunction(case.req, None)
+            self.assertEqual(
+                json_format.MessageToDict(case.want),
+                json_format.MessageToDict(got),
+                "-want, +got",
+            )
+
+    async def test_error_both_functions(self) -> None:
+        """Test that having both compose and operate functions returns an error."""
+        runner = fn.FunctionRunner()
+        script = both_functions_script
+        req = fnv1.RunFunctionRequest(input=resource.dict_to_struct({"script": script}))
+
+        got = await runner.RunFunction(req, None)
+
+        # Should have a fatal error
+        self.assertEqual(len(got.results), 1)
+        self.assertEqual(got.results[0].severity, fnv1.Severity.SEVERITY_FATAL)
+        self.assertIn("only one function: compose or operate", got.results[0].message)
+
+    async def test_error_no_functions(self) -> None:
+        """Test that having neither compose nor operate functions returns an error."""
+        runner = fn.FunctionRunner()
+        script = no_function_script
+        req = fnv1.RunFunctionRequest(input=resource.dict_to_struct({"script": script}))
+
+        got = await runner.RunFunction(req, None)
+
+        # Should have a fatal error
+        self.assertEqual(len(got.results), 1)
+        self.assertEqual(got.results[0].severity, fnv1.Severity.SEVERITY_FATAL)
+        self.assertIn("compose or operate function", got.results[0].message)
+
 
 if __name__ == "__main__":
     unittest.main()