Skip to main content

Create a composition with Python

Upbound Crossplane allows you to choose how you want to write your composition logic based on your preferred language.

You can choose:

Go - High performance. IDE support with full type safety.

Go Templates - Good for YAML-like configurations. IDE support with YAML language server.

KCL - Concise. Good for transitioning from another configuration language like HCL. IDE support with language server.

Python (this guide) - Highly accessible, supports complex logic. Provides type hints and autocompletion in your IDE.

Overview​

This guide explains how to create compositions that turn your XRs into actual cloud resources. Compositions allow you to implement the business logic that powers your control plane.

important

This guide assumes you're familiar with Python. If you'd like to become more familiar with Python, the official Python tutorial is a good place to start.

Use this guide after you define your API schema and need to write the logic that creates and manages the underlying resources.

Prerequisites​

Before you begin, make sure:

  • You designed your XRD
  • You've added provider dependencies
  • You have a Docker-compatible container runtime running. up builds Python functions inside a container
  • You have Python 3.11, 3.12, or 3.13 installed for IDE support
  • You have the Python VS Code extension installed
  • You understand your XRD schema and what resources you need to create

Create your composition scaffold​

Use the XRD you created in the previous step to generate a new composition:

up composition generate apis/<your_resource_name>/definition.yaml

This command creates apis/<your_resource_name>/composition.yaml which references the XRD.

Generate your function​

Use your chosen programming language to generate a new function:

up function generate --language=python compose-resources apis/<your_resource_name>/composition.yaml

This command creates a functions/compose-resources directory with your function code and updates your composition file to reference it.

The function is a Python project built on the Crossplane Python function SDK:

functions/compose-resources/
├── pyproject.toml      # Project metadata and dependencies
├── README.md
└── function/
    ├── __init__.py
    ├── __version__.py
    ├── fn.py           # Your composition logic
    └── main.py         # Entrypoint that serves the function

Write your composition logic in function/fn.py. You don't need to edit main.py. The generated function/fn.py should be similar to:

functions/compose-resources/function/fn.py
"""A Crossplane composition function."""

import grpc
from crossplane.function import logging, response
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1


class FunctionRunner(grpcv1.FunctionRunnerService):
    """A FunctionRunner handles gRPC RunFunctionRequests."""

    def __init__(self):
        """Create a new FunctionRunner."""
        self.log = logging.get_logger()

    async def RunFunction(
        self, req: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext
    ) -> fnv1.RunFunctionResponse:
        """Run the function."""
        log = self.log.bind(tag=req.meta.tag)
        log.info("Running function")

        rsp = response.to(req)

        # Add your composition logic here. For example, to compose desired
        # resources, populate rsp.desired.resources using
        # crossplane.function.resource.

        return rsp

Crossplane calls RunFunction each time it reconciles your composite resource. response.to(req) creates a response that already contains the desired state and context from earlier steps in the pipeline. Update rsp and return it.

The shorter examples on this page omit the generated __init__ method and logging. Each example is still a complete function/fn.py.

Add dependencies​

Your function's dependencies are in the dependencies list in its pyproject.toml:

functions/compose-resources/pyproject.toml
dependencies = [
  "crossplane-function-sdk-python==0.11.0",
  "click==8.3.2",
  "grpcio>=1.73.1",
  "crossplane-models @ file:./../../.up/python",
]

To use another package from PyPI, add it to this list. up project build installs it when it builds your function.

up installs dependencies from wheels only, so it can build your function for more than one architecture. Each package must publish a pure-Python wheel or a manylinux2014 wheel for every architecture you build. Packages that only publish a source distribution fail to install.

The crossplane-models entry installs the models up generates for your project. up function generate only adds this entry when your project already has generated Python models. If it's missing, run up project build, then add the entry with the relative path from your function directory to .up/python.

Models​

Upbound Official Providers and some other packages include Pydantic models for their resources. These models enable in-line documentation, linting, autocompletion, and other features when working with Crossplane resources in embedded Python functions.

up generates models for your project's dependencies and XRDs in the .up/python directory. It packages them as a Python distribution named crossplane-models, which provides a top-level models package. Your function depends on crossplane-models in its pyproject.toml, so import models from models:

from models.com.example.platform.storagebucket import v1alpha1
from models.io.upbound.m.aws.s3.bucket import v1beta1 as bucketv1beta1

Each import path is the resource's API group in reverse, followed by its lowercase kind and API version. For example, the Bucket kind in the s3.aws.m.upbound.io/v1beta1 API version is in models.io.upbound.m.aws.s3.bucket.v1beta1.

To add new models, add the package dependency with the up CLI:

up dependency add xpkg.upbound.io/upbound/provider-aws-s3

When you change an XRD, rebuild your project to regenerate its models:

up project build  # Generate models in .up/python directory
tip

The generated model module and class name match your XRD's kind exactly. The examples on this page assume a v2 XRD (apiextensions.crossplane.io/v2) with kind: StorageBucket and optional spec.parameters.region, spec.parameters.acl, and spec.parameters.versioning fields. If your XRD uses the legacy v1 API (apiextensions.crossplane.io/v1), your XRD's kind uses an X prefix, for example kind: XStorageBucket, so import from models...xstoragebucket and use v1alpha1.XStorageBucket instead.

Version 2 Upbound Official Providers include namespaced resources, like s3.aws.m.upbound.io, and cluster-scoped resources, like s3.aws.upbound.io. A v2 XRD creates namespaced composite resources by default, and namespaced composite resources can only compose namespaced resources. The examples on this page use the namespaced models in models.io.upbound.m.

The imports in this example are specifically for AWS S3 buckets. They follow a similar structure for all resources:

  • fnv1 - Provides protocol buffer types for function communication
  • grpcv1 - Provides the FunctionRunnerService base class for your function
  • v1alpha1 - References your XRD's generated Pydantic model
  • bucketv1beta1 - The AWS S3 provider's Pydantic model

Optional and required fields​

Upbound's Python models know which resource fields Crossplane requires and which are optional.

Required fields have a specific type, like str - a string.

Python raises an exception if you create a model without supplying a required field. This can be a problem when updating the desired composite resource (XR).

You should only include the fields your function has an opinion about when you update the desired XR. This can be a problem if for example Crossplane requires an XR spec field, but your function only wants to update a status field.

When updating the desired XR, you can avoid issues due to required fields by using the resource's status model directly. This example assumes your XRD defines a status.replicas field:

functions/compose-resources/function/fn.py
import grpc
from crossplane.function import resource, response
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1

from models.com.example.platform.storagebucket import v1alpha1


class FunctionRunner(grpcv1.FunctionRunnerService):
    async def RunFunction(
        self, req: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext
    ) -> fnv1.RunFunctionResponse:
        rsp = response.to(req)

        # Include any desired status from previous functions in the pipeline.
        desired_xr = resource.struct_to_dict(req.desired.composite.resource)
        desired_xr_status = v1alpha1.Status(**desired_xr.get("status", {}))

        # Update only the status field your function is concerned with.
        desired_xr_status.replicas = 3

        # Dump the model as a Python dictionary.
        resource.update(
            rsp.desired.composite,
            {"status": desired_xr_status.model_dump(exclude_none=True)},
        )

        return rsp

Optional fields have a union type with None, like str | None. This means the field can be a string, or None - Python's null value.

Your editor's type checker warns you when you copy an optional field to a required field.

For example, the type checker warns you if you try to copy an optional spec.parameters.region field from an XR to a required spec.forProvider.region field of an MR:

functions/compose-resources/function/fn.py
import grpc
from crossplane.function import resource, response
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1

from models.com.example.platform.storagebucket import v1alpha1
from models.io.upbound.m.aws.s3.bucket import v1beta1 as bucketv1beta1


class FunctionRunner(grpcv1.FunctionRunnerService):
    async def RunFunction(
        self, req: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext
    ) -> fnv1.RunFunctionResponse:
        rsp = response.to(req)

        observed_xr = v1alpha1.StorageBucket(**resource.struct_to_dict(req.observed.composite.resource))
        params = observed_xr.spec.parameters or v1alpha1.Parameters()

        desired_bucket = bucketv1beta1.Bucket(
            spec=bucketv1beta1.Spec(
                forProvider=bucketv1beta1.ForProvider(
                    region=params.region,  # Warning: Argument of type "str | None" cannot be assigned to parameter "region" of type "str"
                ),
            ),
        )
        resource.update(rsp.desired.resources["bucket"], desired_bucket)

        return rsp

spec.parameters is also optional, so this example falls back to an empty v1alpha1.Parameters() model when the XR doesn't set it.

You can address this warning two ways.

If the optional field could be None in practice, handle that case by specifying a default value.

functions/compose-resources/function/fn.py
import grpc
from crossplane.function import resource, response
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1

from models.com.example.platform.storagebucket import v1alpha1
from models.io.upbound.m.aws.s3.bucket import v1beta1 as bucketv1beta1


class FunctionRunner(grpcv1.FunctionRunnerService):
    async def RunFunction(
        self, req: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext
    ) -> fnv1.RunFunctionResponse:
        rsp = response.to(req)

        observed_xr = v1alpha1.StorageBucket(**resource.struct_to_dict(req.observed.composite.resource))
        params = observed_xr.spec.parameters or v1alpha1.Parameters()

        desired_bucket = bucketv1beta1.Bucket(
            spec=bucketv1beta1.Spec(
                forProvider=bucketv1beta1.ForProvider(
                    region=params.region or "us-west-2",  # Default to "us-west-2" if region is None.
                ),
            ),
        )
        resource.update(rsp.desired.resources["bucket"], desired_bucket)

        return rsp

If the optional field can't be None in practice, use a type: ignore comment to silence the warning.

functions/compose-resources/function/fn.py
import grpc
from crossplane.function import resource, response
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1

from models.com.example.platform.storagebucket import v1alpha1
from models.io.k8s.apimachinery.pkg.apis.meta import v1 as metav1
from models.io.upbound.m.aws.s3.bucket import v1beta1 as bucketv1beta1


class FunctionRunner(grpcv1.FunctionRunnerService):
    async def RunFunction(
        self, req: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext
    ) -> fnv1.RunFunctionResponse:
        rsp = response.to(req)

        observed_xr = v1alpha1.StorageBucket(**resource.struct_to_dict(req.observed.composite.resource))

        desired_bucket = bucketv1beta1.Bucket(
            metadata=metav1.ObjectMeta(
                name=observed_xr.metadata.name + "-bucket",  # type: ignore  # The observed XR always has a name.
            ),
            spec=bucketv1beta1.Spec(
                forProvider=bucketv1beta1.ForProvider(
                    region="us-west-2",
                ),
            ),
        )
        resource.update(rsp.desired.resources["bucket"], desired_bucket)

        return rsp

Create your function logic​

Next, add the function logic. The example below creates an S3 bucket:

functions/compose-resources/function/fn.py
import grpc
from crossplane.function import logging, resource, response
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1

from models.com.example.platform.storagebucket import v1alpha1
from models.io.upbound.m.aws.s3.bucket import v1beta1 as bucketv1beta1


class FunctionRunner(grpcv1.FunctionRunnerService):
    """A FunctionRunner handles gRPC RunFunctionRequests."""

    def __init__(self):
        """Create a new FunctionRunner."""
        self.log = logging.get_logger()

    async def RunFunction(
        self, req: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext
    ) -> fnv1.RunFunctionResponse:
        """Run the function."""
        log = self.log.bind(tag=req.meta.tag)
        log.info("Running function")

        # Start from the desired state and context of earlier pipeline steps.
        rsp = response.to(req)

        # Load the observed XR into a Pydantic model.
        observed_xr = v1alpha1.StorageBucket(**resource.struct_to_dict(req.observed.composite.resource))
        params = observed_xr.spec.parameters or v1alpha1.Parameters()

        # Create the cloud resource specification.
        desired_bucket = bucketv1beta1.Bucket(
            spec=bucketv1beta1.Spec(
                forProvider=bucketv1beta1.ForProvider(
                    region=params.region or "us-west-2",
                ),
            ),
        )
        resource.update(rsp.desired.resources["bucket"], desired_bucket)

        return rsp

Inputs​

Function logic determines how Crossplane handles your resource creation. In the RunFunctionRequest, there are four inputs that Crossplane can parse:

  1. Observed state: What real resources currently exist?

    # The API request
    observed_xr = v1alpha1.StorageBucket(**resource.struct_to_dict(req.observed.composite.resource))

  2. Desired state: What resources should exist?

    # Create the cloud resource specification
    desired_bucket = bucketv1beta1.Bucket(
        spec=bucketv1beta1.Spec(
            forProvider=bucketv1beta1.ForProvider(
                region=params.region or "us-west-2",
            ),
        ),
    )

  3. Function logic - What does Crossplane do?

    # Reconcile the desired and observed states
    resource.update(rsp.desired.resources["bucket"], desired_bucket)
  1. Pipeline context - Information to pass to subsequent functions in the pipeline.

For a more complex version of a Python function, expand the example below:

A more advanced Python function

The function/fn.py file below takes a composite resource (XR) as input and produces managed resources (MRs) from the S3 provider based on its parameters.

The function always composes an S3 bucket. When the S3 bucket exists, it also composes a bucket access control list (ACL), ownership controls, a public access block, and a server-side encryption configuration. Each of these resources references the bucket by name.

If the composite resource's spec.parameters.versioning field is True, the function enables versioning by composing a bucket versioning configuration. Like the ACL, the versioning configuration references the bucket by name.

functions/compose-resources/function/fn.py
import grpc
from crossplane.function import logging, resource, response
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1

from models.com.example.platform.storagebucket import v1alpha1
from models.io.upbound.m.aws.s3.bucket import v1beta1 as bucketv1beta1
from models.io.upbound.m.aws.s3.bucketacl import v1beta1 as aclv1beta1
from models.io.upbound.m.aws.s3.bucketownershipcontrols import v1beta1 as bocv1beta1
from models.io.upbound.m.aws.s3.bucketpublicaccessblock import v1beta1 as pabv1beta1
from models.io.upbound.m.aws.s3.bucketserversideencryptionconfiguration import (
    v1beta1 as ssev1beta1,
)
from models.io.upbound.m.aws.s3.bucketversioning import v1beta1 as verv1beta1


class FunctionRunner(grpcv1.FunctionRunnerService):
    """A FunctionRunner handles gRPC RunFunctionRequests."""

    def __init__(self):
        """Create a new FunctionRunner."""
        self.log = logging.get_logger()

    async def RunFunction(
        self, req: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext
    ) -> fnv1.RunFunctionResponse:
        """Run the function."""
        log = self.log.bind(tag=req.meta.tag)
        rsp = response.to(req)

        observed_xr = v1alpha1.StorageBucket(**resource.struct_to_dict(req.observed.composite.resource))
        params = observed_xr.spec.parameters or v1alpha1.Parameters()

        if params.region is None:
            response.fatal(rsp, "spec.parameters.region is required")
            return rsp

        desired_bucket = bucketv1beta1.Bucket(
            spec=bucketv1beta1.Spec(
                forProvider=bucketv1beta1.ForProvider(
                    region=params.region,
                ),
            ),
        )
        resource.update(rsp.desired.resources["bucket"], desired_bucket)

        # The desired ACL, encryption, and versioning resources all need to refer to
        # the bucket by its external name, which is stored in its external name
        # annotation. Return early if the Bucket's external-name annotation isn't
        # set yet.
        if "bucket" not in req.observed.resources:
            log.info("Waiting for the bucket to be created")
            return rsp

        observed_bucket = bucketv1beta1.Bucket(
            **resource.struct_to_dict(req.observed.resources["bucket"].resource)
        )
        if observed_bucket.metadata is None or observed_bucket.metadata.annotations is None:
            return rsp
        if "crossplane.io/external-name" not in observed_bucket.metadata.annotations:
            return rsp

        bucket_external_name = observed_bucket.metadata.annotations[
            "crossplane.io/external-name"
        ]

        desired_acl = aclv1beta1.BucketACL(
            spec=aclv1beta1.Spec(
                forProvider=aclv1beta1.ForProvider(
                    region=params.region,
                    bucket=bucket_external_name,
                    acl=params.acl,
                ),
            ),
        )
        resource.update(rsp.desired.resources["acl"], desired_acl)

        desired_boc = bocv1beta1.BucketOwnershipControls(
            spec=bocv1beta1.Spec(
                forProvider=bocv1beta1.ForProvider(
                    region=params.region,
                    bucket=bucket_external_name,
                    rule=bocv1beta1.Rule(
                        objectOwnership="BucketOwnerPreferred",
                    ),
                ),
            ),
        )
        resource.update(rsp.desired.resources["boc"], desired_boc)

        desired_pab = pabv1beta1.BucketPublicAccessBlock(
            spec=pabv1beta1.Spec(
                forProvider=pabv1beta1.ForProvider(
                    region=params.region,
                    bucket=bucket_external_name,
                    blockPublicAcls=False,
                    ignorePublicAcls=False,
                    restrictPublicBuckets=False,
                    blockPublicPolicy=False,
                ),
            ),
        )
        resource.update(rsp.desired.resources["pab"], desired_pab)

        desired_sse = ssev1beta1.BucketServerSideEncryptionConfiguration(
            spec=ssev1beta1.Spec(
                forProvider=ssev1beta1.ForProvider(
                    region=params.region,
                    bucket=bucket_external_name,
                    rule=[
                        ssev1beta1.RuleItem(
                            applyServerSideEncryptionByDefault=ssev1beta1.ApplyServerSideEncryptionByDefault(
                                sseAlgorithm="AES256",
                            ),
                            bucketKeyEnabled=True,
                        ),
                    ],
                ),
            ),
        )
        resource.update(rsp.desired.resources["sse"], desired_sse)

        # Return early without composing a BucketVersioning MR if the XR doesn't
        # have versioning enabled.
        if not params.versioning:
            return rsp

        desired_versioning = verv1beta1.BucketVersioning(
            spec=verv1beta1.Spec(
                forProvider=verv1beta1.ForProvider(
                    region=params.region,
                    bucket=bucket_external_name,
                    versioningConfiguration=verv1beta1.VersioningConfiguration(
                        status="Enabled",
                    ),
                ),
            ),
        )
        resource.update(rsp.desired.resources["versioning"], desired_versioning)

        return rsp

Outputs​

RunFunctionResponse returns three outputs which update the state of the control plane:

  1. Desired state of your composed resources
  2. Status conditions to apply to the composite resource or claim
  3. Context to pass to other functions in the pipeline

response.to(req) creates a RunFunctionResponse that's pre-populated with the request's desired state and context. Python functions only need to update the fields in the objects that need to change, then return the response.

tip

You can select the RunFunctionResponse object in Visual Studio Code to see what fields it has.

The Python function SDK generates the RunFunctionResponse object from a protobuf definition. Read the Python Generated Code Guide to learn about protobuf generated code.

You can add or update composed resources using the resource.update helper function in the Crossplane Python SDK:

composed = ...  # Construct a composed resource
resource.update(rsp.desired.resources["my-resource"], composed)

Similarly, you can update the status of the composite resource by updating it in the response:

status_update = {
    "status": {
        "someInformation": "cool-status"
    }
}
resource.update(rsp.desired.composite, status_update)
tip

If you don't want to use a model, you can also pass resource.update a Python dictionary.

functions/compose-resources/function/fn.py
import grpc
from crossplane.function import resource, response
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1


class FunctionRunner(grpcv1.FunctionRunnerService):
    async def RunFunction(
        self, req: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext
    ) -> fnv1.RunFunctionResponse:
        rsp = response.to(req)

        resource.update(rsp.desired.composite, {
            "status": {
                "replicas": 3,
            },
        })

        return rsp

Migrate from the single-file format​

Before up v0.50.0, up function generate --language=python created a single main.py file with a compose(req, rsp) function, a requirements.txt file, and a model link to the generated models.

up project build still builds functions in the single-file format. up chooses how to build each function from the files in its directory:

  • A pyproject.toml file and a function/ directory use the format this guide describes.
  • A main.py file uses the single-file format.

Migrate a function to add dependencies from PyPI. To migrate a function, for example compose-resources:

  1. Move the existing function out of the functions directory:

    mv functions/compose-resources compose-resources-old

  2. Generate a new function with the same name. Leave out the composition path, because your composition already references the function:

    up function generate --language=python compose-resources

  3. Move your logic from compose() in compose-resources-old/main.py into RunFunction in functions/compose-resources/function/fn.py:

    • Keep rsp = response.to(req) at the start of RunFunction and return rsp at the end.
    • Change any other return statement to return rsp.
    • Change model imports from from .model. to from models..

  4. Add any packages from compose-resources-old/requirements.txt to the dependencies list in functions/compose-resources/pyproject.toml. The function SDK already depends on Pydantic.

  5. Build your project, then delete the old function:

    up project build
    rm -r compose-resources-old