Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 96 additions & 1 deletion crossplane/function/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,11 @@
Provides reusable options and a run helper so that every composition
function shares a standard set of flags, environment variables, and defaults.

Standard flags include ``--address``, ``--debug``, ``--insecure``,
``--tls-server-certs-dir``, gRPC message size limits, and ``--ttl``. Each
option also supports a corresponding environment variable (for example
``ADDRESS``, ``DEBUG``, ``TTL``).

Usage in a function's main.py::

import click
Expand All @@ -38,20 +43,96 @@ def cli(cache_size, **kwargs):
sdkcli.run(runner, **kwargs)
"""

import datetime
import functools
import re
from collections.abc import Callable
from typing import TypeVar

import click

from crossplane.function import logging, runtime
from crossplane.function import logging, response, runtime
from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1

F = TypeVar("F", bound=Callable)

DEFAULT_ADDRESS = "0.0.0.0:9443"
DEFAULT_MAX_RECV_MESSAGE_SIZE = 4 # MB

_UNIT_TO_SECONDS = {
"s": 1,
"m": 60,
"h": 3600,
"d": 86400,
}
_DURATION_COMPONENT_RE = re.compile(r"(\d+(?:\.\d+)?)([smhd])")


def parse_duration(value: str) -> datetime.timedelta:
"""Parse a duration string into a :class:`datetime.timedelta`.

Accepts Go-style duration strings (e.g. ``60s``, ``1m``, ``1h30m``) and bare
integers interpreted as seconds (e.g. ``60``).

Args:
value: The duration string to parse.

Returns:
The parsed duration.

Raises:
ValueError: If the string is empty, invalid, or negative.
"""
value = value.strip()
if not value:
msg = "duration must not be empty"
raise ValueError(msg)

if value.isdigit():
return datetime.timedelta(seconds=int(value))

total_seconds = 0.0
pos = 0
for match in _DURATION_COMPONENT_RE.finditer(value):
if match.start() != pos:
msg = f"invalid duration: {value}"
raise ValueError(msg)
total_seconds += float(match.group(1)) * _UNIT_TO_SECONDS[match.group(2)]
pos = match.end()

if pos != len(value):
msg = f"invalid duration: {value}"
raise ValueError(msg)

if total_seconds < 0:
msg = "duration must not be negative"
raise ValueError(msg)

return datetime.timedelta(seconds=total_seconds)


class DurationParamType(click.ParamType):
"""A Click parameter type that parses duration strings."""

name = "duration"

def convert(
self,
value: object,
param: click.Parameter | None,
ctx: click.Context | None,
) -> datetime.timedelta:
"""Convert a CLI value to a :class:`datetime.timedelta`."""
if isinstance(value, datetime.timedelta):
return value
try:
return parse_duration(str(value))
except ValueError as e:
self.fail(str(e), param, ctx)


DURATION = DurationParamType()


def standard_options(func: F) -> F:
"""Apply the standard Composition Function CLI options to a Click command."""
Expand Down Expand Up @@ -101,6 +182,16 @@ def standard_options(func: F) -> F:
envvar="DEBUG",
help="Emit debug logs.",
)
@click.option(
"--ttl",
type=DURATION,
default=None,
show_default="1m",
envvar="TTL",
help="Default TTL for RunFunctionResponses. "
"Controls how long Crossplane may cache the response "
"before re-invoking the function.",
)
@functools.wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
Expand All @@ -117,11 +208,15 @@ def run( # noqa: PLR0913
insecure: bool,
max_recv_message_size: int,
max_send_message_size: int | None,
ttl: datetime.timedelta | None,
) -> None:
"""Start a composition function gRPC server with standard options."""
level = logging.Level.DEBUG if debug else logging.Level.INFO
logging.configure(level=level)

if ttl is not None:
response.set_default_ttl(ttl)

if max_send_message_size is None:
max_send_message_size = max_recv_message_size

Expand Down
26 changes: 24 additions & 2 deletions crossplane/function/response.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,23 +25,45 @@
"""The default TTL for which a RunFunctionResponse may be cached."""
DEFAULT_TTL = datetime.timedelta(minutes=1)

_default_ttl = DEFAULT_TTL


def get_default_ttl() -> datetime.timedelta:
"""Return the process-wide default TTL for RunFunctionResponses."""
return _default_ttl


def set_default_ttl(ttl: datetime.timedelta) -> None:
"""Set the process-wide default TTL for RunFunctionResponses.

Args:
ttl: How long Crossplane may optionally cache responses when no explicit
TTL is passed to :func:`to`.
"""
global _default_ttl # noqa: PLW0603
_default_ttl = ttl


def to(
req: fnv1.RunFunctionRequest,
ttl: datetime.timedelta = DEFAULT_TTL,
ttl: datetime.timedelta | None = None,
) -> fnv1.RunFunctionResponse:
"""Create a response to the supplied request.

Args:
req: The request to respond to.
ttl: How long Crossplane may optionally cache the response.
ttl: How long Crossplane may optionally cache the response. Defaults to
the process-wide default TTL set by :func:`set_default_ttl`.

Returns:
A response to the supplied request.

The request's tag, desired resources, and context is automatically copied to
the response. Using response.to is a good pattern to ensure
"""
if ttl is None:
ttl = get_default_ttl()

dttl = durationpb.Duration()
dttl.FromTimedelta(ttl)
return fnv1.RunFunctionResponse(
Expand Down
191 changes: 191 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,191 @@
# Copyright 2026 The Crossplane Authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import dataclasses
import datetime
import unittest
from unittest import mock

import click
from click.testing import CliRunner
from google.protobuf import duration_pb2 as durationpb
from google.protobuf import json_format

from crossplane.function import cli, response
from crossplane.function.proto.v1 import run_function_pb2 as fnv1


class TestParseDuration(unittest.TestCase):
def test_parse_duration(self) -> None:
@dataclasses.dataclass
class TestCase:
reason: str
value: str
want: datetime.timedelta

cases = [
TestCase(
reason="Bare integers are interpreted as seconds.",
value="60",
want=datetime.timedelta(seconds=60),
),
TestCase(
reason="Seconds suffix should work.",
value="60s",
want=datetime.timedelta(seconds=60),
),
TestCase(
reason="Minutes suffix should work.",
value="1m",
want=datetime.timedelta(minutes=1),
),
TestCase(
reason="Hours suffix should work.",
value="1h",
want=datetime.timedelta(hours=1),
),
TestCase(
reason="Days suffix should work.",
value="1d",
want=datetime.timedelta(days=1),
),
TestCase(
reason="Combined durations should be summed.",
value="1h30m",
want=datetime.timedelta(hours=1, minutes=30),
),
TestCase(
reason="Zero seconds should work.",
value="0s",
want=datetime.timedelta(seconds=0),
),
TestCase(
reason="Zero as a bare integer should work.",
value="0",
want=datetime.timedelta(seconds=0),
),
TestCase(
reason="Fractional seconds should work.",
value="1.5s",
want=datetime.timedelta(seconds=1.5),
),
]

for case in cases:
got = cli.parse_duration(case.value)
self.assertEqual(case.want, got, case.reason)

def test_parse_duration_invalid(self) -> None:
for value in ("", "1x", "-5m", "1m2"):
with self.assertRaises(ValueError, msg=value):
cli.parse_duration(value)


class TestStandardOptions(unittest.TestCase):
def setUp(self) -> None:
self._saved_default_ttl = response.get_default_ttl()

def tearDown(self) -> None:
response.set_default_ttl(self._saved_default_ttl)

def test_run_sets_default_ttl_from_flag(self) -> None:
@click.command()
@cli.standard_options
def main(**kwargs):
cli.run(mock.Mock(), **kwargs)

runner = CliRunner()
with mock.patch("crossplane.function.cli.runtime.serve"):
result = runner.invoke(main, ["--ttl", "5m", "--insecure"])
self.assertEqual(0, result.exit_code, result.output)
self.assertEqual(datetime.timedelta(minutes=5), response.get_default_ttl())

def test_run_sets_default_ttl_from_env(self) -> None:
@click.command()
@cli.standard_options
def main(**kwargs):
cli.run(mock.Mock(), **kwargs)

runner = CliRunner()
with mock.patch("crossplane.function.cli.runtime.serve"):
result = runner.invoke(
main,
["--insecure"],
env={"TTL": "10m"},
)
self.assertEqual(0, result.exit_code, result.output)
self.assertEqual(datetime.timedelta(minutes=10), response.get_default_ttl())

def test_run_leaves_default_ttl_when_flag_omitted(self) -> None:
response.set_default_ttl(response.DEFAULT_TTL)

@click.command()
@cli.standard_options
def main(**kwargs):
cli.run(mock.Mock(), **kwargs)

runner = CliRunner()
with mock.patch("crossplane.function.cli.runtime.serve"):
result = runner.invoke(main, ["--insecure"])
self.assertEqual(0, result.exit_code, result.output)
self.assertEqual(response.DEFAULT_TTL, response.get_default_ttl())

def test_invalid_ttl_flag(self) -> None:
@click.command()
@cli.standard_options
def main(**kwargs):
cli.run(mock.Mock(), **kwargs)

runner = CliRunner()
result = runner.invoke(main, ["--ttl", "1x", "--insecure"])
self.assertNotEqual(0, result.exit_code)

def test_ttl_flag_affects_response_to(self) -> None:
@click.command()
@cli.standard_options
def main(**kwargs):
cli.run(mock.Mock(), **kwargs)

runner = CliRunner()
with mock.patch("crossplane.function.cli.runtime.serve"):
result = runner.invoke(main, ["--ttl", "5m", "--insecure"])
self.assertEqual(0, result.exit_code, result.output)

req = fnv1.RunFunctionRequest(meta=fnv1.RequestMeta(tag="hi"))
got = response.to(req)
want = fnv1.RunFunctionResponse(
meta=fnv1.ResponseMeta(tag="hi", ttl=durationpb.Duration(seconds=60 * 5)),
desired=req.desired,
context=req.context,
)
self.assertEqual(
json_format.MessageToJson(want, sort_keys=True),
json_format.MessageToJson(got, sort_keys=True),
)


class TestDurationParamType(unittest.TestCase):
def test_convert(self) -> None:
param_type = cli.DurationParamType()
got = param_type.convert("5m", None, None)
self.assertEqual(datetime.timedelta(minutes=5), got)

def test_convert_invalid(self) -> None:
param_type = cli.DurationParamType()
with self.assertRaises(click.BadParameter):
param_type.convert("1x", None, None)


if __name__ == "__main__":
unittest.main()
Loading