function.response

  1# Copyright 2023 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"""Utilities for working with RunFunctionResponses."""
 16
 17import datetime
 18
 19from google.protobuf import duration_pb2 as durationpb
 20from google.protobuf import struct_pb2 as structpb
 21
 22import crossplane.function.proto.v1.run_function_pb2 as fnv1
 23from crossplane.function import resource
 24
 25"""The default TTL for which a RunFunctionResponse may be cached."""
 26DEFAULT_TTL = datetime.timedelta(minutes=1)
 27
 28
 29def to(
 30    req: fnv1.RunFunctionRequest,
 31    ttl: datetime.timedelta = DEFAULT_TTL,
 32) -> fnv1.RunFunctionResponse:
 33    """Create a response to the supplied request.
 34
 35    Args:
 36        req: The request to respond to.
 37        ttl: How long Crossplane may optionally cache the response.
 38
 39    Returns:
 40        A response to the supplied request.
 41
 42    The request's tag, desired resources, and context is automatically copied to
 43    the response. Using response.to is a good pattern to ensure
 44    """
 45    dttl = durationpb.Duration()
 46    dttl.FromTimedelta(ttl)
 47    return fnv1.RunFunctionResponse(
 48        meta=fnv1.ResponseMeta(tag=req.meta.tag, ttl=dttl),
 49        desired=req.desired,
 50        context=req.context,
 51    )
 52
 53
 54def normal(rsp: fnv1.RunFunctionResponse, message: str) -> None:
 55    """Add a normal result to the response."""
 56    rsp.results.append(
 57        fnv1.Result(
 58            severity=fnv1.SEVERITY_NORMAL,
 59            message=message,
 60        )
 61    )
 62
 63
 64def warning(rsp: fnv1.RunFunctionResponse, message: str) -> None:
 65    """Add a warning result to the response."""
 66    rsp.results.append(
 67        fnv1.Result(
 68            severity=fnv1.SEVERITY_WARNING,
 69            message=message,
 70        )
 71    )
 72
 73
 74def fatal(rsp: fnv1.RunFunctionResponse, message: str) -> None:
 75    """Add a fatal result to the response."""
 76    rsp.results.append(
 77        fnv1.Result(
 78            severity=fnv1.SEVERITY_FATAL,
 79            message=message,
 80        )
 81    )
 82
 83
 84_STATUS_MAP = {
 85    "True": fnv1.STATUS_CONDITION_TRUE,
 86    "False": fnv1.STATUS_CONDITION_FALSE,
 87    "Unknown": fnv1.STATUS_CONDITION_UNKNOWN,
 88}
 89
 90
 91def set_conditions(
 92    rsp: fnv1.RunFunctionResponse,
 93    *conditions: resource.Condition,
 94) -> None:
 95    """Set one or more conditions on the composite resource (XR).
 96
 97    Args:
 98        rsp: The RunFunctionResponse to update.
 99        *conditions: The conditions to set.
100
101    Each condition is appended to ``rsp.conditions``. Crossplane uses the
102    conditions returned by a function to set custom status conditions on
103    the composite resource.
104
105    The ``last_transition_time`` field of each condition is ignored.
106    Crossplane sets the transition time itself.
107
108    Do not set the ``Ready`` condition type. Crossplane manages it based
109    on resource readiness.
110    """
111    for condition in conditions:
112        c = fnv1.Condition(
113            type=condition.typ,
114            status=_STATUS_MAP.get(condition.status, fnv1.STATUS_CONDITION_UNKNOWN),
115            reason=condition.reason or "",
116        )
117        if condition.message:
118            c.message = condition.message
119        rsp.conditions.append(c)
120
121
122def set_output(rsp: fnv1.RunFunctionResponse, output: dict | structpb.Struct) -> None:
123    """Set the output field in a RunFunctionResponse for operation functions.
124
125    Args:
126        rsp: The RunFunctionResponse to update.
127        output: The output data as a dictionary or protobuf Struct.
128
129    Operation functions can return arbitrary output data that will be written
130    to the Operation's status.pipeline field. This function sets that output
131    on the response.
132    """
133    match output:
134        case dict():
135            rsp.output.CopyFrom(resource.dict_to_struct(output))
136        case structpb.Struct():
137            rsp.output.CopyFrom(output)
138        case _:
139            t = type(output)
140            msg = f"Unsupported output type: {t}"
141            raise TypeError(msg)
142
143
144def require_resources(  # noqa: PLR0913
145    rsp: fnv1.RunFunctionResponse,
146    name: str,
147    api_version: str,
148    kind: str,
149    *,
150    match_name: str | None = None,
151    match_labels: dict[str, str] | None = None,
152    namespace: str | None = None,
153) -> None:
154    """Add a resource requirement to the response.
155
156    Args:
157        rsp: The RunFunctionResponse to update.
158        name: The name to use for this requirement.
159        api_version: The API version of resources to require.
160        kind: The kind of resources to require.
161        match_name: Match a resource by name (mutually exclusive with match_labels).
162        match_labels: Match resources by labels (mutually exclusive with match_name).
163        namespace: The namespace to search in (optional).
164
165    Raises:
166        ValueError: If both match_name and match_labels are provided.
167
168    This tells Crossplane to fetch the specified resources and include them
169    in the next call to the function in req.required_resources[name].
170
171    If neither match_name nor match_labels is provided, all resources of the
172    given api_version and kind are matched.
173    """
174    if match_name is not None and match_labels is not None:
175        msg = "match_name and match_labels are mutually exclusive"
176        raise ValueError(msg)
177
178    selector = fnv1.ResourceSelector(
179        api_version=api_version,
180        kind=kind,
181    )
182
183    if match_name is not None:
184        selector.match_name = match_name
185
186    if match_labels is not None:
187        selector.match_labels.labels.update(match_labels)
188
189    if namespace is not None:
190        selector.namespace = namespace
191
192    rsp.requirements.resources[name].CopyFrom(selector)
193
194
195def require_schema(
196    rsp: fnv1.RunFunctionResponse,
197    name: str,
198    api_version: str,
199    kind: str,
200) -> None:
201    """Add a schema requirement to the response.
202
203    Args:
204        rsp: The RunFunctionResponse to update.
205        name: The name to use for this requirement.
206        api_version: The API version of the resource kind, e.g. "example.org/v1".
207        kind: The kind of resource, e.g. "MyResource".
208
209    This tells Crossplane to fetch the OpenAPI schema for the specified resource
210    kind and include it in the next call to the function in
211    req.required_schemas[name]. Use request.get_required_schema to retrieve it.
212
213    For CRDs, Crossplane returns the spec.versions[].schema.openAPIV3Schema field.
214    If Crossplane cannot find a schema for the requested kind, the schema will be
215    empty (get_required_schema will return None).
216    """
217    selector = fnv1.SchemaSelector(
218        api_version=api_version,
219        kind=kind,
220    )
221    rsp.requirements.schemas[name].CopyFrom(selector)