content: import collections
  import hashlib
  import inspect
  import threading
  from typing import (
      TYPE_CHECKING,
      Any,
      Callable,
      Coroutine,
      Sequence,
      Union,
  )
  
  from eth_typing import (
      ChainId,
  )
  from eth_utils import (
      is_boolean,
      is_bytes,
      is_dict,
      is_list_like,
      is_null,
      is_number,
      is_text,
      to_bytes,
  )
  
  from web3._utils.caching import (
      ASYNC_PROVIDER_TYPE,
      SYNC_PROVIDER_TYPE,
  )
  from web3._utils.caching.request_caching_validation import (
      UNCACHEABLE_BLOCK_IDS,
      always_cache_request,
      async_validate_from_block_id_in_params,
      async_validate_from_blockhash_in_params,
      async_validate_from_blocknum_in_result,
      validate_from_block_id_in_params,
      validate_from_blockhash_in_params,
      validate_from_blocknum_in_result,
  )
  from web3._utils.empty import (
      empty,
  )
  from web3._utils.rpc_abi import (
      RPC,
  )
  from web3.exceptions import (
      Web3TypeError,
  )
  from web3.types import (
      RPCEndpoint,
  )
  from web3.utils import (
      RequestCacheValidationThreshold,
  )
  
  if TYPE_CHECKING:
      from web3.providers import (  # noqa: F401
          AsyncBaseProvider,
          BaseProvider,
          PersistentConnectionProvider,
      )
      from web3.types import (  # noqa: F401
          AsyncMakeRequestFn,
          MakeRequestFn,
          RPCRequest,
          RPCResponse,
      )
  
  
  def generate_cache_key(value: Any) -> str:
      """
      Generates a cache key for the *args and **kwargs
      """
      if is_bytes(value):
          return hashlib.md5(value).hexdigest()
      elif is_text(value):
          return generate_cache_key(to_bytes(text=value))
      elif is_boolean(value) or is_null(value) or is_number(value):
          return generate_cache_key(repr(value))
      elif is_dict(value):
          return generate_cache_key((key, value[key]) for key in sorted(value.keys()))
      elif is_list_like(value) or isinstance(value, collections.abc.Generator):
          return generate_cache_key("".join(generate_cache_key(item) for item in value))
      else:
          raise Web3TypeError(
              f"Cannot generate cache key for value {value} of type {type(value)}"
          )
  
  
  class RequestInformation:
      def __init__(
          self,
          method: RPCEndpoint,
          params: Any,
          response_formatters: tuple[
              dict[str, Callable[..., Any]] | Callable[..., Any],
              Callable[..., Any],
              Callable[..., Any],
          ],
          subscription_id: str = None,
      ):
          self.method = method
          self.params = params
          self.response_formatters = response_formatters
          self.subscription_id = subscription_id
          self.middleware_response_processors: list[Callable[..., Any]] = []
  
  
  DEFAULT_VALIDATION_THRESHOLD = 60 * 60  # 1 hour
  
  CHAIN_VALIDATION_THRESHOLD_DEFAULTS: dict[
      int, RequestCacheValidationThreshold | int
  ] = {
      # Suggested safe values as defaults for each chain. Users can configure a different
      # value if desired.
      ChainId.ETH.value: RequestCacheValidationThreshold.FINALIZED,
      ChainId.ARB1.value: 7 * 24 * 60 * 60,  # 7 days
      ChainId.ZKSYNC.value: 60 * 60,  # 1 hour
      ChainId.OETH.value: 3 * 60,  # 3 minutes
      ChainId.MATIC.value: 30 * 60,  # 30 minutes
      ChainId.ZKEVM.value: 60 * 60,  # 1 hour
      ChainId.BASE.value: 7 * 24 * 60 * 60,  # 7 days
      ChainId.SCR.value: 60 * 60,  # 1 hour
      ChainId.GNO.value: 5 * 60,  # 5 minutes
      ChainId.AVAX.value: 2 * 60,  # 2 minutes
      ChainId.BNB.value: 2 * 60,  # 2 minutes
      ChainId.FTM.value: 60,  # 1 minute
  }
  
  
  def is_cacheable_request(
      provider: ASYNC_PROVIDER_TYPE | SYNC_PROVIDER_TYPE,
      method: RPCEndpoint,
      params: Any,
  ) -> bool:
      if not (provider.cache_allowed_requests and method in provider.cacheable_requests):
          return False
      elif method in BLOCKNUM_IN_PARAMS:
          block_id = params[0]
          if block_id in UNCACHEABLE_BLOCK_IDS:
              return False
      return True
  
  
  # -- request caching -- #
  
  ALWAYS_CACHE = {
      RPC.eth_chainId,
      RPC.web3_clientVersion,
      RPC.net_version,
  }
  BLOCKNUM_IN_PARAMS = {
      RPC.eth_getBlockByNumber,
      RPC.eth_getRawTransactionByBlockNumberAndIndex,
      RPC.eth_getBlockTransactionCountByNumber,
      RPC.eth_getUncleByBlockNumberAndIndex,
      RPC.eth_getUncleCountByBlockNumber,
  }
  BLOCK_IN_RESULT = {
      RPC.eth_getBlockByHash,
      RPC.eth_getTransactionByHash,
      RPC.eth_getTransactionByBlockNumberAndIndex,
      RPC.eth_getTransactionByBlockHashAndIndex,
      RPC.eth_getBlockTransactionCountByHash,
  }
  BLOCKHASH_IN_PARAMS = {
      RPC.eth_getRawTransactionByBlockHashAndIndex,
      RPC.eth_getUncleByBlockHashAndIndex,
      RPC.eth_getUncleCountByBlockHash,
  }
  
  INTERNAL_VALIDATION_MAP: dict[
      RPCEndpoint,
      Callable[
          [SYNC_PROVIDER_TYPE, Sequence[Any], dict[str, Any]],
          bool,
      ],
  ] = {
      **{endpoint: always_cache_request for endpoint in ALWAYS_CACHE},
      **{endpoint: validate_from_block_id_in_params for endpoint in BLOCKNUM_IN_PARAMS},
      **{endpoint: validate_from_blocknum_in_result for endpoint in BLOCK_IN_RESULT},
      **{endpoint: validate_from_blockhash_in_params for endpoint in BLOCKHASH_IN_PARAMS},
  }
  CACHEABLE_REQUESTS = tuple(INTERNAL_VALIDATION_MAP.keys())
  
  
  def set_threshold_if_empty(provider: SYNC_PROVIDER_TYPE) -> None:
      current_threshold = provider.request_cache_validation_threshold
  
      if current_threshold is empty or isinstance(
          current_threshold, RequestCacheValidationThreshold
      ):
          cache_allowed_requests = provider.cache_allowed_requests
          try:
              # turn off momentarily to avoid recursion
              provider.cache_allowed_requests = False
              chain_id_result = provider.make_request(RPCEndpoint("eth_chainId"), [])[
                  "result"
              ]
              chain_id = int(chain_id_result, 16)
  
              if current_threshold is empty:
                  provider.request_cache_validation_threshold = (
                      CHAIN_VALIDATION_THRESHOLD_DEFAULTS.get(
                          chain_id, DEFAULT_VALIDATION_THRESHOLD
                      )
                  )
          except Exception:
              provider.request_cache_validation_threshold = DEFAULT_VALIDATION_THRESHOLD
          finally:
              provider.cache_allowed_requests = cache_allowed_requests
  
  
  def _should_cache_response(
      provider: SYNC_PROVIDER_TYPE,
      method: RPCEndpoint,
      params: Sequence[Any],
      response: "RPCResponse",
  ) -> bool:
      result = response.get("result", None)
      if "error" in response or is_null(result):
          return False
  
      set_threshold_if_empty(provider)
      if (
          method in INTERNAL_VALIDATION_MAP
          and provider.request_cache_validation_threshold is not None
      ):
          return INTERNAL_VALIDATION_MAP[method](provider, params, result)
      return True
  
  
  def handle_request_caching(
      func: Callable[[SYNC_PROVIDER_TYPE, RPCEndpoint, Any], "RPCResponse"],
  ) -> Callable[..., "RPCResponse"]:
      def wrapper(
          provider: SYNC_PROVIDER_TYPE, method: RPCEndpoint, params: Any
      ) -> "RPCResponse":
          if is_cacheable_request(provider, method, params):
              request_cache = provider._request_cache
              cache_key = generate_cache_key(
                  f"{threading.get_ident()}:{(method, params)}"
              )
              cache_result = request_cache.get_cache_entry(cache_key)
              if cache_result is not None:
                  return cache_result
              else:
                  response = func(provider, method, params)
                  if _should_cache_response(provider, method, params, response):
                      with provider._request_cache_lock:
                          request_cache.cache(cache_key, response)
                  return response
          else:
              return func(provider, method, params)
  
      # save a reference to the decorator on the wrapped function
      wrapper._decorator = handle_request_caching  # type: ignore
      return wrapper
  
  
  # -- async -- #
  
  ASYNC_VALIDATOR_TYPE = Callable[
      ["AsyncBaseProvider", Sequence[Any], dict[str, Any]],
      Union[bool, Coroutine[Any, Any, bool]],
  ]
  
  ASYNC_INTERNAL_VALIDATION_MAP: dict[RPCEndpoint, ASYNC_VALIDATOR_TYPE] = {
      **{endpoint: always_cache_request for endpoint in ALWAYS_CACHE},
      **{
          endpoint: async_validate_from_block_id_in_params
          for endpoint in BLOCKNUM_IN_PARAMS
      },
      **{
          endpoint: async_validate_from_blocknum_in_result for endpoint in BLOCK_IN_RESULT
      },
      **{
          endpoint: async_validate_from_blockhash_in_params
          for endpoint in BLOCKHASH_IN_PARAMS
      },
  }
  
  
  async def async_set_threshold_if_empty(provider: ASYNC_PROVIDER_TYPE) -> None:
      current_threshold = provider.request_cache_validation_threshold
  
      if current_threshold is empty or isinstance(
          current_threshold, RequestCacheValidationThreshold
      ):
          cache_allowed_requests = provider.cache_allowed_requests
          try:
              # turn off momentarily to avoid recursion
              provider.cache_allowed_requests = False
              chain_id_result = await provider.make_request(
                  RPCEndpoint("eth_chainId"), []
              )
              chain_id = int(chain_id_result["result"], 16)
  
              if current_threshold is empty:
                  provider.request_cache_validation_threshold = (
                      CHAIN_VALIDATION_THRESHOLD_DEFAULTS.get(
                          chain_id, DEFAULT_VALIDATION_THRESHOLD
                      )
                  )
          except Exception:
              provider.request_cache_validation_threshold = DEFAULT_VALIDATION_THRESHOLD
          finally:
              provider.cache_allowed_requests = cache_allowed_requests
  
  
  async def _async_should_cache_response(
      provider: ASYNC_PROVIDER_TYPE,
      method: RPCEndpoint,
      params: Sequence[Any],
      response: "RPCResponse",
  ) -> bool:
      result = response.get("result", None)
      if "error" in response or is_null(result):
          return False
  
      await async_set_threshold_if_empty(provider)
      if (
          method in ASYNC_INTERNAL_VALIDATION_MAP
          and provider.request_cache_validation_threshold is not None
      ):
          cache_validator = ASYNC_INTERNAL_VALIDATION_MAP[method]
          return (
              await cache_validator(provider, params, result)
              if inspect.iscoroutinefunction(cache_validator)
              else cache_validator(provider, params, result)
          )
      return True
  
  
  def async_handle_request_caching(
      func: Callable[
          [ASYNC_PROVIDER_TYPE, RPCEndpoint, Any], Coroutine[Any, Any, "RPCResponse"]
      ],
  ) -> Callable[..., Coroutine[Any, Any, "RPCResponse"]]:
      async def wrapper(
          provider: ASYNC_PROVIDER_TYPE, method: RPCEndpoint, params: Any
      ) -> "RPCResponse":
          if is_cacheable_request(provider, method, params):
              request_cache = provider._request_cache
              cache_key = generate_cache_key(
                  f"{threading.get_ident()}:{(method, params)}"
              )
              cache_result = request_cache.get_cache_entry(cache_key)
              if cache_result is not None:
                  return cache_result
              else:
                  response = await func(provider, method, params)
                  if await _async_should_cache_response(
                      provider, method, params, response
                  ):
                      async with provider._request_cache_lock:
                          request_cache.cache(cache_key, response)
                  return response
          else:
              return await func(provider, method, params)
  
      # save a reference to the decorator on the wrapped function
      wrapper._decorator = async_handle_request_caching  # type: ignore
      return wrapper
  
  
  def async_handle_send_caching(
      func: Callable[
          [ASYNC_PROVIDER_TYPE, RPCEndpoint, Any],
          Coroutine[Any, Any, "RPCRequest"],
      ],
  ) -> Callable[..., Coroutine[Any, Any, "RPCRequest"]]:
      async def wrapper(
          provider: ASYNC_PROVIDER_TYPE, method: RPCEndpoint, params: Any
      ) -> "RPCRequest":
          if is_cacheable_request(provider, method, params):
              request_cache = provider._request_cache
              cache_key = generate_cache_key(
                  f"{threading.get_ident()}:{(method, params)}"
              )
              cached_response = request_cache.get_cache_entry(cache_key)
              if cached_response is not None:
                  # The request data isn't used, this just prevents a cached request from
                  # being sent - return an empty request object
                  return {"id": -1, "method": RPCEndpoint(""), "params": []}
          return await func(provider, method, params)
  
      # save a reference to the decorator on the wrapped function
      wrapper._decorator = async_handle_send_caching  # type: ignore
      return wrapper
  
  
  def async_handle_recv_caching(
      func: Callable[
          ["PersistentConnectionProvider", "RPCRequest"],
          Coroutine[Any, Any, "RPCResponse"],
      ],
  ) -> Callable[..., Coroutine[Any, Any, "RPCResponse"]]:
      async def wrapper(
          provider: "PersistentConnectionProvider",
          rpc_request: "RPCRequest",
      ) -> "RPCResponse":
          method = rpc_request["method"]
          params = rpc_request["params"]
          if is_cacheable_request(provider, method, params):
              request_cache = provider._request_cache
              cache_key = generate_cache_key(
                  f"{threading.get_ident()}:{(method, params)}"
              )
              cache_result = request_cache.get_cache_entry(cache_key)
              if cache_result is not None:
                  return cache_result
              else:
                  response = await func(provider, rpc_request)
                  if await _async_should_cache_response(
                      provider, method, params, response
                  ):
                      async with provider._request_cache_lock:
                          request_cache.cache(cache_key, response)
                  return response
          else:
              return await func(provider, rpc_request)
  
      # save a reference to the decorator on the wrapped function
      wrapper._decorator = async_handle_recv_caching  # type: ignore
      return wrapper
  