code: def safe_patch(
      autologging_integration,
      destination,
      function_name,
      patch_function,
      manage_run=False,
      extra_tags=None,
  ):
      """Patches the specified `function_name` on the specified `destination` class for autologging
      purposes, preceding its implementation with an error-safe copy of the specified patch
      `patch_function` with the following error handling behavior:
          - Exceptions thrown from the underlying / original function
            (`<destination>.<function_name>`) are propagated to the caller.
          - Exceptions thrown from other parts of the patched implementation (`patch_function`)
            are caught and logged as warnings.
  
      Args:
          autologging_integration: The name of the autologging integration associated with the
              patch.
          destination: The Python class on which the patch is being defined.
          function_name: The name of the function to patch on the specified `destination` class.
          patch_function: The patched function code to apply. This is either a `PatchFunction`
              class definition or a function object. If it is a function object, the
              first argument should be reserved for an `original` method argument
              representing the underlying / original function. Subsequent arguments
              should be identical to those of the original function being patched.
          manage_run: If `True`, applies the `with_managed_run` wrapper to the specified
              `patch_function`, which automatically creates & terminates an MLflow
              active run during patch code execution if necessary. If `False`,
              does not apply the `with_managed_run` wrapper to the specified
              `patch_function`.
          extra_tags: A dictionary of extra tags to set on each managed run created by autologging.
      """
      from mlflow.utils.autologging_utils import (
          autologging_is_disabled,
          get_autologging_config,
      )
  
      if manage_run:
          tags = {MLFLOW_AUTOLOGGING: autologging_integration}
          if extra_tags:
              if isinstance(extra_tags, dict):
                  if MLFLOW_AUTOLOGGING in extra_tags:
                      extra_tags.pop(MLFLOW_AUTOLOGGING)
                      _logger.warning(
                          f"Tag `{MLFLOW_AUTOLOGGING}` is ignored as it is a reserved tag by MLflow "
                          f"autologging."
                      )
                  tags.update(extra_tags)
              else:
                  raise mlflow.exceptions.MlflowException.invalid_parameter_value(
                      f"Invalid `extra_tags` type: expecting dictionary, "
                      f"received `{type(extra_tags).__name__}`"
                  )
          patch_function = with_managed_run(
              autologging_integration,
              patch_function,
              tags=tags,
          )
  
      patch_is_class = inspect.isclass(patch_function)
      if patch_is_class:
          assert issubclass(patch_function, PatchFunction)
      else:
          assert callable(patch_function)
  
      original_fn = gorilla.get_original_attribute(
          destination, function_name, bypass_descriptor_protocol=False
      )
      # Retrieve raw attribute while bypassing the descriptor protocol
      raw_original_obj = gorilla.get_original_attribute(
          destination, function_name, bypass_descriptor_protocol=True
      )
      if original_fn != raw_original_obj:
          raise RuntimeError(
              f"Unsupport patch on {destination}.{function_name}"
          )
      elif isinstance(original_fn, property):
          is_property_method = True
  
          # For property decorated methods (a kind of method delegation), e.g.
          # class A:
          #   @property
          #   def f1(self):
          #     ...
          #     return delegated_f1
          #
          # suppose `a1` is an instance of class `A`,
          # `A.f1.fget` will get the original `def f1(self)` method,
          # and `A.f1.fget(a1)` will be equivalent to `a1.f1()` and
          # its return value will be the `delegated_f1` function.
          # So using the `property.fget` we can construct the (delegated) "original_fn"
          def original(self, *args, **kwargs):
              # the `original_fn.fget` will get the original method decorated by `property`
              # the `original_fn.fget(self)` will get the delegated function returned by the
              # property decorated method.
              bound_delegate_method = original_fn.fget(self)
              return bound_delegate_method(*args, **kwargs)
  
      else:
          original = original_fn
          is_property_method = False
  
      def safe_patch_function(*args, **kwargs):
          """
          A safe wrapper around the specified `patch_function` implementation designed to
          handle exceptions thrown during the execution of `patch_function`. This wrapper
          distinguishes exceptions thrown from the underlying / original function
          (`<destination>.<function_name>`) from exceptions thrown from other parts of
          `patch_function`. This distinction is made by passing an augmented version of the
          underlying / original function to `patch_function` that uses nonlocal state to track
          whether or not it has been executed and whether or not it threw an exception.
          Exceptions thrown from the underlying / original function are propagated to the caller,
          while exceptions thrown from other parts of `patch_function` are caught and logged as
          warnings.
          """
          # Reroute warnings encountered during the patch function implementation to an MLflow event
          # logger, and enforce silent mode if applicable (i.e. if the corresponding autologging
          # integration was called with `silent=True`), hiding MLflow event logging statements and
          # hiding all warnings in the autologging preamble and postamble (i.e. the code surrounding
          # the user's original / underlying ML function). Non-MLflow warnings are enabled during the
          # execution of the original / underlying ML function
          #
          # Note that we've opted *not* to apply this context manager as a decorator on
          # `safe_patch_function` because the context-manager-as-decorator pattern uses
          # `contextlib.ContextDecorator`, which creates generator expressions that cannot be pickled
          # during model serialization by ML frameworks such as scikit-learn
          is_silent_mode = get_autologging_config(
              autologging_integration, "silent", False
          )
          with set_mlflow_events_and_warnings_behavior_globally(
              # MLflow warnings emitted during autologging training sessions are likely not
              # actionable and result from the autologging implementation invoking another MLflow
              # API. Accordingly, we reroute these warnings to the MLflow event logger with level
              # WARNING For reference, see recommended warning and event logging behaviors from
              # https://docs.python.org/3/howto/logging.html#when-to-use-logging
              reroute_warnings=True,
              disable_event_logs=is_silent_mode,
              disable_warnings=is_silent_mode,
          ), set_non_mlflow_warnings_behavior_for_current_thread(
              # non-MLflow Warnings emitted during the autologging preamble (before the original /
              # underlying ML function is called) and postamble (after the original / underlying ML
              # function is called) are likely not actionable and result from the autologging
              # implementation invoking an API from a dependent library. Accordingly, we reroute
              # these warnings to the MLflow event logger with level WARNING. For reference, see
              # recommended warning and event logging behaviors from
              # https://docs.python.org/3/howto/logging.html#when-to-use-logging
              reroute_warnings=True,
              disable_warnings=is_silent_mode,
          ):
              if is_testing():
                  preexisting_run_for_testing = mlflow.active_run()
  
              # Whether or not to exclude autologged content from user-created fluent runs
              # (i.e. runs created manually via `mlflow.start_run()`)
              exclusive = get_autologging_config(
                  autologging_integration, "exclusive", False
              )
              user_created_fluent_run_is_active = (
                  mlflow.active_run()
                  and not _AutologgingSessionManager.active_session()
              )
              active_session_failed = (
                  _AutologgingSessionManager.active_session()
                  is not None
                  and _AutologgingSessionManager.active_session().state
                  == "failed"
              )
  
              if (
                  active_session_failed
                  or autologging_is_disabled(autologging_integration)
                  or (user_created_fluent_run_is_active and exclusive)
                  or mlflow.utils.autologging_utils._AUTOLOGGING_GLOBALLY_DISABLED
              ):
                  # If the autologging integration associated with this patch is disabled,
                  # or if the current autologging integration is in exclusive mode and a user-created
                  # fluent run is active, call the original function and return. Restore the original
                  # warning behavior during original function execution, since autologging is being
                  # skipped
                  with set_non_mlflow_warnings_behavior_for_current_thread(
                      disable_warnings=False,
                      reroute_warnings=False,
                  ):
                      return original(*args, **kwargs)
  
              # Whether or not the original / underlying function has been called during the
              # execution of patched code
              original_has_been_called = False
              # The value returned by the call to the original / underlying function during
              # the execution of patched code
              original_result = None
              # Whether or not an exception was raised from within the original / underlying function
              # during the execution of patched code
              failed_during_original = False
              # The active MLflow run (if any) associated with patch code execution
              patch_function_run_for_testing = None
              # The exception raised during executing patching function
              patch_function_exception = None
  
              def try_log_autologging_event(log_fn, *args):
                  try:
                      log_fn(*args)
                  except Exception as e:
                      _logger.debug(
                          "Failed to log autologging event via '%s'. Exception: %s",
                          log_fn,
                          e,
                      )
  
              def call_original_fn_with_event_logging(
                  original_fn, og_args, og_kwargs
              ):
                  try:
                      try_log_autologging_event(
                          AutologgingEventLogger.get_logger().log_original_function_start,
                          session,
                          destination,
                          function_name,
                          og_args,
                          og_kwargs,
                      )
                      original_fn_result = original_fn(
                          *og_args, **og_kwargs
                      )
  
                      try_log_autologging_event(
                          AutologgingEventLogger.get_logger().log_original_function_success,
                          session,
                          destination,
                          function_name,
                          og_args,
                          og_kwargs,
                      )
                      return original_fn_result
                  except Exception as original_fn_e:
                      try_log_autologging_event(
                          AutologgingEventLogger.get_logger().log_original_function_error,
                          session,
                          destination,
                          function_name,
                          og_args,
                          og_kwargs,
                          original_fn_e,
                      )
  
                      nonlocal failed_during_original
                      failed_during_original = True
                      raise
  
              with _AutologgingSessionManager.start_session(
                  autologging_integration
              ) as session:
                  try:
  
                      def call_original(*og_args, **og_kwargs):
                          def _original_fn(*_og_args, **_og_kwargs):
                              if is_testing():
                                  _validate_args(
                                      autologging_integration,
                                      function_name,
                                      args,
                                      kwargs,
                                      og_args,
                                      og_kwargs,
                                  )
                                  # By the time `original` is called by the patch implementation, we
                                  # assume that either: 1. the patch implementation has already
                                  # created an MLflow run or 2. the patch code will not create an
                                  # MLflow run during the current execution. Here, we capture a
                                  # reference to the active run, which we will use later on to
                                  # determine whether or not the patch implementation created
                                  # a run and perform validation if necessary
                                  nonlocal patch_function_run_for_testing
                                  patch_function_run_for_testing = (
                                      mlflow.active_run()
                                  )
  
                              nonlocal original_has_been_called
                              original_has_been_called = True
  
                              nonlocal original_result
                              # Show all non-MLflow warnings as normal (i.e. not as event logs)
                              # during original function execution, even if silent mode is enabled
                              # (`silent=True`), since these warnings originate from the ML framework
                              # or one of its dependencies and are likely relevant to the caller
                              with set_non_mlflow_warnings_behavior_for_current_thread(
                                  disable_warnings=False,
                                  reroute_warnings=False,
                              ):
                                  original_result = original(
                                      *_og_args, **_og_kwargs
                                  )
                                  return original_result
  
                          return call_original_fn_with_event_logging(
                              _original_fn, og_args, og_kwargs
                          )
  
                      # Apply the name, docstring, and signature of `original` to `call_original`.
                      # This is important because several autologging patch implementations inspect
                      # the signature of the `original` argument during execution
                      call_original = update_wrapper_extended(
                          call_original, original
                      )
  
                      try_log_autologging_event(
                          AutologgingEventLogger.get_logger().log_patch_function_start,
                          session,
                          destination,
                          function_name,
                          args,
                          kwargs,
                      )
  
                      if patch_is_class:
                          patch_function.call(
                              call_original, *args, **kwargs
                          )
                      else:
                          patch_function(call_original, *args, **kwargs)
  
                      session.state = "succeeded"
  
                      try_log_autologging_event(
                          AutologgingEventLogger.get_logger().log_patch_function_success,
                          session,
                          destination,
                          function_name,
                          args,
                          kwargs,
                      )
                  except Exception as e:
                      session.state = "failed"
                      patch_function_exception = e
                      # Exceptions thrown during execution of the original function should be
                      # propagated to the caller. Additionally, exceptions encountered during test
                      # mode should be reraised to detect bugs in autologging implementations
                      if failed_during_original or is_testing():
                          raise
  
                  if is_testing() and not preexisting_run_for_testing:
                      # If an MLflow run was created during the execution of patch code, verify that
                      # it is no longer active and that it contains expected autologging tags
                      assert (
                          not mlflow.active_run()
                      ), f"Autologging integration {autologging_integration} leaked an active run"
                      if patch_function_run_for_testing:
                          _validate_autologging_run(
                              autologging_integration,
                              patch_function_run_for_testing.info.run_id,
                          )
                  try:
                      if original_has_been_called:
                          return original_result
                      else:
                          return call_original_fn_with_event_logging(
                              original, args, kwargs
                          )
                  finally:
                      # If original function succeeds, but `patch_function_exception` exists,
                      # it represent patching code unexpected failure, so we call
                      # `log_patch_function_error` in this case.
                      # If original function failed, we don't call `log_patch_function_error`
                      # even if `patch_function_exception` exists, because original function failure
                      # means there's some error in user code (e.g. user provide wrong arguments)
                      if (
                          patch_function_exception is not None
                          and not failed_during_original
                      ):
                          try_log_autologging_event(
                              AutologgingEventLogger.get_logger().log_patch_function_error,
                              session,
                              destination,
                              function_name,
                              args,
                              kwargs,
                              patch_function_exception,
                          )
  
                          _logger.warning(
                              "Encountered unexpected error during %s autologging: %s",
                              autologging_integration,
                              patch_function_exception,
                          )
  
      if is_property_method:
          # Create a patched function (also property decorated)
          # like:
          #
          # class A:
          # @property
          # def get_bound_safe_patch_fn(self):
          #   original_fn.fget(self) # do availability check
          #   return bound_safe_patch_fn
          #
          # Suppose `a1` is instance of class A,
          # then `a1.get_bound_safe_patch_fn(*args, **kwargs)` will be equivalent to
          # `bound_safe_patch_fn(*args, **kwargs)`
          def get_bound_safe_patch_fn(self):
              # This `original_fn.fget` call is for availability check, if it raise error
              # then `hasattr(obj, {func_name})` will return False
              # so it mimic the original property behavior.
              original_fn.fget(self)
  
              def bound_safe_patch_fn(*args, **kwargs):
                  return safe_patch_function(self, *args, **kwargs)
  
              # Make bound method `instance.target_method` keep the same doc and signature.
              # Here return the bound safe patch function because user call property decorated
              # method will like `instance.property_decorated_method(...)`, and internally it will
              # call the `bound_safe_patch_fn`, the argument list don't include the `self` argument,
              # so return bound function here.
              return update_wrapper_extended(
                  bound_safe_patch_fn, original_fn.fget
              )
  
          # Make unbound method `class.target_method` keep the same doc and signature
          get_bound_safe_patch_fn = update_wrapper_extended(
              get_bound_safe_patch_fn, original_fn.fget
          )
          safe_patch_obj = property(get_bound_safe_patch_fn)
      else:
          safe_patch_obj = update_wrapper_extended(
              safe_patch_function, original
          )
  
      new_patch = _wrap_patch(
          destination, function_name, safe_patch_obj
      )
      _store_patch(autologging_integration, new_patch)
  
imports: import abc
  import functools
  import inspect
  import itertools
  import typing
  import uuid
  from abc import abstractmethod
  from contextlib import contextmanager
  import mlflow
  import mlflow.utils.autologging_utils
  from mlflow.entities.run_status import RunStatus
  from mlflow.environment_variables import _MLFLOW_AUTOLOGGING_TESTING
  from mlflow.tracking.client import MlflowClient
  from mlflow.utils import gorilla, is_iterator
  from mlflow.utils.autologging_utils import _logger
  from mlflow.utils.autologging_utils.events import AutologgingEventLogger
  from mlflow.utils.autologging_utils.logging_and_warnings import set_mlflow_events_and_warnings_behavior_globally, set_non_mlflow_warnings_behavior_for_current_thread
  from mlflow.utils.mlflow_tags import MLFLOW_AUTOLOGGING
label_desc: Use image node_id 7 for calling a global function with example usage: safe_patch(autologging_integration, destination, function_name, patch_function, manage_run, extra_tags) and returns: update_wrapper_extended, bound_delegate_method, safe_patch_function, original, original_fn_result, call_original_fn_with_event_logging, original_result, call_original_fn_with_event_logging, original_result