### Combined Options Example

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/cli.md

Starts the server with a combination of debug mode, insecure connection, a custom address, and specific message size limits.

```bash
python -m function.main \
    --debug \
    --insecure \
    --address "localhost:9443" \
    --max-recv-message-size 8 \
    --max-send-message-size 8
```

--------------------------------

### Install function-python Locally

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/quickstart.md

Clone the repository, navigate to the directory, and install the package in editable mode. This is typically done for local development.

```bash
git clone https://github.com/crossplane-contrib/function-python.git
cd function-python
pip install -e .
```

--------------------------------

### Install Dependencies

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/OVERVIEW.md

Installs the Python Function and its dependencies in editable mode.

```bash
pip install -e .
```

--------------------------------

### Test Function with Example Manifests

Source: https://github.com/crossplane-contrib/function-python/blob/main/example/composition/README.md

Call the locally running Python function with example manifests for an XR, composition, and the function itself. The `-r` flag indicates a render operation.

```shell
# Then, in another terminal, call it with these example manifests
$ crossplane render xr.yaml composition.yaml functions.yaml -r
```

--------------------------------

### RunFunction Example

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/FunctionRunner.md

Demonstrates how to use the FunctionRunner to execute a Python script that defines a compose function. This example prepares a request with a script, runs it, and checks for success or errors.

```python
import grpc
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function import resource
from function.fn import FunctionRunner

# Prepare a composition script
script = """
from crossplane.function.proto.v1 import run_function_pb2 as fnv1

def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
    rsp.desired.resources["my-resource"].resource.update({
        "apiVersion": "example.com/v1",
        "kind": "Example",
        "spec": {
            "name": "my-example"
        }
    })
"""

# Create request with script
req = fnv1.RunFunctionRequest(
    input=resource.dict_to_struct({"script": script})
)

# Execute
runner = FunctionRunner()
response = await runner.RunFunction(req, None)

# Check for errors
if response.results:
    for result in response.results:
        print(f"Message: {result.message}")
else:
    print("Success")
```

--------------------------------

### RunFunctionRequest Body Example (JSON)

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/endpoints.md

An example JSON payload for a RunFunctionRequest, demonstrating the structure for input script, observed resources, and desired state.

```json
{
  "meta": {
    "tag": "unique-request-id"
  },
  "input": {
    "script": "from crossplane.function.proto.v1 import run_function_pb2 as fnv1\n\ndef compose(req, rsp):\n    rsp.desired.resources[\"bucket\"].resource.update({...})"
  },
  "observed": {
    "composite": {
      "resource": {
        "apiVersion": "example.crossplane.io/v1",
        "kind": "XR",
        "metadata": {
          "name": "example",
          "namespace": "default"
        },
        "spec": {
          "region": "us-east-1"
        }
      }
    },
    "resources": {}
  },
  "desired": {
    "resources": {},
    "context": {}
  },
  "context": {}
}
```

--------------------------------

### Run Function Request Example (Composition)

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/endpoints.md

This example demonstrates how to send a request to the FunctionRunnerService to compose a new AWS S3 bucket resource. It uses grpcurl to send a JSON payload containing the desired script and observed composite resource.

```bash
grpcurl -d @ \
  -H "Content-Type: application/json" \
  localhost:9443 \
  crossplane.function.proto.v1.FunctionRunnerService/RunFunction <<'EOF'\
{
  "meta": {"tag": "test-request-123"},
  "input": {
    "script": "from crossplane.function.proto.v1 import run_function_pb2 as fnv1\n\ndef compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):\n    rsp.desired.resources[\"bucket\"].resource.update({\"apiVersion\": \"s3.aws.upbound.io/v1beta2\", \"kind\": \"Bucket\"})"
  },
  "observed": {
    "composite": {
      "resource": {
        "apiVersion": "example.crossplane.io/v1",
        "kind": "XR",
        "spec": {"region": "us-east-1"}
      }
    }
  }
}
EOF
```

--------------------------------

### Simple Composition Request

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/endpoints.md

A minimal request example for simple composition. The script defines a 'compose' function that specifies a desired resource of kind 'Example'.

```json
{
  "meta": {"tag": "compose-1"},
  "input": {"script": "def compose(req, rsp):\n    rsp.desired.resources[\"res\"].resource.update({\"kind\": \"Example\"})"},
  "observed": {"composite": {"resource": {"spec": {}}}}
}
```

--------------------------------

### RunFunctionResponse Body Example (JSON)

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/endpoints.md

An example JSON payload for a RunFunctionResponse, showing successful execution results, composed resources, and output status.

```json
{
  "meta": {
    "ttl": {
      "seconds": 60
    }
  },
  "results": [
    {
      "severity": "SEVERITY_NORMAL",
      "message": "Function executed successfully"
    }
  ],
  "desired": {
    "resources": {
      "bucket": {
        "resource": {
          "apiVersion": "s3.aws.upbound.io/v1beta2",
          "kind": "Bucket",
          "metadata": {
            "name": "example-bucket"
          },
          "spec": {
            "forProvider": {
              "region": "us-east-1",
              "acl": "private"
            }
          }
        },
        "ready": true
      }
    },
    "context": {}
  },
  "observed": {
    "resources": {},
    "context": {}
  },
  "context": {},
  "output": {
    "status": "completed"
  }
}
```

--------------------------------

### Run Function Response Example

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/endpoints.md

This is an example of a successful response from the FunctionRunnerService after a composition request. It indicates the desired resources to be created or updated, including metadata like TTL for caching.

```json
{
  "meta": {
    "ttl": {"seconds": 60}
  },
  "results": [],
  "desired": {
    "resources": {
      "bucket": {
        "resource": {
          "apiVersion": "s3.aws.upbound.io/v1beta2",
          "kind": "Bucket"
        }
      }
    }
  }
}
```

--------------------------------

### Run Development Server

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/OVERVIEW.md

Starts the development server for the Python Function. This command is used for local development and testing.

```bash
hatch run development
# Runs: python function/main.py --insecure --debug
```

--------------------------------

### Run Production Deployment

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/quickstart.md

Start the function-python server for production deployment, specifying the address and TLS certificates directory. Ensure the TLS certificates are correctly configured.

```bash
python -m function.main \
    --address 0.0.0.0:9443 \
    --tls-certs-dir /etc/crossplane/certs
```

--------------------------------

### Test Function Locally

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/quickstart.md

Commands to start the Python function server and test a composition locally using `crossplane render`.

```bash
# Terminal 1: Start the function server
python -m function.main --insecure --debug

# Terminal 2: Test the composition
crossplane render example/composition/xr.yaml \
                 example/composition/composition.yaml \
                 example/composition/functions.yaml -r
```

--------------------------------

### Complete Composition Example

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/SDK_Integration.md

Demonstrates how to compose a new AWS S3 bucket resource based on input from a composite resource. It extracts region and ACL specifications, validates input, and defines the desired bucket resource.

```python
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function import resource, response, logging

def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
    log = logging.get_logger()
    
    try:
        # Extract input
        log.info("Extracting composite resource data")
        composite = req.observed.composite.resource
        
        if not composite or "spec" not in composite:
            response.fatal(rsp, "Invalid composite resource")
            return
        
        spec = composite["spec"]
        region = spec.get("region", "us-east-1")
        name = composite["metadata"]["name"]
        
        # Validate
        if not region:
            response.warning(rsp, "Region not specified, using default")
        
        # Compose resources
        bucket_def = {
            "apiVersion": "s3.aws.upbound.io/v1beta2",
            "kind": "Bucket",
            "metadata": {"name": name},
            "spec": {
                "forProvider": {
                    "region": region,
                    "acl": spec.get("acl", "private"),
                    "versioning": {
                        "enabled": spec.get("versioning", False)
                    }
                }
            }
        }
        
        rsp.desired.resources["bucket"].resource.update(bucket_def)
        rsp.desired.resources["bucket"].ready = True
        
        response.normal(rsp, f"Composed bucket {name} in region {region}")
        
    except Exception as e:
        log.error(f"Composition failed: {e}")
        response.fatal(rsp, f"Composition error: {str(e)}")
```

--------------------------------

### Complete Operation Example

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/SDK_Integration.md

Demonstrates how to perform an operation by checking the SSL certificate expiry for a given hostname. It extracts the hostname, retrieves the certificate, calculates days until expiry, and updates the response status.

```python
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function import request, response, logging
from datetime import datetime
import ssl
import socket

def operate(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
    log = logging.get_logger()
    
    try:
        # Get required resource
        ingress = request.get_required_resource(req, "ingress")
        
        if not ingress:
            response.fatal(rsp, "Required Ingress resource not found")
            return
        
        # Extract hostname
        hostname = ingress["spec"]["rules"][0]["host"]
        log.info(f"Checking certificate for {hostname}")
        
        # Get SSL certificate
        context = ssl.create_default_context()
        with socket.create_connection((hostname, 443)) as sock:
            with context.wrap_socket(sock, server_hostname=hostname) as ssock:
                cert = ssock.getpeercert()
        
        # Calculate expiry
        expiry_str = cert["notAfter"]
        expiry_date = datetime.strptime(expiry_str, "%b %d %H:%M:%S %Y %Z")
        days_until_expiry = (expiry_date - datetime.now()).days
        
        # Set output
        response.set_output(rsp, {
            "hostname": hostname,
            "certificateExpires": expiry_str,
            "daysUntilExpiry": days_until_expiry,
            "status": "warning" if days_until_expiry < 30 else "ok"
        })
        
        # Update resource if needed
        if days_until_expiry < 30:
            response.warning(rsp, f"Certificate expires in {days_until_expiry} days")
        
    except Exception as e:
        log.error(f"Operation failed: {e}")
        response.fatal(rsp, f"Operation error: {str(e)}")
```

--------------------------------

### Async Composition Function Example

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/OVERVIEW.md

An asynchronous Python function for composition. It demonstrates the use of await for non-blocking operations before defining desired resources.

```python
async def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
    # Can use async operations
    await fetch_data()
    
    rsp.desired.resources["data"].resource.update({...})
```

--------------------------------

### Custom Address and Message Size Configuration

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/cli.md

Starts the server on a custom address and port, and configures the maximum send and receive message sizes for gRPC communication.

```bash
python -m function.main \
    --address "127.0.0.1:5000" \
    --max-recv-message-size 16 \
    --max-send-message-size 16
```

--------------------------------

### Python Function Script Input Example

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/OVERVIEW.md

Example of how to define a Python script for the Function CRD. This snippet shows the basic structure for a script that receives a RunFunctionRequest and returns a RunFunctionResponse.

```yaml
input:
  apiVersion: python.fn.crossplane.io/v1beta1
  kind: Script
  script: |
    from crossplane.function.proto.v1 import run_function_pb2 as fnv1
    
    def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
        # Script code here
        pass
```

--------------------------------

### Simple Composition Response

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/endpoints.md

The corresponding response for the simple composition request. It indicates the desired resource 'res' of kind 'Example' to be managed.

```json
{
  "meta": {"ttl": {"seconds": 60}},
  "desired": {"resources": {"res": {"resource": {"kind": "Example"}}}}
}
```

--------------------------------

### Error Handling Example

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/cli.md

Illustrates the format of error messages displayed to the user when exceptions occur during function execution. Common errors include certificate loading failures, network binding issues, and invalid certificate formats.

```bash
Cannot run function: {exception}
```

--------------------------------

### Composition Function Example

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/OVERVIEW.md

A Python function for composition. It accesses the composite resource spec to determine desired resources and sets their state.

```python
from crossplane.function.proto.v1 import run_function_pb2 as fnv1

def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
    # Access composite resource spec
    region = req.observed.composite.resource["spec"]["region"]
    
    # Add desired resources
    rsp.desired.resources["my-resource"].resource.update({
        "apiVersion": "example.com/v1",
        "kind": "Example",
        "spec": {"region": region}
    })
    rsp.desired.resources["my-resource"].ready = True
```

--------------------------------

### Basic Composition Function

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/FunctionRunner.md

A basic example of a `compose` function that accesses request data to configure a desired resource. It updates the `spec` of a resource named 'bucket'.

```python
from crossplane.function.proto.v1 import run_function_pb2 as fnv1

def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
    # Access request data
    region = req.observed.composite.resource["spec"]["region"]
    
    # Add or update desired resources
    rsp.desired.resources["bucket"].resource.update({
        "apiVersion": "s3.aws.upbound.io/v1beta2",
        "kind": "Bucket",
        "spec": {
            "forProvider": {
                "region": region
            }
        },
    })
    rsp.desired.resources["bucket"].ready = True
```

--------------------------------

### Compose Multiple Resources with Python Function

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/quickstart.md

This example shows a Python function composing multiple related resources: an S3 bucket, an IAM role, and a bucket policy. It demonstrates how to define and update each resource within the function.

```yaml
script: |
  from crossplane.function.proto.v1 import run_function_pb2 as fnv1

  def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
      name = req.observed.composite.resource["metadata"]["name"]
      
      # Create S3 bucket
      rsp.desired.resources["bucket"].resource.update({
          "apiVersion": "s3.aws.upbound.io/v1beta2",
          "kind": "Bucket",
          "metadata": {"name": name},
          "spec": {"forProvider": {"region": "us-east-1"}}
      })
      rsp.desired.resources["bucket"].ready = True
      
      # Create IAM role for the bucket
      rsp.desired.resources["role"].resource.update({
          "apiVersion": "iam.aws.upbound.io/v1beta1",
          "kind": "Role",
          "metadata": {"name": f"{name}-role"},
          "spec": {
              "forProvider": {
                  "assumeRolePolicy": '{"Version":"2012-10-17"...}'
              }
          }
      })
      rsp.desired.resources["role"].ready = True
      
      # Create bucket policy
      rsp.desired.resources["policy"].resource.update({
          "apiVersion": "s3.aws.upbound.io/v1beta2",
          "kind": "BucketPolicy",
          "metadata": {"name": f"{name}-policy"},
          "spec": {
              "forProvider": {
                  "bucket": name,
                  "policy": '{"Version":"2012-10-17"...}'
              }
          }
      })
      rsp.desired.resources["policy"].ready = True
```

--------------------------------

### Example Imports for Crossplane Python Functions

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/configuration.md

Demonstrates common imports for Python scripts used in Crossplane Functions, including standard library, Crossplane SDK, and third-party packages like `grpcio` and `aioboto3`.

```python
# Standard library
import json
import time
from datetime import datetime
import asyncio

# Crossplane SDK
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function import resource, logging, response

# Third-party
import boto3
from aioboto3 import Session
```

--------------------------------

### Compose Simple S3 Bucket with Python Function

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/quickstart.md

This example demonstrates how to use a Python function within a Crossplane Composition to create a simple S3 bucket. It extracts parameters like region and name from the composite resource.

```yaml
apiVersion: apiextensions.crossplane.io/v1
kind: Composition
metadata:
  name: compose-bucket-with-python
spec:
  compositeTypeRef:
    apiVersion: example.crossplane.io/v1
    kind: XR
  mode: Pipeline
  pipeline:
  - step: compose-with-python
    functionRef:
      name: function-python
    input:
      apiVersion: python.fn.crossplane.io/v1beta1
      kind: Script
      script: |
        from crossplane.function.proto.v1 import run_function_pb2 as fnv1

        def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
            # Get parameters from the composite resource
            region = req.observed.composite.resource["spec"]["region"]
            name = req.observed.composite.resource["metadata"]["name"]
            
            # Compose an S3 bucket
            rsp.desired.resources["bucket"].resource.update({
                "apiVersion": "s3.aws.upbound.io/v1beta2",
                "kind": "Bucket",
                "metadata": {
                    "name": name
                },
                "spec": {
                    "forProvider": {
                        "region": region,
                        "acl": "private",
                        "versioning": {
                            "enabled": True
                        }
                    }
                }
            })
            rsp.desired.resources["bucket"].ready = True
```

--------------------------------

### Operation Function Example

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/OVERVIEW.md

A Python function for performing operational tasks. It executes a task, captures the result, and sets output for monitoring.

```python
def operate(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
    # Perform operational task
    result = perform_task()
    
    # Set output for monitoring
    rsp.output["status"] = "completed"
    rsp.output["result"] = result
```

--------------------------------

### Operation Function

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/FunctionRunner.md

An example of an `operate` function that performs an operational task and sets output for monitoring. It calls a placeholder function `perform_some_task` and updates `rsp.output`.

```python
from crossplane.function.proto.v1 import run_function_pb2 as fnv1

def operate(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
    # Perform operational task
    result = perform_some_task()
    
    # Set operation output for monitoring
    rsp.output["status"] = "completed"
    rsp.output["result"] = result
```

--------------------------------

### Composition Pipeline Configuration

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/configuration.md

Configures the Python function as a pipeline step within a Crossplane Composition. This example demonstrates accessing request data to dynamically configure a resource.

```yaml
apiVersion: apiextensions.crossplane.io/v1
kind: Composition
metadata:
  name: example-composition
spec:
  compositeTypeRef:
    apiVersion: example.crossplane.io/v1
    kind: XR
  mode: Pipeline
  pipeline:
  - step: compose-with-python
    functionRef:
      name: function-python
    input:
      apiVersion: python.fn.crossplane.io/v1beta1
      kind: Script
      script: |
        from crossplane.function.proto.v1 import run_function_pb2 as fnv1
        
        def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
            # Access request data
            region = req.observed.composite.resource["spec"]["region"]
            
            # Compose resources
            rsp.desired.resources["bucket"].resource.update({
                "apiVersion": "s3.aws.upbound.io/v1beta2",
                "kind": "Bucket",
                "spec": {
                    "forProvider": {"region": region}
                }
            })
```

--------------------------------

### Test Python Function with Crossplane

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/INDEX.md

Starts the Python function server in one terminal and renders a composition in another to test function integration. Ensure the server is running before rendering.

```bash
# Terminal 1: Start server
python -m function.main --insecure --debug

# Terminal 2: Render composition
crossplane render example/composition/xr.yaml \
                 example/composition/composition.yaml \
                 example/composition/functions.yaml -r
```

--------------------------------

### Development Mode Usage (Insecure)

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/cli.md

Starts the Crossplane Python function server in development mode with debug logging and no TLS encryption. The server listens on the default address `0.0.0.0:9443`.

```bash
python -m function.main --insecure --debug
```

--------------------------------

### Compose a Resource with Python Script

Source: https://github.com/crossplane-contrib/function-python/blob/main/README.md

Example of a Crossplane Composition that uses a Python script to define and add a composed S3 bucket resource. The script accesses request details to configure the bucket.

```yaml
apiVersion: apiextensions.crossplane.io/v1
kind: Composition
metadata:
  name: compose-a-resource-with-python
spec:
  compositeTypeRef:
    apiVersion: example.crossplane.io/v1
    kind: XR
  mode: Pipeline
  pipeline:
  - step: compose-a-resource-with-python
    functionRef:
      name: function-python
    input:
      apiVersion: python.fn.crossplane.io/v1beta1
      kind: Script
      script: |
        from crossplane.function.proto.v1 import run_function_pb2 as fnv1

        def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
            rsp.desired.resources["bucket"].resource.update({
                "apiVersion": "s3.aws.upbound.io/v1beta2",
                "kind": "Bucket",
                "spec": {
                    "forProvider": {
                        "region": req.observed.composite.resource["spec"]["region"]
                    }
                },
            })
            rsp.desired.resources["bucket"].ready = True
```

--------------------------------

### Production Mode Usage (mTLS)

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/cli.md

Starts the Crossplane Python function server in production mode, enabling mTLS encryption. It requires specifying the address and the directory containing TLS certificates.

```bash
python -m function.main \
    --address "0.0.0.0:9443" \
    --tls-certs-dir "/etc/crossplane/certs"
```

--------------------------------

### Running Server via Package Entry Point

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/cli.md

Demonstrates how to run the Crossplane Python function server using the defined package entry point.

```bash
function --insecure --debug
```

--------------------------------

### Script CRD Example: Composition

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/configuration.md

Example of the Script CRD used for composition. It defines a Python script that implements the `compose` function to generate desired resources.

```yaml
apiVersion: python.fn.crossplane.io/v1beta1
kind: Script
script: |
  from crossplane.function.proto.v1 import run_function_pb2 as fnv1
  
  def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
      rsp.desired.resources["example"].resource.update({
          "apiVersion": "v1",
          "kind": "ConfigMap",
          "data": {"key": "value"}
      })
```

--------------------------------

### Script CRD Example: Operation

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/configuration.md

Example of the Script CRD used for operations. It defines a Python script that implements the `operate` function to set output status and timestamp.

```yaml
apiVersion: python.fn.crossplane.io/v1beta1
kind: Script
script: |
  from crossplane.function.proto.v1 import run_function_pb2 as fnv1
  
  def operate(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
      rsp.output["status"] = "completed"
      rsp.output["timestamp"] = "2024-01-01T00:00:00Z"
```

--------------------------------

### Package Entry Point Configuration

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/cli.md

Defines the `cli` function as a package entry point in `pyproject.toml`, allowing the server to be run using a simplified command.

```toml
[project.scripts]
function = "function.main:cli"
```

--------------------------------

### Operation Example: Check Certificate Expiry

Source: https://github.com/crossplane-contrib/function-python/blob/main/README.md

An example of a Crossplane Operation using function-python to perform a certificate expiry check. It demonstrates setting operation output with check time, status, and a message.

```yaml
apiVersion: ops.crossplane.io/v1alpha1
kind: Operation
metadata:
  name: check-cert-expiry
spec:
  template:
    spec:
      pipeline:
      - step: check-certificate
        functionRef:
          name: function-python
        input:
          apiVersion: python.fn.crossplane.io/v1beta1
          kind: Script
          script: |
            from crossplane.function.proto.v1 import run_function_pb2 as fnv1
            import datetime

            def operate(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
                # Example: Check certificate expiration
                # In real use, you'd retrieve and validate actual certificates

                # Set operation output
                rsp.output["check_time"] = datetime.datetime.now().isoformat()
                rsp.output["status"] = "healthy"
                rsp.output["message"] = "Certificate expiry check completed"
```

--------------------------------

### cli

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/OVERVIEW.md

Command-line entry point for the Python Function.

```APIDOC
## cli

### Description
Command-line interface entry point for the Python Function.

### Method
CLI Command

### Endpoint
N/A

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **--insecure** - Optional flag to enable insecure mode.
- **--debug** - Optional flag to enable debug mode.

### Request Example
```bash
function --insecure --debug
```

### Response
None (executes CLI commands)
```

--------------------------------

### Local Testing with Crossplane

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/OVERVIEW.md

Demonstrates how to test the Python Function locally with Crossplane. It involves running the function server in one terminal and using Crossplane to render resources in another.

```bash
# Terminal 1: Run function server
hatch run development

# Terminal 2: Test with Crossplane render
crossplane render xr.yaml composition.yaml functions.yaml -r
```

--------------------------------

### Async Composition with HTTP Requests

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/quickstart.md

Use this snippet for functions that need to make asynchronous network requests or perform I/O operations. It demonstrates fetching configuration from a remote API using httpx and updating desired resources.

```yaml
script: |
  from crossplane.function.proto.v1 import run_function_pb2 as fnv1
  import asyncio
  import httpx

  async def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
      # Async HTTP request to fetch configuration
      async with httpx.AsyncClient() as client:
          config = await client.get("https://api.example.com/config")
          data = config.json()
      
      # Use the fetched data to compose resources
      rsp.desired.resources["config"].resource.update({
          "apiVersion": "v1",
          "kind": "ConfigMap",
          "data": {
              "remote_config": str(data)
          }
      })
      rsp.desired.resources["config"].ready = True
```

--------------------------------

### Asynchronous Composition Function

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/FunctionRunner.md

An example of an asynchronous `compose` function that can perform I/O operations, demonstrated with `asyncio.sleep`. It updates a 'data' resource with an async result.

```python
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
import asyncio

async def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
    # Can perform async I/O operations
    await asyncio.sleep(0.1)
    
    rsp.desired.resources["data"].resource.update({
        "apiVersion": "v1",
        "kind": "ConfigMap",
        "data": {"result": "async result"}
    })
```

--------------------------------

### Error Response

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/endpoints.md

An example of an error response from the FunctionRunnerService. It includes a 'results' field with a 'SEVERITY_FATAL' message indicating the script's failure to define a valid function.

```json
{
  "meta": {"ttl": {"seconds": 60}},
  "results": [
    {
      "severity": "SEVERITY_FATAL",
      "message": "script must define a compose or operate function"
    }
  ]
}
```

--------------------------------

### Conditional Composition Logic in Python

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/quickstart.md

Python code demonstrating conditional resource creation based on a boolean flag and interval from the composite resource spec.

```python
def compose(req, rsp):
    spec = req.observed.composite.resource["spec"]
    
    # Only create backup if enabled
    if spec.get("backupEnabled", False):
        rsp.desired.resources["backup"].resource.update({
            "apiVersion": "backup.example.com/v1",
            "kind": "BackupPolicy",
            "spec": {"interval": spec.get("backupInterval", "daily")}
        })
        rsp.desired.resources["backup"].ready = True
```

--------------------------------

### Operation Pipeline Configuration

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/configuration.md

Configures the Python function as a pipeline step within a Crossplane Operation. This example shows a simple operation that sets an output result.

```yaml
apiVersion: ops.crossplane.io/v1alpha1
kind: Operation
metadata:
  name: example-operation
spec:
  mode: Pipeline
  pipeline:
  - step: run-operation
    functionRef:
      name: function-python
    input:
      apiVersion: python.fn.crossplane.io/v1beta1
      kind: Script
      script: |
        from crossplane.function.proto.v1 import run_function_pb2 as fnv1
        
        def operate(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
            # Perform operational task
            rsp.output["result"] = "success"
```

--------------------------------

### Async Pattern with SDK

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/SDK_Integration.md

Illustrates an asynchronous composition pattern using `httpx` for making non-blocking HTTP requests. This snippet fetches configuration data from an external API and uses it to compose a Kubernetes ConfigMap resource.

```python
import asyncio
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function import resource, response
import httpx

async def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
    try:
        # Async HTTP request
        async with httpx.AsyncClient() as client:
            config_response = await client.get(
                "https://api.example.com/config"
            )
            config = config_response.json()
        
        # Use fetched config to compose resources
        rsp.desired.resources["config"].resource.update({
            "apiVersion": "v1",
            "kind": "ConfigMap",
            "data": config
        })
        rsp.desired.resources["config"].ready = True
        
    except Exception as e:
        response.fatal(rsp, f"Async composition failed: {str(e)}")
```

--------------------------------

### Project File Manifest

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/MANIFEST.md

Lists the primary files and their purposes within the project directory. Useful for understanding the project's organization and locating specific documentation.

```text
/workspace/home/output/
├── README.md                      # This directory overview
├── MANIFEST.md                    # This file - inventory
├── INDEX.md                       # Navigation and quick reference
├── OVERVIEW.md                    # Architecture overview
├── configuration.md               # Configuration reference
├── endpoints.md                   # gRPC endpoint spec
├── types.md                       # Type reference
├── errors.md                      # Error catalog
└── api-reference/
    ├── quickstart.md              # Quick start guide
    ├── FunctionRunner.md           # Service class reference
    ├── load_module.md              # Utility function reference
    ├── cli.md                      # CLI reference
    └── SDK_Integration.md          # SDK utilities reference

```

--------------------------------

### Get Required Resource from Request

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/SDK_Integration.md

The `request.get_required_resource` function retrieves a specific resource defined in the request's requirements, essential for Operations. Import `crossplane.function.request` and `run_function_pb2`.

```python
from crossplane.function import request
from crossplane.function.proto.v1 import run_function_pb2 as fnv1

def operate(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse):
    # Get the Ingress resource (must be defined in requirements)
    ingress = request.get_required_resource(req, "ingress")
    
    if ingress is None:
        response.fatal(rsp, "Required Ingress resource not found")
        return
    
    hostname = ingress["spec"]["rules"][0]["host"]
```

--------------------------------

### Add Labels and Annotations in Python

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/quickstart.md

Python code to create a resource (e.g., an S3 bucket) and apply labels derived from the composite resource's tags.

```python
def compose(req, rsp):
    composite = req.observed.composite.resource
    
    # Get tags from composite
    tags = composite["spec"].get("tags", {})
    
    # Create resource with tags as labels
    rsp.desired.resources["bucket"].resource.update({
        "apiVersion": "s3.aws.upbound.io/v1beta2",
        "kind": "Bucket",
        "metadata": {
            "name": composite["metadata"]["name"],
            "labels": tags
        },
        "spec": {"forProvider": {"region": "us-east-1"}}
    })
```

--------------------------------

### Load and Use a User Script Module

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/load_module.md

Demonstrates how to use the load_module function to load a Python script string into a module and access its contents, such as constants and functions.

```python
from function.fn import load_module

# Load a script as a module
script = """
def compose(req, rsp):
    rsp.desired.resources["example"].resource.update({
        "apiVersion": "v1",
        "kind": "ConfigMap"
    })
    
MY_CONSTANT = 42
"""

module = load_module("user_script", script)

# Access module contents
print(module.MY_CONSTANT)  # Output: 42
print(callable(module.compose))  # Output: True
```

--------------------------------

### Importing Protocol Buffer Messages and gRPC

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/SDK_Integration.md

Import the main protocol buffer message types and gRPC service definitions for function execution from `crossplane.function.proto.v1`.

```python
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1
```

--------------------------------

### Validate Python Function Definition

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/quickstart.md

Ensure your Python script defines exactly one function named 'compose' to be valid for Crossplane function execution. This snippet shows valid and invalid examples.

```python
# ✓ Valid
def compose(req, rsp):
    pass
```

```python
# ✗ Invalid - both functions
def compose(req, rsp):
    pass

def operate(req, rsp):
    pass
```

```python
# ✗ Invalid - wrong function name
def my_function(req, rsp):
    pass
```

--------------------------------

### Usage with Environment Variables

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/cli.md

Demonstrates using the `TLS_SERVER_CERTS_DIR` environment variable to specify the path for mTLS certificates, which is an alternative to using the `--tls-certs-dir` flag.

```bash
export TLS_SERVER_CERTS_DIR=/etc/crossplane/certs
python -m function.main
```

--------------------------------

### Enable Debug Logging

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/configuration.md

Enable detailed debug-level logging for troubleshooting. Use the short or long flag for this setting.

```bash
function --debug
function -d
```

--------------------------------

### Get Multiple Required Resources from Request

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/SDK_Integration.md

Use `request.get_required_resources` to fetch all resources matching a specified name from the request, useful when dealing with multiple instances of a resource type. The `crossplane.function.request` module is needed.

```python
from crossplane.function import request

def operate(req, rsp):
    # Get all matching resources
    databases = request.get_required_resources(req, "database")
    
    for db in databases:
        # Process each database
        backup_database(db)
```

--------------------------------

### Function Definition and Parameters

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/cli.md

The `cli` function is the main entry point for the Crossplane Python function server. It accepts various command-line arguments to configure its behavior, such as debug logging, network address, TLS settings, and message size limits.

```APIDOC
## Function: cli

### Description

The `cli` function is the entry point for the function-python server. It configures logging, initializes the gRPC server, loads mTLS credentials, sets message size limits, and starts the server.

### Method

Not applicable (CLI command)

### Endpoint

Not applicable (CLI command)

### Parameters

#### Command-line Options

- **--debug, -d** (`bool`) - Optional - Default: `False` - Enable debug-level logging.
- **--address** (`str`) - Optional - Default: `0.0.0.0:9443` - Host and port for gRPC server to listen on. Format: `host:port`.
- **--tls-certs-dir** (`str`) - Optional - Default: `None` - Path to directory containing mTLS certificates. Can also be set via `TLS_SERVER_CERTS_DIR` environment variable. Ignored if `--insecure` is set.
- **--insecure** (`bool`) - Optional - Default: `False` - Run without mTLS encryption. If set, `tls_certs_dir` is ignored.
- **--max-recv-message-size** (`int`) - Optional - Default: `4` - Maximum size of received gRPC messages in MB.
- **--max-send-message-size** (`int`) - Optional - Default: `4` - Maximum size of sent gRPC messages in MB.

### Returns

**Type:** `None`

### Throws/Rejects

The function catches all exceptions and displays them to the user via `click.echo()`. Common errors include certificate loading failures, network binding failures, and invalid certificate formats.
```

--------------------------------

### Run Unit Tests

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/OVERVIEW.md

Executes the unit tests for the Python Function. This command is useful for verifying the correctness of the function's components.

```bash
hatch run -e test unit
# Runs: python -m unittest tests/*.py
```

--------------------------------

### Enable Insecure Mode

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/configuration.md

Run the server without mTLS encryption for development and testing. This flag ignores the --tls-certs-dir option.

```bash
function --insecure
function --insecure --debug
```

--------------------------------

### Handle Module Spec Creation Failure in Python

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/errors.md

Illustrates how to catch a `RuntimeError` that occurs if the Python module specification cannot be created during script loading. This indicates a potential environment or Python interpreter issue.

```python
from function.fn import load_module

try:
    module = load_module("script", "def compose(req, rsp): pass")
except RuntimeError as e:
    if "cannot create module spec" in str(e):
        print(f"Critical error: {e}")
        # This indicates an environment problem, not user code
```

--------------------------------

### Define and Use Custom ComposedResource Class

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/types.md

Define custom data classes like `ComposedResource` to structure complex resource definitions within your Python functions. This example shows how to use the custom class to update desired resources.

```python
from dataclasses import dataclass
from typing import Any

@dataclass
class ComposedResource:
    name: str
    definition: dict[str, Any]
    ready: bool = False

def compose(req, rsp):
    resource = ComposedResource(
        name="example",
        definition={
            "apiVersion": "v1",
            "kind": "ConfigMap"
        }
    )
    
    rsp.desired.resources[resource.name].resource.update(
        resource.definition
    )
    rsp.desired.resources[resource.name].ready = resource.ready
```

--------------------------------

### Set Server Address

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/configuration.md

Configure the host and port for the gRPC server. Use this to specify where the function listens for requests.

```bash
function --address 0.0.0.0:9443
function --address 127.0.0.1:5000
function --address localhost:9443
```

--------------------------------

### Handle Neither Compose nor Operate Function Defined Error in Python

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/errors.md

Demonstrates triggering and checking for the error when a Python script defines neither a 'compose' nor an 'operate' function. One of these must be present for the script to be executable.

```python
from function.fn import FunctionRunner
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function import resource

async def handle_no_function_error():
    runner = FunctionRunner()
    
    # Script with no compose or operate function
    script = """
from crossplane.function.proto.v1 import run_function_pb2 as fnv1

def some_helper_function():
    return "not a compose or operate function"
"""
    
    req = fnv1.RunFunctionRequest(
        input=resource.dict_to_struct({"script": script})
    )
    
    response = await runner.RunFunction(req, None)
    
    # Check for this error
    if response.results:
        for result in response.results:
            if "compose or operate function" in result.message:
                print(f"Error: {result.message}")
```

--------------------------------

### Get Logger Instance for Structured Logging

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/SDK_Integration.md

Obtain a logger instance using `logging.get_logger()` for structured logging within Crossplane Functions. The `crossplane.function.logging` module is required. Log levels like debug, info, warning, and error are available.

```python
from crossplane.function import logging

log = logging.get_logger()

def compose(req, rsp):
    log.info("Starting composition")
    
    # With context
    log = log.bind(resource="bucket", region="us-east-1")
    log.debug("Creating bucket in region")
```

--------------------------------

### Create and Access Protobuf Duration

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/types.md

Instantiate `google.protobuf.Duration` to represent time intervals and access its `seconds` and `nanos` properties. This is useful for setting timeouts or time-based configurations.

```python
from google.protobuf import duration_pb2

# Create duration of 60 seconds
ttl = duration_pb2.Duration(seconds=60)

# Access properties
print(ttl.seconds)  # Output: 60
```

--------------------------------

### Render Operation with Required Resources

Source: https://github.com/crossplane-contrib/function-python/blob/main/example/operation/README.md

Render a Crossplane Operation, including function results and specifying local files as required resources for testing.

```bash
crossplane alpha render op operation.yaml functions.yaml --required-resources . -r
```

--------------------------------

### Type-Safe Function Composition

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/SDK_Integration.md

Demonstrates how to use the Crossplane Function SDK for Python with type annotations for IDE autocompletion and type checking. Shows accessing nested request data and setting response values.

```python
from crossplane.function.proto.v1 import run_function_pb2 as fnv1

# IDE knows the types and can provide autocomplete
def compose(req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse) -> None:
    # req has attributes: meta, input, observed, desired, context
    # rsp has attributes: meta, results, desired, observed, context, output
    
    # Accessing nested structures with IDE support
    region: str = req.observed.composite.resource["spec"]["region"]
    
    # Setting response values with type checking
    rsp.desired.resources["bucket"].ready = True
```

--------------------------------

### Enable Debug Logging for Function Execution

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/errors.md

Run the function with the `--debug` flag to enable verbose logging. This is useful for development and troubleshooting, providing insights into script loading, function detection, and execution details.

```bash
function --debug --insecure
```

--------------------------------

### Import Crossplane SDK Utilities

Source: https://github.com/crossplane-contrib/function-python/blob/main/_autodocs/api-reference/quickstart.md

Import necessary modules from the Crossplane SDK to access protocol buffers and other utilities within your Python function.

```python
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function import resource, response, logging
```

--------------------------------

### Run Function in Development Mode

Source: https://github.com/crossplane-contrib/function-python/blob/main/example/operation/README.md

Execute the function in development mode to facilitate local testing of Crossplane Operations.

```bash
hatch run development
```