code: def testGetLiveOutputArtifactsOfNodeByOutputKeyAsync(self):
      c1 = self.put_context("node", "my-pipeline.my-node")
  
      with self.subTest("With no LIVE node artifacts"):
          result = (
              store_ext.get_live_output_artifacts_of_node_by_output_key(
                  self.store,
                  pipeline_id="my-pipeline",
                  node_id="my-node",
                  pipeline_run_id="",
              )
          )
          self.assertEmpty(result)
  
      x1 = self.put_artifact("X")
      x2 = self.put_artifact("X")
      x3 = self.put_artifact("X")
      # Intermediate artifact
      x4 = self.put_artifact("X")
  
      y1 = self.put_artifact("Y")
      y2 = self.put_artifact("Y", state="DELETED")
      y3 = self.put_artifact("Y")
      y4 = self.put_artifact("Y")
      y5 = self.put_artifact("Y")
      # Intermediate artifact
      y6 = self.put_artifact("Y")
  
      z1 = self.put_artifact("Z")
      z2 = self.put_artifact("Z")
      z3 = self.put_artifact("Z", state="ABANDONED")
      # Intermediate artifact
      z4 = self.put_artifact("Z")
  
      self.put_execution(
          "E",
          last_known_state="CACHED",
          inputs={"x": [x1]},
          outputs={"y": [y1], "z": [z1]},
          contexts=[c1],
      )
      self.put_execution(
          "E",
          last_known_state="COMPLETE",
          inputs={"x": [x2]},
          outputs={"y": [y2, y3, y4], "z": [z2]},
          contexts=[c1],
      )
      self.put_execution(
          "E",
          last_known_state="COMPLETE",
          inputs={"x": [x3]},
          outputs={"y": [y5], "z": [z3]},
          contexts=[c1],
      )
      # A special case for intermediate artifacts, which can be LIVE even the
      # execution is FAILED.
      self.put_execution(
          "E",
          last_known_state="FAILED",
          inputs={"x": [x4]},
          outputs={"y": [y6], "z": [z4]},
          contexts=[c1],
      )
  
      with self.subTest("With execution_states unspecified"):
          result = (
              store_ext.get_live_output_artifacts_of_node_by_output_key(
                  self.store,
                  pipeline_id="my-pipeline",
                  node_id="my-node",
                  pipeline_run_id="",
              )
          )
          self.assertDictEqual(
              result,
              {
                  "y": [[y6], [y5], [y3, y4], [y1]],
                  "z": [[z4], [], [z2], [z1]],
              },
          )
  
      with self.subTest(
          "With execution_states=[COMPLETE, CACHED, FAILED]"
      ):
          result = (
              store_ext.get_live_output_artifacts_of_node_by_output_key(
                  self.store,
                  pipeline_id="my-pipeline",
                  node_id="my-node",
                  pipeline_run_id="",
                  execution_states=[
                      metadata_store_pb2.Execution.COMPLETE,
                      metadata_store_pb2.Execution.CACHED,
                      metadata_store_pb2.Execution.FAILED,
                  ],
              )
          )
          self.assertDictEqual(
              result,
              {
                  "y": [[y6], [y5], [y3, y4], [y1]],
                  "z": [[z4], [], [z2], [z1]],
              },
          )
  
file_path: tfx/tfx/orchestration/portable/mlmd/store_ext_test.py
imports: import time
  import tensorflow
  from tfx.orchestration.portable.mlmd import store_ext
  from tfx.utils import test_case_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 6 for calling the StoreExtTest obj's underlying member method code with example usage: obj.testGetLiveOutputArtifactsOfNodeByOutputKeyAsync() without return types