Add convenience helpers for conditions, status, and resource names

crossplane/function/resource.py

@@ -16,6 +16,7 @@
 
 import dataclasses
 import datetime
+import hashlib
 
 import pydantic
 from google.protobuf import json_format
@@ -59,6 +60,25 @@ def update(r: fnv1.Resource, source: dict | structpb.Struct | pydantic.BaseModel
             raise TypeError(msg)
 
 
+def update_status(
+    r: fnv1.Resource,
+    status: dict | pydantic.BaseModel,
+) -> None:
+    """Update a resource's status.
+
+    Args:
+        r: A composite or composed resource to update.
+        status: The status to set, as a dictionary or Pydantic model.
+
+    Sets ``r.resource.status`` from the supplied status. When the status
+    is a Pydantic model, fields set to their default value are excluded,
+    matching the behavior of :func:`update`.
+    """
+    if isinstance(status, pydantic.BaseModel):
+        status = status.model_dump(exclude_defaults=True, warnings=False)
+    update(r, {"status": status})
+
+
 def dict_to_struct(d: dict) -> structpb.Struct:
     """Create a Struct well-known type from the supplied dict.
 
@@ -99,21 +119,41 @@ class Condition:
     last_transition_time: datetime.time | None = None
 
 
-def get_condition(resource: structpb.Struct, typ: str) -> Condition:
+def get_condition(
+    resource: structpb.Struct | fnv1.Resource | None,
+    typ: str,
+) -> Condition:
     """Get the supplied status condition of the supplied resource.
 
     Args:
-        resource: A Crossplane resource.
+        resource: A Crossplane resource. Can be a protobuf Struct (the raw
+            resource), an fnv1.Resource wrapper, or None. When an
+            fnv1.Resource is supplied, the Struct is extracted automatically.
+            When None is supplied, an unknown condition is returned.
         typ: The type of status condition to get (e.g. Ready).
 
     Returns:
         The requested status condition.
 
     A status condition is always returned. If the status condition isn't present
     in the supplied resource, a condition with status "Unknown" is returned.
+
+    Accepting fnv1.Resource and None makes it safe to pass the result of a
+    protobuf map ``.get()`` call directly. This avoids auto-vivification, which
+    silently inserts a default entry when using bracket access on a missing
+    key::
+
+        # Safe — .get() returns None without mutating the map.
+        c = get_condition(req.observed.resources.get("bucket"), "Ready")
+
+        # Unsafe — bracket access auto-vivifies an empty Resource.
+        c = get_condition(req.observed.resources["bucket"].resource, "Ready")
     """
     unknown = Condition(typ=typ, status="Unknown")
 
+    if isinstance(resource, fnv1.Resource):
+        resource = resource.resource
+
     if not resource or "status" not in resource:
         return unknown
 
@@ -140,3 +180,37 @@ def get_condition(resource: structpb.Struct, typ: str) -> Condition:
         return condition
 
     return unknown
+
+
+_DNS_LABEL_MAX = 63
+_HASH_LEN = 5
+
+
+def child_name(*parts: str, sep: str = "-") -> str:
+    """Build a deterministic, DNS-label-safe name for a child resource.
+
+    Args:
+        *parts: Name components to join (e.g. parent name, suffix).
+        sep: Separator between parts. Defaults to "-".
+
+    Returns:
+        A name that is at most 63 characters long.
+
+    Composition functions often derive child resource names from a parent
+    name and a discriminator. The resulting name must be a valid DNS label
+    (at most 63 characters). This function joins the parts, appends a
+    deterministic 5-character hash suffix for uniqueness, and truncates
+    the prefix to fit within the limit.
+
+    The hash suffix is always appended, even for short names, so that
+    names are visually consistent regardless of length::
+
+        child_name("my-xr", "bucket")       # "my-xr-bucket-a1b2c"
+        child_name("my-very-long-xr-name",
+                   "with-a-very-long-suffix") # truncated to 63 chars
+    """
+    full = sep.join(parts)
+    h = hashlib.sha256(full.encode()).hexdigest()[:_HASH_LEN]
+    max_prefix = _DNS_LABEL_MAX - _HASH_LEN - 1
+    prefix = full[:max_prefix].rstrip(sep)
+    return f"{prefix}{sep}{h}"

crossplane/function/response.py

@@ -81,6 +81,44 @@ def fatal(rsp: fnv1.RunFunctionResponse, message: str) -> None:
     )
 
 
+_STATUS_MAP = {
+    "True": fnv1.STATUS_CONDITION_TRUE,
+    "False": fnv1.STATUS_CONDITION_FALSE,
+    "Unknown": fnv1.STATUS_CONDITION_UNKNOWN,
+}
+
+
+def set_conditions(
+    rsp: fnv1.RunFunctionResponse,
+    *conditions: resource.Condition,
+) -> None:
+    """Set one or more conditions on the composite resource (XR).
+
+    Args:
+        rsp: The RunFunctionResponse to update.
+        *conditions: The conditions to set.
+
+    Each condition is appended to ``rsp.conditions``. Crossplane uses the
+    conditions returned by a function to set custom status conditions on
+    the composite resource.
+
+    The ``last_transition_time`` field of each condition is ignored.
+    Crossplane sets the transition time itself.
+
+    Do not set the ``Ready`` condition type. Crossplane manages it based
+    on resource readiness.
+    """
+    for condition in conditions:
+        c = fnv1.Condition(
+            type=condition.typ,
+            status=_STATUS_MAP.get(condition.status, fnv1.STATUS_CONDITION_UNKNOWN),
+            reason=condition.reason or "",
+        )
+        if condition.message:
+            c.message = condition.message
+        rsp.conditions.append(c)
+
+
 def set_output(rsp: fnv1.RunFunctionResponse, output: dict | structpb.Struct) -> None:
     """Set the output field in a RunFunctionResponse for operation functions.
 

tests/test_resource.py

@@ -29,6 +29,56 @@ class TestResource(unittest.TestCase):
     def setUp(self) -> None:
         logging.configure(level=logging.Level.DISABLED)
 
+    def test_update_status(self) -> None:
+        @dataclasses.dataclass
+        class TestCase:
+            reason: str
+            r: fnv1.Resource
+            status: dict | pydantic.BaseModel
+            want: dict
+
+        cases = [
+            TestCase(
+                reason="Setting status from a dict should work.",
+                r=fnv1.Resource(
+                    resource=resource.dict_to_struct(
+                        {"apiVersion": "example.org", "kind": "XR"}
+                    ),
+                ),
+                status={"ready": True},
+                want={
+                    "apiVersion": "example.org",
+                    "kind": "XR",
+                    "status": {"ready": True},
+                },
+            ),
+            TestCase(
+                reason="Setting status from a Pydantic model should work.",
+                r=fnv1.Resource(
+                    resource=resource.dict_to_struct(
+                        {"apiVersion": "example.org", "kind": "XR"}
+                    ),
+                ),
+                status=v1beta2.ForProvider(region="us-west-2"),
+                want={
+                    "apiVersion": "example.org",
+                    "kind": "XR",
+                    "status": {"region": "us-west-2"},
+                },
+            ),
+            TestCase(
+                reason="Setting status on an empty resource should work.",
+                r=fnv1.Resource(),
+                status={"replicas": 3},
+                want={"status": {"replicas": 3}},
+            ),
+        ]
+
+        for case in cases:
+            resource.update_status(case.r, case.status)
+            got = resource.struct_to_dict(case.r.resource)
+            self.assertEqual(case.want, got, case.reason)
+
     def test_add(self) -> None:
         @dataclasses.dataclass
         class TestCase:
@@ -112,11 +162,17 @@ def test_get_condition(self) -> None:
         @dataclasses.dataclass
         class TestCase:
             reason: str
-            res: structpb.Struct
+            res: structpb.Struct | fnv1.Resource | None
             typ: str
             want: resource.Condition
 
         cases = [
+            TestCase(
+                reason="Return an unknown condition if the resource is None.",
+                res=None,
+                typ="Ready",
+                want=resource.Condition(typ="Ready", status="Unknown"),
+            ),
             TestCase(
                 reason="Return an unknown condition if the resource has no status.",
                 res=resource.dict_to_struct({}),
@@ -197,6 +253,31 @@ class TestCase:
                     ),
                 ),
             ),
+            TestCase(
+                reason="Unwrap an fnv1.Resource to get the condition from its Struct.",
+                res=fnv1.Resource(
+                    resource=resource.dict_to_struct(
+                        {
+                            "status": {
+                                "conditions": [
+                                    {
+                                        "type": "Ready",
+                                        "status": "True",
+                                    }
+                                ]
+                            }
+                        }
+                    ),
+                ),
+                typ="Ready",
+                want=resource.Condition(typ="Ready", status="True"),
+            ),
+            TestCase(
+                reason="Return an unknown condition from an empty fnv1.Resource.",
+                res=fnv1.Resource(),
+                typ="Ready",
+                want=resource.Condition(typ="Ready", status="Unknown"),
+            ),
         ]
 
         for case in cases:
@@ -324,6 +405,59 @@ class TestCase:
             got = resource.struct_to_dict(case.s)
             self.assertEqual(case.want, got, "-want, +got")
 
+    def test_child_name(self) -> None:
+        @dataclasses.dataclass
+        class TestCase:
+            reason: str
+            parts: list[str]
+            want: str
+
+        cases = [
+            TestCase(
+                reason="A short name should be joined with a hash suffix.",
+                parts=["my-xr", "bucket"],
+                want="my-xr-bucket-05ecb",
+            ),
+            TestCase(
+                reason="A single part should get a hash suffix.",
+                parts=["my-xr"],
+                want="my-xr-9d53f",
+            ),
+            TestCase(
+                reason="A long name should be truncated to fit within 63 characters.",
+                parts=["a" * 40, "b" * 40],
+                want="a" * 40 + "-" + "b" * 16 + "-" + "f5e42",
+            ),
+            TestCase(
+                reason="A name that would end with a trailing separator "
+                "after truncation should have the separator stripped.",
+                parts=["a" * 56 + "-", "x"],
+                # Without stripping, this would be "aaa..a--<hash>".
+                # The trailing separator from the truncation is stripped.
+                want="a" * 56 + "-" + "995eb",
+            ),
+            TestCase(
+                reason="The same inputs should always produce the same name.",
+                parts=["parent", "child"],
+                want="parent-child-2f0c9",
+            ),
+        ]
+
+        for case in cases:
+            got = resource.child_name(*case.parts)
+            self.assertEqual(case.want, got, case.reason)
+            self.assertLessEqual(len(got), 63, case.reason)
+
+    def test_child_name_deterministic(self) -> None:
+        a = resource.child_name("parent", "child")
+        b = resource.child_name("parent", "child")
+        self.assertEqual(a, b)
+
+    def test_child_name_unique(self) -> None:
+        a = resource.child_name("parent", "child-a")
+        b = resource.child_name("parent", "child-b")
+        self.assertNotEqual(a, b)
+
 
 if __name__ == "__main__":
     unittest.main()