code: def rolling_range(
      artifacts,
      *,
      start_span_number: int = 0,
      num_spans: int = 1,
      skip_num_recent_spans: int = 0,
      keep_all_versions: bool = False,
      exclude_span_numbers: Sequence[int] = (),
      min_spans: Optional[int] = None,
  ):
      """Returns artifacts with spans in a rolling range.
  
      A rolling range covers the latest (largest) spans. It's calculated in the
      following order:
  
      1. Sort the artifacts by span in ascending order.
      2. Remove the last skip_num_recent_spans number of spans (removing the largest
         spans).
      3. Select the last num_spans number of spans (the remaining largest spans).
      4. Exclude the spans of exclude_span_numbers. Note that this exclusion
         happens last for backward compatibility. This can result in having less
         than num_spans spans, meaning the consumer component would be skipped due
         to lack of inputs. To avoid this, you would have to decrease min_spans.
  
      Pythonically, this range is equivalent to:
      sorted_spans[:-skip_num_recent_spans][-num_spans:]
  
      This resolver function is based on the span-version semantics, which only
      considers the latest version of each span. If you want to keep all versions,
      then set keep_all_versions=True. Input artifacts must have both "span" int
      property and "version" int property.
  
      Please note that the spans in exclude_span_numbers are excluded AFTER getting
      the latest spans.
  
      If there are less than min_spans unique spans present in the resolved
      artifacts, then the component execution will be skipped.
  
      Corresponds to RollingRange in TFX.
  
      Example usage:
  
        Consider 6 artifacts with:
          spans    = [1, 2, 3, 3, 7, 8]
          versions = [0, 0, 1, 0, 1, 2]
  
        rolling_range(
            start_span_number=3,
            num_spans=5,
            skip_num_recent_spans=1,
            keep_all_versions=True,
            exclude_span_numbers=[7],
            min_spans=1)
  
        spans 1 and 2 are removed because they are < start_span_number=3. The
        sorted unique spans are [3, 7, 8].
  
        span 8 is removed because skip_num_recent_spans=1, leaving spans [3, 7].
  
        Although num_spans=5, only two unique span numbers are available, 3 and 7,
        so both spans [3, 7] are kept.
  
        Because keep_all_versions=True, both artifacts with span=3 are kept.
  
        Because exclude_span_numbers=[7], the artifact with span=7 will not be
        kept, even though it is in the range.
  
        The artifacts that will be returned are:
          spans = [3, 3]
          versions = [1, 0]
  
        Note min_spans=1, so a SkipSignal will not be present in the compiled IR.
  
      Args:
        artifacts: The artifacts to filter.
        start_span_number: The smallest span number to keep, inclusive. Defaults to
          0.
        num_spans: The length of the range. If num_spans <= 0, then num_spans is set
          to the total number of unique spans.
        skip_num_recent_spans: Number of most recently available (largest) spans to
          skip. Defaults to 0.
        keep_all_versions: If true, all artifacts with spans in the range are kept.
          If false then if multiple artifacts have the same span, only the span with
          the latest version is kept. Defaults to False.
        exclude_span_numbers: The span numbers to exclude.
        min_spans: Minimum number of desired example spans in the range. If
          min_spans is None, it is set to num_spans.
  
      Returns:
        Artifacts with spans in the rolling range.
      """
      resolved_artifacts = ops.LatestSpan(
          artifacts,
          min_span=start_span_number,
          n=num_spans,
          skip_last_n=skip_num_recent_spans,
          keep_all_versions=keep_all_versions,
      )
      if exclude_span_numbers:
          resolved_artifacts = ops.ExcludeSpans(
              resolved_artifacts, denylist=exclude_span_numbers
          )
  
      if min_spans is None:
          logging.warning(
              "min_spans for rolling_range(...) was not set, so it is defaulting to "
              "num_spans = %s. If skip_num_recent_spans is set, this may delay "
              "the component triggering on the first run until sufficient Examples "
              "artifacts are available.",
              num_spans,
          )
          min_spans = num_spans
  
      return ops.SkipIfLessThanNSpans(resolved_artifacts, n=min_spans)
  
imports: from typing import Optional, Sequence, Union
  from absl import logging
  from tfx.dsl.input_resolution import resolver_function
  from tfx.dsl.input_resolution.ops import ops
  from tfx.types import artifact
  from tfx.types import channel
label_desc: Use image node_id 4 for calling a global function with example usage: rolling_range(artifacts) and returns: ops