Terraform External Python Script Framework

README.md

Terraform External Python Script

The below example script handles most of the annoying parts of implementing the protocol between Python and Terraform when using the external resource type. In order to prevent required dependencies, this script only uses built-in Python modules for argument parsing and result serialization. It will current do the following:

• Automatically parse and validate input arguments based on defined dataclass as defined by the Terraform protocol specification.
• Automatically serialize output results based on defined dataclass as defined by the Terraform protocol specification.
• Automatically handle any uncaught exceptions and output errors in accordance with the Terraform protocol specification.

Defining Input Arguments

You should first define the dataclass representing your input arguments by editing the QueryInput class. This is a standard Python dataclass. The parsing is not complex, but is capable of recognizing arguments with defaults as optional (via dataclasses.field(default=...) or dataclasses.field(default_factor=...)).

The parse_input() function will also automatically return errors to Terraform using exit_error if any of the following conditions are detected:

• Parsing the JSON document from stdin fails.
• The JSON object passed via stdin is not a dictionary.
• A query argument was passed that does not exist in QueryInput.
• A field of QueryInput was missing from the input and has no defaults.

Terraform only allows a single key-value input to external scripts, and values are restricted to the string type. The parser implemented in parse_input will attempt to coerce input strings to the declared field type by simply calling field.type(value). This does not work in all cases. For more complex types, you should probably override your constructor in order to sainly handle this.

Defining Response Parameters

The Response dataclass is used to return results from your main function. It will be serialized automatically to JSON before returning it to Terraform. You can add any fields you like.

NOTE: Terraform only allows string key-value pairs in the response, so the exit_result function will attempt to coerce any values in your Result object to a string using str(). Additionally, bool values are coerced using str(value).lower() in order to produce the nicer true and false strings.

Defining The Main Routine

The main() function is used for your main script execution. It is only called if the input was correctly validated from QueryInput, and is passed an instance of QueryInput as it's only argument.

Any exceptions raised in main() will be caught. An error message containing the string representation of the exception object will be printed to stderr and an exit code of 1 will be returned from the process in accordance with the Terraform protocol specification.

The main() function should return an instance of the Result class which will be automatically serialized and returned to Terraform to be stored in the Terraform State.

example.py

# Author:      Caleb Stewart
# Description: Core structure for parsing arguments, returning results and handling errors
#              in a Terraform external Python script source.
# License:     MIT
import dataclasses
import json
import sys
from typing import IO


@dataclasses.dataclass
class QueryInput(object):
    """Defines the expected input parameters from Terraform"""

    # A required argument
    name: str
    # An optional argument
    greeting: str = dataclasses.field(default="hello")


@dataclasses.dataclass
class Response(object):
    """Defines the expected output structure back to terraform."""

    greeting: str


def main(query: QueryInput) -> Response:
    """Main entrypoint. Receives the parsed query arguments and returns
    a response object which is automatically serialized and output properly.
    Any uncaptured exceptions will be treated as errors and the exception
    message will be printed with exit_error. Otherwise, the response is
    returned with exit_result."""

    return Response(greeting=f"{query.greeting} {query.name}")


def entrypoint():
    """Entrypoint called below if __name__ is __main__"""

    try:
        # Parse input query into a python object
        query = parse_input(sys.stdin)

        # Execute main entrypoint with arguments
        result = main(query)

        # Return the result
        exit_result(result)
    except Exception as exc:
        # Respond in accordance with terraform spec on errors.
        exit_error("uncaught exception: {exc}", exc=exc)


def parse_input(stream: IO):
    """Parse and validate arguments and return them"""

    # Load the input document
    document = json.load(stream)
    assert isinstance(document, dict), "query input is not a dictionary"

    # Get a mapping of field names to Field objects
    fields = {f.name: f for f in dataclasses.fields(QueryInput)}

    # The resulting kwargs passed to the dataclass constructor
    kwargs = {}

    for key, value in document.items():
        assert key in fields, f"unknown query argument: {key}"

        try:
            # Initialize the args array with the correct type conversion
            # We expect that terraform will send strings to us.
            kwargs[key] = fields[key].type(value)
        except ValueError as exc:
            raise AssertionError(f"query argument value error: {key}: {exc}") from exc

    # Ensure there are no missing keys
    for key in fields.keys():
        if (
            key not in kwargs
            and fields[key].default is None
            and fields[key].default_factory is None
        ):
            raise AssertionError(f"missing query argument: {key}")

    return QueryInput(**kwargs)


def exit_error(message: str, status: int = 1, **kwargs):
    """Format the message with kwargs, and then print it to stderr and
    exit with a non-zero exit status. You can control the exit status
    if needed with the status argument."""

    print(message.format(**kwargs), file=sys.stderr)
    sys.exit(status)


def exit_result(response: Response):
    """Serialize the response and print to stdout. Thn exit with a
    zero status to indicate success."""

    # We can only return a dataclass
    assert dataclasses.is_dataclass(response), "only dataclasses can be returned"

    # Resulting serialized dataclass object
    output = {}

    # Iterate over all fields
    for field in dataclasses.fields(response):
        key = field.name
        value = getattr(response, field.name)

        # Special handling for bool, since we like "true" instead of "True"
        # for output back to Terraform
        if isinstance(value, bool):
            value = str(value).lower()
        elif not isinstance(value, str):
            # Terraform only accepts string outputs.
            value = str(value)

        output[key] = value

    # Dump the JSON blob for the output, and exit with a success code
    print(json.dumps(output))
    sys.exit(0)


if __name__ == "__main__":
    entrypoint()

example.tf

data "external" "greeting" {
  program = ["python3", "${path.module}/external/example.py"]
  
  query = {
    # Required argument
    name = "caleb"
    # We could also provide the optional argumeent but are not required
    # greeting = "goodbye"
  }
}