class_docstr: Plan to backfill a dataset for a given revision.
  
  The plan is composed of tasks to delete and create jobs.
  
  Args:
      dataset: dataset name
      revision: revision to backfill
      priority: priority of the jobs to create
      only_first_processing_steps: if True, only the first processing steps are backfilled
      processing_graph: processing graph
class_docstr_tok: ["Plan","to","backfill","a","dataset","for","a","given","revision",".","The","plan","is","composed","of","tasks","to","delete","and","create","jobs",".","Args",":","dataset",":","dataset","name","revision",":","revision","to","backfill","priority",":","priority","of","the","jobs","to","create","only_first_processing_steps",":","if","True",",","only","the","first","processing","steps","are","backfilled","processing_graph",":","processing","graph"]
docstr: Get the artifact states for a step.
  
  Args:
      processing_step (ProcessingStep): the processing step
      config (str, optional): if not None, and step input type is config or split, only return the artifact
        states for this config
      split (str, optional): if not None, and step input type is split, only return the artifact states for
        this split (config must be specified)
  
  Returns:
      the artifact states for the step
docstr_tok: ["Get","the","artifact","states","for","a","step",".","Args",":","processing_step","(","ProcessingStep",")",":","the","processing","step","config","(","str",",","optional",")",":","if","not","None",",","and","step","input","type","is","config","or","split",",","only","return","the","artifact","states","for","this","config","split","(","str",",","optional",")",":","if","not","None",",","and","step","input","type","is","split",",","only","return","the","artifact","states","for","this","split","(","config","must","be","specified",")","Returns",":","the","artifact","states","for","the","step"]
code: def _get_artifact_states_for_step(
      self,
      processing_step: ProcessingStep,
      config: Optional[str] = None,
      split: Optional[str] = None,
  ) -> list[ArtifactState]:
      """Get the artifact states for a step.
  
      Args:
          processing_step (ProcessingStep): the processing step
          config (str, optional): if not None, and step input type is config or split, only return the artifact
            states for this config
          split (str, optional): if not None, and step input type is split, only return the artifact states for
            this split (config must be specified)
  
      Returns:
          the artifact states for the step
      """
      if processing_step.input_type == "dataset":
          artifact_states = [
              self.dataset_state.artifact_state_by_step[
                  processing_step.name
              ]
          ]
      elif processing_step.input_type == "config":
          if config is None:
              artifact_states = [
                  config_state.artifact_state_by_step[
                      processing_step.name
                  ]
                  for config_state in self.dataset_state.config_states
              ]
          else:
              artifact_states = [
                  config_state.artifact_state_by_step[
                      processing_step.name
                  ]
                  for config_state in self.dataset_state.config_states
                  if config_state.config == config
              ]
      elif processing_step.input_type == "split":
          if config is None:
              artifact_states = [
                  split_state.artifact_state_by_step[
                      processing_step.name
                  ]
                  for config_state in self.dataset_state.config_states
                  for split_state in config_state.split_states
              ]
          elif split is None:
              artifact_states = [
                  split_state.artifact_state_by_step[
                      processing_step.name
                  ]
                  for config_state in self.dataset_state.config_states
                  if config_state.config == config
                  for split_state in config_state.split_states
              ]
          else:
              artifact_states = [
                  split_state.artifact_state_by_step[
                      processing_step.name
                  ]
                  for config_state in self.dataset_state.config_states
                  if config_state.config == config
                  for split_state in config_state.split_states
                  if split_state.split == split
              ]
      else:
          raise ValueError(
              f"Invalid input type: {processing_step.input_type}"
          )
      artifact_states_ids = {
          artifact_state.id for artifact_state in artifact_states
      }
      if len(artifact_states_ids) != len(artifact_states):
          raise ValueError(
              f"Duplicate artifact states for processing_step {processing_step}"
          )
      return artifact_states
  
imports: import logging
  from abc import ABC, abstractmethod
  from dataclasses import dataclass, field
  from http import HTTPStatus
  from typing import Optional, Union
  import pandas
  from libcommon.constants import CONFIG_INFO_KINDS, CONFIG_SPLIT_NAMES_KINDS, DATASET_CONFIG_NAMES_KINDS, DEFAULT_DIFFICULTY_MAX, DIFFICULTY_BONUS_BY_FAILED_RUNS
  from libcommon.processing_graph import ProcessingGraph, ProcessingStep, ProcessingStepDoesNotExist, processing_graph
  from libcommon.prometheus import StepProfiler
  from libcommon.queue import Queue
  from libcommon.simple_cache import CacheEntryDoesNotExistError, delete_dataset_responses, fetch_names, get_best_response, get_cache_entries_df, get_response_metadata, upsert_response_params
  from libcommon.state import ArtifactState, DatasetState, FirstStepsDatasetState
  from libcommon.storage_client import StorageClient
  from libcommon.utils import JobInfo, JobResult, Priority
label_desc: Use image node_id 2 for calling the DatasetBackfillPlan obj's underlying member method code with example usage: obj._get_artifact_states_for_step(processing_step, config, split) and returns: artifact_states