def METHOD_NAME(
    pipeline: beam.Pipeline,
    cache_key: str,
    cache_manager: cache.CacheManager,
    element_type: Optional[type] = None,
    source_label: Optional[str] = None,
    unreify_label: Optional[str] = None) -> beam.pvalue.PCollection:
  """Reads from cache and unreifies elements from windowed values.
  pipeline: The pipeline that's reading from the cache.
  cache_key: The key of the cache.
  cache_manager: The cache manager to manage the cache.
  element_type: (optional) The element type of the PCollection's elements.
  source_label: (optional) A transform label for the cache-reading transform.
  unreify_label: (optional) A transform label for the Unreify transform.
  """
  if not source_label:
    source_label = '{}{}'.format(READ_CACHE, cache_key)
  if not unreify_label:
    unreify_label = '{}{}{}'.format('UnreifyAfter_', READ_CACHE, cache_key)
  read_cache = pipeline | source_label >> cache.ReadCache(
      cache_manager, cache_key)
  if element_type:
    # If the PCollection is schema-aware, explicitly sets the output types.
    return read_cache | unreify_label >> beam.ParDo(
        Unreify()).with_output_types(element_type)
  return read_cache | unreify_label >> beam.ParDo(Unreify())