fix(sdk/python): Allow for duplicate output values in python programs

I used the node SDK as inspiration for this change - there were a few things
we were doing differently in python (specifically handling all cases in
the outer layer of `massage` rather than allowing for more fine-grained
control in a `massage_complex` function like in node.  We also take the stack
concept from the node SDK to resolve the immediate issue of duplicate outputs.

changelog/pending/20221206--sdk-python--allows-for-duplicate-output-values-in-python.yaml

@@ -0,0 +1,4 @@
+changes:
+- type: fix
+  scope: sdk/python
+  description: Allows for duplicate output values in python

sdk/python/.gitignore

@@ -6,3 +6,4 @@
 .venv/
 venv/
 .coverage
+build/

sdk/python/lib/pulumi/runtime/stack.py

@@ -174,54 +174,67 @@ def massage(attr: Any, seen: List[Any]):
     if is_primitive(attr):
         return attr
 
+    if isinstance(attr, Output):
+        return attr.apply(lambda v: massage(v, seen))
+
+    if isawaitable(attr):
+        return Output.from_input(attr).apply(lambda v: massage(v, seen))
+
     # from this point on, we have complex objects.  If we see them again, we don't want to emit them
     # again fully or else we'd loop infinitely.
     if reference_contains(attr, seen):
         # Note: for Resources we hit again, emit their urn so cycles can be easily understood in
         # the popo objects.
         if isinstance(attr, Resource):
-            return attr.urn
-
+            return massage(attr.urn, seen)
         # otherwise just emit as nothing to stop the looping.
         return None
 
-    seen.append(attr)
-
-    # first check if the value is an actual dictionary.  If so, massage the values of it to deeply
-    # make sure this is a popo.
-    if isinstance(attr, dict):
-        result = {}
-        # Don't use attr.items() here, as it will error in the case of outputs with an `items` property.
-        for key in attr:
-            # ignore private keys
-            if not key.startswith("_"):
-                result[key] = massage(attr[key], seen)
+    try:
+        seen.append(attr)
+        return massage_complex(attr, seen)
+    finally:
+        popped = seen.pop()
+        if popped is not attr:
+            raise Exception("Invariant broken when processing stack outputs")
 
-        return result
 
-    if isinstance(attr, Output):
-        return attr.apply(lambda v: massage(v, seen))
+def massage_complex(attr: Any, seen: List[Any]) -> Any:
+    def is_public_key(key: str) -> bool:
+        return not key.startswith("_")
 
-    if isawaitable(attr):
-        return Output.from_input(attr).apply(lambda v: massage(v, seen))
+    def serialize_all_keys(include: Callable[[str], bool]):
+        plain_object: Dict[str, Any] = {}
+        for key in attr.__dict__.keys():
+            if include(key):
+                plain_object[key] = massage(attr.__dict__[key], seen)
+        return plain_object
 
     if isinstance(attr, Resource):
-        result = massage(attr.__dict__, seen)
+        serialized_attr = serialize_all_keys(is_public_key)
 
         # In preview only, we mark the result with "@isPulumiResource" to indicate that it is derived
         # from a resource. This allows the engine to perform resource-specific filtering of unknowns
         # from output diffs during a preview. This filtering is not necessary during an update because
         # all property values are known.
-        if is_dry_run():
-            result["@isPulumiResource"] = True
-        return result
+        return (
+            serialized_attr
+            if not is_dry_run()
+            else {**serialized_attr, "@isPulumiResource": True}
+        )
 
-    if hasattr(attr, "__dict__"):
-        # recurse on the dictionary itself.  It will be handled above.
-        return massage(attr.__dict__, seen)
+    # first check if the value is an actual dictionary.  If so, massage the values of it to deeply
+    # make sure this is a popo.
+    if isinstance(attr, dict):
+        # Don't use attr.items() here, as it will error in the case of outputs with an `items` property.
+        return {
+            key: massage(attr[key], seen) for key in attr if not key.startswith("_")
+        }
+
+    if hasattr(attr, "__iter__"):
+        return [massage(item, seen) for item in attr]
 
-    # finally, recurse through iterables, converting into a list of massaged values.
-    return [massage(a, seen) for a in attr]
+    return serialize_all_keys(is_public_key)
 
 
 def reference_contains(val1: Any, seen: List[Any]) -> bool:

sdk/python/lib/test/langhost/stack_output/__main__.py

@@ -18,6 +18,11 @@ def __init__(self):
         self.num = 1
         self._private = 2
 
+class DuplicateOutputClass:
+    id = 1
+
+my_duplicate_class = DuplicateOutputClass()
+
 recursive = {"a": 1}
 recursive["b"] = 2
 recursive["c"] = recursive
@@ -34,3 +39,5 @@ def __init__(self):
 pulumi.export("output", pulumi.Output.from_input(1))
 pulumi.export("class", TestClass())
 pulumi.export("recursive", recursive)
+pulumi.export("duplicate_output_0", my_duplicate_class.id)
+pulumi.export("duplicate_output_1", my_duplicate_class.id)

sdk/python/lib/test/langhost/stack_output/test_stack_output.py

@@ -39,4 +39,6 @@ def register_resource_outputs(self, _ctx, _dry_run, _urn, ty, _name, _resource,
             "output": 1.0,
             "class": {"num": 1.0},
             "recursive": {"a": 1.0, "b": 2.0},
+            "duplicate_output_0": 1.0,
+            "duplicate_output_1": 1.0,
         }, outputs)