code: def internal_api_call(func: Callable[PS, RT]) -> Callable[PS, RT]:
      """
      Allow methods to be executed in database isolation mode.
  
      If [core]database_access_isolation is true then such method are not executed locally,
      but instead RPC call is made to Database API (aka Internal API). This makes some components
      decouple from direct Airflow database access.
      Each decorated method must be present in METHODS list in airflow.api_internal.endpoints.rpc_api_endpoint.
      Only static methods can be decorated. This decorator must be before "provide_session".
  
      See [AIP-44](https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-44+Airflow+Internal+API)
      for more information .
      """
      headers = {
          "Content-Type": "application/json",
      }
  
      def make_jsonrpc_request(
          method_name: str, params_json: str
      ) -> bytes:
          data = {
              "jsonrpc": "2.0",
              "method": method_name,
              "params": params_json,
          }
          internal_api_endpoint = (
              InternalApiConfig.get_internal_api_endpoint()
          )
          response = requests.post(
              url=internal_api_endpoint,
              data=json.dumps(data),
              headers=headers,
          )
          if response.status_code != 200:
              raise AirflowException(
                  f"Got {response.status_code}:{response.reason} when sending the internal api request."
              )
          return response.content
  
      @wraps(func)
      def wrapper(*args, **kwargs):
          use_internal_api = InternalApiConfig.get_use_internal_api()
          if not use_internal_api:
              return func(*args, **kwargs)
  
          from airflow.serialization.serialized_objects import (
              BaseSerialization,
          )  # avoid circular import
  
          bound = inspect.signature(func).bind(*args, **kwargs)
          arguments_dict = dict(bound.arguments)
          if "session" in arguments_dict:
              del arguments_dict["session"]
          if "cls" in arguments_dict:  # used by @classmethod
              del arguments_dict["cls"]
  
          args_json = json.dumps(
              BaseSerialization.serialize(
                  arguments_dict, use_pydantic_models=True
              ),
              default=BaseSerialization.serialize,
          )
          method_name = f"{func.__module__}.{func.__qualname__}"
          result = make_jsonrpc_request(method_name, args_json)
          if result is None or result == b"":
              return None
          return BaseSerialization.deserialize(
              json.loads(result), use_pydantic_models=True
          )
  
      return wrapper
  
file_path: airflow/airflow/api_internal/internal_api_call.py
imports: from __future__ import annotations
  import inspect
  import json
  from functools import wraps
  from typing import Callable, TypeVar
  import requests
  from airflow.configuration import conf
  from airflow.exceptions import AirflowConfigException, AirflowException
  from airflow.settings import _ENABLE_AIP_44
  from airflow.typing_compat import ParamSpec
label_desc: Use image node_id 1 for calling a global function with example usage: internal_api_call(func) and returns: wrapper, response, BaseSerialization, func, None