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    if name not in req.required_resources:
 46        return []
 47
 48    return [
 49        resource.struct_to_dict(item.resource)
 50        for item in req.required_resources[name].items
 51    ]
 52
 53
 54def get_watched_resource(req: fnv1.RunFunctionRequest) -> dict | None:
 55    """Get the watched resource that triggered this operation.
 56
 57    Args:
 58        req: The RunFunctionRequest to check for a watched resource.
 59
 60    Returns:
 61        The watched resource as a dictionary, or None if not found.
 62
 63    When a WatchOperation creates an Operation, it injects the resource that
 64    changed using the special requirement name 'ops.crossplane.io/watched-resource'.
 65    This helper makes it easy to access that resource.
 66    """
 67    watched = get_required_resources(req, "ops.crossplane.io/watched-resource")
 68    return watched[0] if watched else None
 69
 70
 71def get_required_resource(req: fnv1.RunFunctionRequest, name: str) -> dict | None:
 72    """Get a single required resource by name from the request.
 73
 74    Args:
 75        req: The RunFunctionRequest containing required resources.
 76        name: The name of the required resource to get.
 77
 78    Returns:
 79        The first resource as a dictionary, or None if not found.
 80
 81    This is a convenience function for when you know there should be exactly
 82    one resource with the given requirement name.
 83    """
 84    resources = get_required_resources(req, name)
 85    return resources[0] if resources else None
 86
 87
 88def get_credentials(req: fnv1.RunFunctionRequest, name: str) -> Credentials:
 89    """Get the supplied credentials from the request.
 90
 91    Args:
 92        req: The RunFunctionRequest containing credentials.
 93        name: The name of the credentials to get.
 94
 95    Returns:
 96        The requested credentials with type and data.
 97
 98    If the credentials don't exist, returns empty credentials with type "data"
 99    and empty data dictionary.
100    """
101    empty = Credentials(type="data", data={})
102
103    if not req or name not in req.credentials:
104        return empty
105
106    cred = req.credentials[name]
107
108    # Use WhichOneof to determine which field in the oneof is set
109    source_type = cred.WhichOneof("source")
110    if source_type == "credential_data":
111        # Convert bytes data to string data for backward compatibility
112        data = {}
113        for key, value in cred.credential_data.data.items():
114            data[key] = value.decode("utf-8")
115        return Credentials(type="credential_data", data=data)
116
117    # If no recognized source type is set, return empty
118    return empty