code: def _filter_by_artifact_query(
      store: mlmd.MetadataStore,
      artifacts: Sequence[metadata_store_pb2.Artifact],
      artifact_query: pipeline_pb2.InputSpec.Channel.ArtifactQuery,
      live_only: bool = True,
  ) -> Tuple[
      List[metadata_store_pb2.Artifact], metadata_store_pb2.ArtifactType
  ]:
      """Filters initial artifacts with given ArtifactQuery."""
      predicates = []
      if not artifact_query.type or not artifact_query.type.name:
          raise ValueError("ArtifactQuery.type.name should be set.")
      artifact_type = artifact_query.type
      if not artifact_type.id:
          try:
              artifact_type = store.get_artifact_type(
                  artifact_query.type.name
              )
          except errors.NotFoundError:
              # If artifac type does not exist, it means no artifacts satisfy the
              # artifact query. Returns empty.
              return [], artifact_type
      predicates.append(lambda a: a.type_id == artifact_type.id)
      if artifact_query.property_predicate:
          logging.warning(
              "ArtifactQuery.property_predicate is not supported."
          )
      if live_only:
          predicates.append(
              lambda a: a.state == metadata_store_pb2.Artifact.LIVE
          )
      return (
          [a for a in artifacts if all(p(a) for p in predicates)],
          artifact_type,
      )
  
file_path: tfx/tfx/orchestration/portable/input_resolution/channel_resolver.py
imports: from typing import List, Optional, Sequence, Tuple
  from absl import logging
  from tfx import types
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable.mlmd import event_lib
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.types import artifact_utils
  import ml_metadata
  from ml_metadata import errors
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 4 for calling a global function with example usage: _filter_by_artifact_query(store, artifacts, artifact_query, live_only) and returns: unknown,  artifact_type, list,  artifact_type