class_docstr: Class abstracting management of custom extractors.
code: def extract_metadata(
      self, dagrun, task, complete: bool = False, task_instance=None
  ) -> OperatorLineage:
      extractor = self._get_extractor(task)
      task_info = (
          f"task_type={task.task_type} "
          f"airflow_dag_id={task.dag_id} "
          f"task_id={task.task_id} "
          f"airflow_run_id={dagrun.run_id} "
      )
  
      if extractor:
          # Extracting advanced metadata is only possible when extractor for particular operator
          # is defined. Without it, we can't extract any input or output data.
          try:
              self.log.debug(
                  "Using extractor %s %s",
                  extractor.__class__.__name__,
                  str(task_info),
              )
              if complete:
                  task_metadata = extractor.extract_on_complete(
                      task_instance
                  )
              else:
                  task_metadata = extractor.extract()
  
              self.log.debug(
                  "Found task metadata for operation %s: %s",
                  task.task_id,
                  str(task_metadata),
              )
              task_metadata = self.validate_task_metadata(task_metadata)
              if task_metadata:
                  if (not task_metadata.inputs) and (
                      not task_metadata.outputs
                  ):
                      self.extract_inlets_and_outlets(
                          task_metadata, task.inlets, task.outlets
                      )
  
                  return task_metadata
  
          except Exception as e:
              self.log.warning(
                  "Failed to extract metadata using found extractor %s - %s %s",
                  extractor,
                  e,
                  task_info,
              )
      else:
          self.log.debug("Unable to find an extractor %s", task_info)
  
          # Only include the unkonwnSourceAttribute facet if there is no extractor
          task_metadata = OperatorLineage(
              run_facets={
                  "unknownSourceAttribute": UnknownOperatorAttributeRunFacet(
                      unknownItems=[
                          UnknownOperatorInstance(
                              name=task.task_type,
                              properties=get_filtered_unknown_operator_keys(
                                  task
                              ),
                          )
                      ]
                  )
              },
          )
          inlets = task.get_inlet_defs()
          outlets = task.get_outlet_defs()
          self.extract_inlets_and_outlets(
              task_metadata, inlets, outlets
          )
          return task_metadata
  
      return OperatorLineage()
  
imports: from __future__ import annotations
  import os
  from contextlib import suppress
  from typing import TYPE_CHECKING, Iterator
  from airflow.configuration import conf
  from airflow.providers.openlineage.extractors import BaseExtractor, OperatorLineage
  from airflow.providers.openlineage.extractors.base import DefaultExtractor
  from airflow.providers.openlineage.extractors.bash import BashExtractor
  from airflow.providers.openlineage.extractors.python import PythonExtractor
  from airflow.providers.openlineage.plugins.facets import UnknownOperatorAttributeRunFacet, UnknownOperatorInstance
  from airflow.providers.openlineage.utils.utils import get_filtered_unknown_operator_keys
  from airflow.utils.log.logging_mixin import LoggingMixin
  from airflow.utils.module_loading import import_string
label_desc: Use image node_id 3 for calling the ExtractorManager obj's underlying member method code with example usage: obj.extract_metadata(dagrun, task, complete, task_instance) and returns: OperatorLineage, task_metadata, task_metadata