Allow (JSON formatted) input strings to be dictionaries

[tma-unwire]

Hello!

• Vote on this issue by adding a 👍 reaction
• If you want to implement this feature, comment to let us know (we'll work with you on design, scheduling, etc.)

Issue details

When working with AWS resources you often have input parameters that are JSON formatted strings that contains the configuration for the resource. A typical example is the various IAM resources with roles and policies. Sometimes these even include Output from other resources.

In all cases, it can be nice - and a lot more readable - to inline the dictionary instead of having to use json.dumps(...), Output.apply(...) or even Output.all(...).apply(...)

A real world example

import json
from typing import Any, Mapping

import pulumi
import pulumi_random as random
from pulumi_aws import rds, ssm


def setup_rds_password_and_rotation(cluster: rds.Cluster, password: random.RandomPassword) -> None:
    def make_secret_string(data: Mapping[str, Any]) -> str:
        return json.dumps({
            'engine': 'mysql',
            'dbClusterIdentifier': 'rds-cluster',
            'host': data.get('endpoint'),
            'username': 'root',
            'password': data.get('password')
        })

    param = ssm.Parameter(
        '...',
        name='...',
        type='SecureString',
        data_type='text',
        value=pulumi.Output.all(endpoint=cluster.endpoint, password=password.result).apply(make_secret_string),
        description="Admin password used for RDS cluster",
    )

It would be so much nicer to be able to write this:

def setup_rds_password_and_rotation(cluster: rds.Cluster, password: random.RandomPassword) -> None:
    param = ssm.Parameter(
        '...',
        name='...',
        type='SecureString',
        data_type='text',
        value={
            'engine': 'mysql',
            'dbClusterIdentifier': 'rds-cluster',
            'host': cluster.endpoint,
            'username': 'root',
            'password': password.result
        },
        description="Admin password used for RDS cluster",
    )

As value is defined as a string, then Pulumi should automatically wait ion the result and jsonify the results..

Affected area/feature

Primarily in AWS, but something similar could also be useful in Kubernetes.

[tma-unwire]

I really though this would simplify my current sources, so I implemented it :-)

Please have a look at give me some feedback.

DATA_TYPE = Union[Sequence, Mapping]


def jsonify_structure(data: DATA_TYPE) -> Union[str, pulumi.Output[str]]:
    """Given a recursive list or map of data, returns a JSON formatted string.

    If the structure includes any Output elements, the returned string is packaged as an Object using
    Output.all(...).apply(...). """

    found_outputs: List[pulumi.Output] = []
    # pulumi.log.info(f"IN: {pprint.pformat(data)}")
    resulting_data = collect_outputs(found_outputs=found_outputs, data=data)
    # pulumi.log.info(f"OUT: {resulting_data}")

    if len(found_outputs) == 0:
        return json.dumps(data)

    fname = 'substitute_args'
    exec(f"""def {fname}(args): return {resulting_data}""", globals(), locals())
    f = locals().get(fname)

    return pulumi.Output.all(*found_outputs).apply(lambda args: json.dumps(f(args)))


def collect_outputs(*, data: DATA_TYPE, found_outputs: List[pulumi.Output]) -> str:
    """Given a data structure returns a string representation of this with where
    any references to found Output object is replaced with 'args[...]' and all of these accumulated in found_outputs."""
    if isinstance(data, dict):
        return '{' + ', '.join([collect_outputs(found_outputs=found_outputs, data=n) +
                                ': ' +
                                collect_outputs(found_outputs=found_outputs, data=v)
                                for n, v in data.items()]) + '}'
    if isinstance(data, list):
        return '[' + ', '.join([repr(collect_outputs(found_outputs=found_outputs, data=i)) for i in data]) + ']'
    if isinstance(data, pulumi.Output):
        if data not in found_outputs:
            found_outputs.append(data)
        return f'args[{found_outputs.index(data)}]'

    # constant - just passed through
    return repr(data)

[guineveresaenger]

Hi @tma-unwire - as you noted in your AWS Native issue, this is behavior that could be great to implement elsewhere as well.

I would be curious to see if you had interest to bring this to pulumi/pulumi SDK generation implementation so it is available across all providers, provided the maintaining team is open to the idea.

cc @stack72 for thoughts on this?

[tma-unwire]

I'll be happy to do that.

It misses some tests, but otherwise it works fine for me as is. So you're welcome to take the code, modify, include, delete, etc as you see fit...

The 'json.dump(...)' could be supplied as an argument and that way we could support YAML and other sorts of serializations.

[guineveresaenger]

@tma-unwire - we would love to see a PR with your suggested changes if you're up for it! :)

[tma-unwire]

I found an error. New version below:

DATA_TYPE = Union[Sequence, Mapping]


def stringify_structure(data: DATA_TYPE, serializer: Callable[[DATA_TYPE], str] = json.dumps) -> Union[str, pulumi.Output[str]]:
    """Given a recursive list or map of data, returns a JSON formatted string.

    If the structure includes any Output elements, the returned string is packaged as an Object using
    Output.all(...).apply(...). """

    found_outputs: List[pulumi.Output] = []
    # pulumi.log.info(f"IN: {pprint.pformat(data)}")
    resulting_data = collect_outputs(found_outputs=found_outputs, data=data)
    # pulumi.log.info(f"OUT: {resulting_data}")

    if len(found_outputs) == 0:
        return serializer(data)

    fname = 'substitute_args'
    exec(f"""def {fname}(args): return {resulting_data}""", globals(), locals())
    f = locals().get(fname)

    return pulumi.Output.all(*found_outputs).apply(lambda args: serializer(f(args)))


def collect_outputs(*, data: DATA_TYPE, found_outputs: List[pulumi.Output]) -> str:
    """Given a data structure returns a string representation of this with where
    any references to found Output object is replaced with 'args[...]' and all of these accumulated in found_outputs."""
    # pulumi.log.info(f"IN: {pprint.pformat(data)}")
    if isinstance(data, dict):
        return '{' + ', '.join([collect_outputs(found_outputs=found_outputs, data=n) +
                                ': ' +
                                collect_outputs(found_outputs=found_outputs, data=v)
                                for n, v in data.items()]) + '}'
    if isinstance(data, list):
        return '[' + ', '.join([collect_outputs(found_outputs=found_outputs, data=i) for i in data]) + ']'
    if isinstance(data, pulumi.Output):
        if data not in found_outputs:
            found_outputs.append(data)
        return f'args[{found_outputs.index(data)}]'

    # constant - just passed through
    return repr(data)

[lukehoban]

This feature has been implemented in pulumi/pulumi#11607.