function.cli

  1# Copyright 2026 The Crossplane Authors.
  2#
  3# Licensed under the Apache License, Version 2.0 (the "License");
  4# you may not use this file except in compliance with the License.
  5# You may obtain a copy of the License at
  6#
  7#     http://www.apache.org/licenses/LICENSE-2.0
  8#
  9# Unless required by applicable law or agreed to in writing, software
 10# distributed under the License is distributed on an "AS IS" BASIS,
 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12# See the License for the specific language governing permissions and
 13# limitations under the License.
 14
 15"""Standard CLI options for Python composition functions.
 16
 17Provides reusable options and a run helper so that every composition
 18function shares a standard set of flags, environment variables, and defaults.
 19
 20Standard flags include ``--address``, ``--debug``, ``--insecure``,
 21``--tls-server-certs-dir``, gRPC message size limits, and ``--ttl``. Each
 22option also supports a corresponding environment variable (for example
 23``ADDRESS``, ``DEBUG``, ``TTL``).
 24
 25Usage in a function's main.py::
 26
 27    import click
 28    from crossplane.function import cli as sdkcli
 29    from function import fn
 30
 31    @click.command()
 32    @sdkcli.standard_options
 33    def cli(**kwargs):
 34        sdkcli.run(fn.FunctionRunner(), **kwargs)
 35
 36To add custom options, stack them with the decorator::
 37
 38    @click.command()
 39    @sdkcli.standard_options
 40    @click.option("--cache-size", default=100, envvar="CACHE_SIZE")
 41    def cli(cache_size, **kwargs):
 42        runner = fn.FunctionRunner(cache_size=cache_size)
 43        sdkcli.run(runner, **kwargs)
 44"""
 45
 46import datetime
 47import functools
 48import re
 49from collections.abc import Callable
 50from typing import TypeVar
 51
 52import click
 53
 54from crossplane.function import logging, response, runtime
 55from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1
 56
 57F = TypeVar("F", bound=Callable)
 58
 59DEFAULT_ADDRESS = "0.0.0.0:9443"
 60DEFAULT_MAX_RECV_MESSAGE_SIZE = 4  # MB
 61
 62_UNIT_TO_SECONDS = {
 63    "s": 1,
 64    "m": 60,
 65    "h": 3600,
 66    "d": 86400,
 67}
 68_DURATION_COMPONENT_RE = re.compile(r"(\d+(?:\.\d+)?)([smhd])")
 69
 70
 71def parse_duration(value: str) -> datetime.timedelta:
 72    """Parse a duration string into a :class:`datetime.timedelta`.
 73
 74    Accepts Go-style duration strings (e.g. ``60s``, ``1m``, ``1h30m``) and bare
 75    integers interpreted as seconds (e.g. ``60``).
 76
 77    Args:
 78        value: The duration string to parse.
 79
 80    Returns:
 81        The parsed duration.
 82
 83    Raises:
 84        ValueError: If the string is empty, invalid, or negative.
 85    """
 86    value = value.strip()
 87    if not value:
 88        msg = "duration must not be empty"
 89        raise ValueError(msg)
 90
 91    if value.isdigit():
 92        return datetime.timedelta(seconds=int(value))
 93
 94    total_seconds = 0.0
 95    pos = 0
 96    for match in _DURATION_COMPONENT_RE.finditer(value):
 97        if match.start() != pos:
 98            msg = f"invalid duration: {value}"
 99            raise ValueError(msg)
100        total_seconds += float(match.group(1)) * _UNIT_TO_SECONDS[match.group(2)]
101        pos = match.end()
102
103    if pos != len(value):
104        msg = f"invalid duration: {value}"
105        raise ValueError(msg)
106
107    if total_seconds < 0:
108        msg = "duration must not be negative"
109        raise ValueError(msg)
110
111    return datetime.timedelta(seconds=total_seconds)
112
113
114class DurationParamType(click.ParamType):
115    """A Click parameter type that parses duration strings."""
116
117    name = "duration"
118
119    def convert(
120        self,
121        value: object,
122        param: click.Parameter | None,
123        ctx: click.Context | None,
124    ) -> datetime.timedelta:
125        """Convert a CLI value to a :class:`datetime.timedelta`."""
126        if isinstance(value, datetime.timedelta):
127            return value
128        try:
129            return parse_duration(str(value))
130        except ValueError as e:
131            self.fail(str(e), param, ctx)
132
133
134DURATION = DurationParamType()
135
136
137def standard_options(func: F) -> F:
138    """Apply the standard Composition Function CLI options to a Click command."""
139
140    @click.option(
141        "--max-send-message-size",
142        type=int,
143        default=None,
144        envvar="MAX_SEND_MESSAGE_SIZE",
145        help="Maximum size of sent gRPC messages in MB. "
146        "Defaults to --max-recv-message-size.",
147    )
148    @click.option(
149        "--max-recv-message-size",
150        "--max-grpc-message-size",
151        type=int,
152        default=DEFAULT_MAX_RECV_MESSAGE_SIZE,
153        show_default=True,
154        envvar=["MAX_RECV_MESSAGE_SIZE", "MAX_GRPC_MESSAGE_SIZE"],
155        help="Maximum size of received gRPC messages in MB.",
156    )
157    @click.option(
158        "--insecure",
159        is_flag=True,
160        envvar="INSECURE",
161        help="Run without mTLS credentials. "
162        "If you supply this flag --tls-server-certs-dir will be ignored.",
163    )
164    @click.option(
165        "--tls-server-certs-dir",
166        "--tls-certs-dir",
167        "tls_certs_dir",
168        envvar="TLS_SERVER_CERTS_DIR",
169        help="Serve using mTLS certificates.",
170    )
171    @click.option(
172        "--address",
173        default=DEFAULT_ADDRESS,
174        show_default=True,
175        envvar="ADDRESS",
176        help="Address at which to listen for gRPC connections.",
177    )
178    @click.option(
179        "--debug",
180        "-d",
181        is_flag=True,
182        envvar="DEBUG",
183        help="Emit debug logs.",
184    )
185    @click.option(
186        "--ttl",
187        type=DURATION,
188        default=None,
189        show_default="1m",
190        envvar="TTL",
191        help="Default TTL for RunFunctionResponses. "
192        "Controls how long Crossplane may cache the response "
193        "before re-invoking the function.",
194    )
195    @functools.wraps(func)
196    def wrapper(*args, **kwargs):
197        return func(*args, **kwargs)
198
199    return wrapper
200
201
202def run(  # noqa: PLR0913
203    function_runner: grpcv1.FunctionRunnerServiceServicer,
204    *,
205    debug: bool,
206    address: str,
207    tls_certs_dir: str | None,
208    insecure: bool,
209    max_recv_message_size: int,
210    max_send_message_size: int | None,
211    ttl: datetime.timedelta | None,
212) -> None:
213    """Start a composition function gRPC server with standard options."""
214    level = logging.Level.DEBUG if debug else logging.Level.INFO
215    logging.configure(level=level)
216
217    if ttl is not None:
218        response.set_default_ttl(ttl)
219
220    if max_send_message_size is None:
221        max_send_message_size = max_recv_message_size
222
223    options = [
224        ("grpc.max_receive_message_length", max_recv_message_size * 1024 * 1024),
225        ("grpc.max_send_message_length", max_send_message_size * 1024 * 1024),
226    ]
227
228    runtime.serve(
229        function_runner,
230        address,
231        creds=runtime.load_credentials(tls_certs_dir),
232        insecure=insecure,
233        options=options,
234    )