function.request

  1# Copyright 2025 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 RunFunctionRequests."""
 16
 17import dataclasses
 18
 19import crossplane.function.proto.v1.run_function_pb2 as fnv1
 20from crossplane.function import resource
 21
 22
 23@dataclasses.dataclass
 24class Credentials:
 25    """Credentials."""
 26
 27    type: str
 28    data: dict
 29
 30
 31def get_required_resources(req: fnv1.RunFunctionRequest, name: str) -> list[dict]:
 32    """Get required resources by name from the request.
 33
 34    Args:
 35        req: The RunFunctionRequest containing required resources.
 36        name: The name of the required resource set to get.
 37
 38    Returns:
 39        A list of resources as dictionaries. Empty list if not found.
 40
 41    Required resources are previously called "extra resources" in composition
 42    functions. For operation functions, there are no observed resources, so
 43    all resources are "required" resources that the function requested.
 44
 45    Note: This returns an empty list both when the requirement hasn't been
 46    resolved yet, and when it was resolved but no resources matched. To
 47    distinguish between these cases, check whether Crossplane resolved the
 48    requirement using `name in req.required_resources`:
 49
 50        # Always declare requirements - Crossplane considers them satisfied
 51        # when they stabilize across calls.
 52        response.require_resources(rsp, name, ...)
 53
 54        if name in req.required_resources:
 55            resources = request.get_required_resources(req, name)
 56            if not resources:
 57                # Crossplane resolved the requirement, but found no matches
 58                response.fatal(rsp, "no matching resources found")
 59    """
 60    if name not in req.required_resources:
 61        return []
 62
 63    return [
 64        resource.struct_to_dict(item.resource)
 65        for item in req.required_resources[name].items
 66    ]
 67
 68
 69def get_watched_resource(req: fnv1.RunFunctionRequest) -> dict | None:
 70    """Get the watched resource that triggered this operation.
 71
 72    Args:
 73        req: The RunFunctionRequest to check for a watched resource.
 74
 75    Returns:
 76        The watched resource as a dictionary, or None if not found.
 77
 78    When a WatchOperation creates an Operation, it injects the resource that
 79    changed using the special requirement name 'ops.crossplane.io/watched-resource'.
 80    This helper makes it easy to access that resource.
 81    """
 82    watched = get_required_resources(req, "ops.crossplane.io/watched-resource")
 83    return watched[0] if watched else None
 84
 85
 86def get_required_resource(req: fnv1.RunFunctionRequest, name: str) -> dict | None:
 87    """Get a single required resource by name from the request.
 88
 89    Args:
 90        req: The RunFunctionRequest containing required resources.
 91        name: The name of the required resource to get.
 92
 93    Returns:
 94        The first resource as a dictionary, or None if not found.
 95
 96    This is a convenience function for when you know there should be exactly
 97    one resource with the given requirement name.
 98    """
 99    resources = get_required_resources(req, name)
100    return resources[0] if resources else None
101
102
103def get_credentials(req: fnv1.RunFunctionRequest, name: str) -> Credentials:
104    """Get the supplied credentials from the request.
105
106    Args:
107        req: The RunFunctionRequest containing credentials.
108        name: The name of the credentials to get.
109
110    Returns:
111        The requested credentials with type and data.
112
113    If the credentials don't exist, returns empty credentials with type "data"
114    and empty data dictionary.
115    """
116    empty = Credentials(type="data", data={})
117
118    if not req or name not in req.credentials:
119        return empty
120
121    cred = req.credentials[name]
122
123    # Use WhichOneof to determine which field in the oneof is set
124    source_type = cred.WhichOneof("source")
125    if source_type == "credential_data":
126        # Convert bytes data to string data for backward compatibility
127        data = {}
128        for key, value in cred.credential_data.data.items():
129            data[key] = value.decode("utf-8")
130        return Credentials(type="credential_data", data=data)
131
132    # If no recognized source type is set, return empty
133    return empty
134
135
136def advertises_capabilities(req: fnv1.RunFunctionRequest) -> bool:
137    """Check whether Crossplane advertises its capabilities.
138
139    Args:
140        req: The RunFunctionRequest to check.
141
142    Returns:
143        True if Crossplane advertises its capabilities.
144
145    Crossplane v2.2 and later advertise their capabilities in the request
146    metadata. If this returns False, the calling Crossplane predates capability
147    advertisement and has_capability will always return False, even for features
148    the older Crossplane does support.
149
150        if not request.advertises_capabilities(req):
151            # Pre-v2.2 Crossplane, capabilities are unknown.
152            ...
153        elif request.has_capability(req, fnv1.CAPABILITY_REQUIRED_SCHEMAS):
154            response.require_schema(rsp, "xr", xr_api_version, xr_kind)
155    """
156    return fnv1.CAPABILITY_CAPABILITIES in req.meta.capabilities
157
158
159def has_capability(
160    req: fnv1.RunFunctionRequest,
161    cap: fnv1.Capability.ValueType,
162) -> bool:
163    """Check whether Crossplane advertises a particular capability.
164
165    Args:
166        req: The RunFunctionRequest to check.
167        cap: The capability to check for, e.g. fnv1.CAPABILITY_REQUIRED_SCHEMAS.
168
169    Returns:
170        True if the capability is present in the request metadata.
171
172    Crossplane sends its capabilities in the request metadata. Functions can use
173    this to determine whether Crossplane will honor certain fields in their
174    response, or populate certain fields in their request.
175
176    Use advertises_capabilities to check whether Crossplane advertises its
177    capabilities at all. If it doesn't, has_capability always returns False even
178    for features the older Crossplane does support.
179
180        if request.has_capability(req, fnv1.CAPABILITY_REQUIRED_SCHEMAS):
181            response.require_schema(rsp, "xr", xr_api_version, xr_kind)
182    """
183    return cap in req.meta.capabilities
184
185
186def get_required_schema(req: fnv1.RunFunctionRequest, name: str) -> dict | None:
187    """Get a required OpenAPI schema by name from the request.
188
189    Args:
190        req: The RunFunctionRequest containing required schemas.
191        name: The name of the required schema to get.
192
193    Returns:
194        The OpenAPI v3 schema as a dictionary, or None if not found.
195
196    Note: This returns None both when the requirement hasn't been resolved yet,
197    and when it was resolved but the schema wasn't found. To distinguish between
198    these cases, check whether Crossplane resolved the requirement using
199    `name in req.required_schemas`:
200
201        # Always declare requirements - Crossplane considers them satisfied
202        # when they stabilize across calls.
203        response.require_schema(rsp, name, "example.org/v1", "MyKind")
204
205        if name in req.required_schemas:
206            schema = request.get_required_schema(req, name)
207            if schema is None:
208                # Crossplane resolved the requirement, but couldn't find it
209                response.fatal(rsp, "schema not found")
210
211    The returned schema can be used with libraries like openapi-schema-validator
212    or jsonschema for validation:
213
214        schema = request.get_required_schema(req, "my-schema")
215        if schema:
216            from openapi_schema_validator import validate
217            validate(resource, schema)
218    """
219    if name not in req.required_schemas:
220        return None
221
222    schema = req.required_schemas[name]
223    if not schema.HasField("openapi_v3"):
224        return None
225
226    return resource.struct_to_dict(schema.openapi_v3)