Merge #11607

11607: Add json_dumps to python sdk r=Frassle a=Frassle

<!--- 
Thanks so much for your contribution! If this is your first time contributing, please ensure that you have read the [CONTRIBUTING](https://github.com/pulumi/pulumi/blob/master/CONTRIBUTING.md) documentation.
-->

# Description

<!--- Please include a summary of the change and which issue is fixed. Please also include relevant motivation and context. -->

Following on with adding to json functions to each SDK, this one is for Python.
It even handles nested Outputs via our _sync_await helper.

## Checklist

<!--- Please provide details if the checkbox below is to be left unchecked. -->
- [x] I have added tests that prove my fix is effective or that my feature works
<!--- 
User-facing changes require a CHANGELOG entry.
-->
- [x] I have run `make changelog` and committed the `changelog/pending/<file>` documenting my change
<!--
If the change(s) in this PR is a modification of an existing call to the Pulumi Service,
then the service should honor older versions of the CLI where this change would not exist.
You must then bump the API version in /pkg/backend/httpstate/client/api.go, as well as add
it to the service.
-->
- [ ] Yes, there are changes in this PR that warrants bumping the Pulumi Service API version
  <!-- `@Pulumi` employees: If yes, you must submit corresponding changes in the service repo. -->


Co-authored-by: Fraser Waters <fraser@pulumi.com>

changelog/pending/20221209--sdk-python--add-jsondumps-to-python-sdk.yaml

@@ -0,0 +1,4 @@
+changes:
+- type: feat
+  scope: sdk/python
+  description: Add json_dumps to python sdk.

sdk/python/lib/pulumi/output.py

@@ -13,9 +13,11 @@
 # limitations under the License.
 import asyncio
 import contextlib
+import json
 from functools import reduce
 from inspect import isawaitable
 from typing import (
+    Tuple,
     TypeVar,
     Generic,
     Set,
@@ -35,6 +37,7 @@
 from . import _types
 from . import runtime
 from .runtime import rpc
+from .runtime.sync_await import _sync_await
 
 if TYPE_CHECKING:
     from .resource import Resource
@@ -536,6 +539,130 @@ def format(
             )
         return Output.from_input(format_string).apply(lambda str: str.format())
 
+    @staticmethod
+    def json_dumps(
+        obj: Input[Any],
+        *,
+        skipkeys: bool = False,
+        ensure_ascii: bool = True,
+        check_circular: bool = True,
+        allow_nan: bool = True,
+        cls: type[json.JSONEncoder] = None,
+        indent: Optional[int | str] = None,
+        separators: tuple[str, str] = None,
+        default: Callable[[Any], Any] = None,
+        sort_keys: bool = False,
+        **kw: Any
+    ) -> "Output[str]":
+        """
+        Uses json.dumps to serialize the given Input[object] value into a JSON string.
+
+        The arguments have the same meaning as in `json.dumps` except obj is an Input.
+        """
+
+        if cls is None:
+            cls = json.JSONEncoder
+
+        output = Output.from_input(obj)
+        result_resources: asyncio.Future[Set["Resource"]] = asyncio.Future()
+        result_is_known: asyncio.Future[bool] = asyncio.Future()
+        result_is_secret: asyncio.Future[bool] = asyncio.Future()
+
+        async def run() -> str:
+            resources: Set["Resource"] = set()
+            try:
+                seen_unknown = False
+                seen_secret = False
+                seen_resources = set()
+
+                class OutputEncoder(cls):  # type: ignore
+                    def default(self, o):
+                        if isinstance(o, Output):
+                            nonlocal seen_unknown
+                            nonlocal seen_secret
+                            nonlocal seen_resources
+
+                            # We need to synchronously wait for o to complete
+                            async def wait_output() -> Tuple[object, bool, bool, set]:
+                                return (
+                                    await o._future,
+                                    await o._is_known,
+                                    await o._is_secret,
+                                    await o._resources,
+                                )
+
+                            (result, known, secret, resources) = _sync_await(
+                                asyncio.ensure_future(wait_output())
+                            )
+                            # Update the secret flag and set of seen resources
+                            seen_secret = seen_secret or secret
+                            seen_resources.update(resources)
+                            if known:
+                                return result
+                            # The value wasn't known set the local seenUnknown variable and just return None
+                            # so the serialization doesn't raise an exception at this point
+                            seen_unknown = True
+                            return None
+
+                        return super().default(o)
+
+                # Await the output's details.
+                resources = await output._resources
+                is_known = await output._is_known
+                is_secret = await output._is_secret
+                value = await output._future
+
+                if not is_known:
+                    result_resources.set_result(resources)
+                    result_is_known.set_result(is_known)
+                    result_is_secret.set_result(is_secret)
+                    return cast(str, None)
+
+                # Try and dump using our special OutputEncoder to handle nested outputs
+                result = json.dumps(
+                    value,
+                    skipkeys=skipkeys,
+                    ensure_ascii=ensure_ascii,
+                    check_circular=check_circular,
+                    allow_nan=allow_nan,
+                    cls=OutputEncoder,
+                    indent=indent,
+                    separators=separators,
+                    default=default,
+                    sort_keys=sort_keys,
+                    **kw
+                )
+
+                # Update the final resources and secret flag based on what we saw while dumping
+                is_secret = is_secret or seen_secret
+                resources = set(resources)
+                resources.update(seen_resources)
+
+                # If we saw an unknown during dumping then throw away the result and return not known
+                if seen_unknown:
+                    result_resources.set_result(resources)
+                    result_is_known.set_result(False)
+                    result_is_secret.set_result(is_secret)
+                    return cast(str, None)
+
+                result_resources.set_result(resources)
+                result_is_known.set_result(True)
+                result_is_secret.set_result(is_secret)
+                return result
+
+            finally:
+                with contextlib.suppress(asyncio.InvalidStateError):
+                    result_resources.set_result(resources)
+
+                with contextlib.suppress(asyncio.InvalidStateError):
+                    result_is_known.set_result(False)
+
+                with contextlib.suppress(asyncio.InvalidStateError):
+                    result_is_secret.set_result(False)
+
+        run_fut = asyncio.ensure_future(run())
+        return Output(result_resources, run_fut, result_is_known, result_is_secret)
+
     def __str__(self) -> str:
         return """Calling __str__ on an Output[T] is not supported.
 

sdk/python/lib/test/test_output.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 import asyncio
 import unittest
 from typing import Mapping, Optional, Sequence, cast
@@ -294,4 +295,77 @@ async def test_args_and_kwags(self):
         s = Output.from_input("hi")
         x = Output.format("{0}, {s}", i, s=s)
         x_val = await x.future()
-        self.assertEqual(x_val, "1, hi")
+        self.assertEqual(x_val, "1, hi")
+
+class OutputJsonTests(unittest.TestCase):
+    @pulumi_test
+    async def test_basic(self):
+        i = Output.from_input([0, 1])
+        x = Output.json_dumps(i)
+        self.assertEqual(await x.future(), "[0, 1]")
+        self.assertEqual(await x.is_secret(), False)
+        self.assertEqual(await x.is_known(), True)
+
+    # from_input handles _most_ nested outputs, so we need to use user defined types to "work around"
+    # that, which means we also need to use a custom encoder
+    class CustomClass(object):
+        def __init__(self, a, b):
+            self.a = a
+            self.b = b
+
+    class CustomEncoder(json.JSONEncoder):
+        def default(self, o):
+            if isinstance(o, OutputJsonTests.CustomClass):
+                return (o.a, o.b)
+            return json.JSONEncoder.default(self, o)
+
+    @pulumi_test
+    async def test_nested(self):
+        i = Output.from_input(OutputJsonTests.CustomClass(Output.from_input(0), Output.from_input(1)))
+        x = Output.json_dumps(i, cls=OutputJsonTests.CustomEncoder)
+        self.assertEqual(await x.future(), "[0, 1]")
+        self.assertEqual(await x.is_secret(), False)
+        self.assertEqual(await x.is_known(), True)
+
+    @pulumi_test
+    async def test_nested_unknown(self):
+        future = asyncio.Future()
+        future.set_result(None)
+        is_known = asyncio.Future()
+        is_known.set_result(False)
+        unknown = Output(resources=set(), future=future, is_known=is_known)
+
+        i = Output.from_input(OutputJsonTests.CustomClass(unknown, Output.from_input(1)))
+        x = Output.json_dumps(i, cls=OutputJsonTests.CustomEncoder)
+        self.assertEqual(await x.is_secret(), False)
+        self.assertEqual(await x.is_known(), False)
+
+    @pulumi_test
+    async def test_nested_secret(self):
+        future = asyncio.Future()
+        future.set_result(0)
+        future_true = asyncio.Future()
+        future_true.set_result(True)
+        inner = Output(resources=set(), future=future, is_known=future_true, is_secret=future_true)
+
+        i = Output.from_input(OutputJsonTests.CustomClass(inner, Output.from_input(1)))
+        x = Output.json_dumps(i, cls=OutputJsonTests.CustomEncoder)
+        self.assertEqual(await x.future(), "[0, 1]")
+        self.assertEqual(await x.is_secret(), True)
+        self.assertEqual(await x.is_known(), True)
+
+    @pulumi_test
+    async def test_nested_dependencies(self):
+        future = asyncio.Future()
+        future.set_result(0)
+        future_true = asyncio.Future()
+        future_true.set_result(True)
+        resource = object()
+        inner = Output(resources=set([resource]), future=future, is_known=future_true)
+
+        i = Output.from_input(OutputJsonTests.CustomClass(inner, Output.from_input(1)))
+        x = Output.json_dumps(i, cls=OutputJsonTests.CustomEncoder)
+        self.assertEqual(await x.future(), "[0, 1]")
+        self.assertEqual(await x.is_secret(), False)
+        self.assertEqual(await x.is_known(), True)
+        self.assertIn(resource, await x.resources())