content: import logging
  import time
  from typing import (
      TYPE_CHECKING,
      Any,
      Iterable,
      cast,
  )
  
  from eth_typing import (
      URI,
  )
  from eth_utils import (
      combomethod,
      to_dict,
  )
  import requests
  
  from web3._utils.empty import (
      Empty,
      empty,
  )
  from web3._utils.http import (
      construct_user_agent,
  )
  from web3.types import (
      RPCEndpoint,
      RPCResponse,
  )
  
  from ..._utils.batching import (
      sort_batch_response_by_response_ids,
  )
  from ..._utils.caching import (
      handle_request_caching,
  )
  from ..._utils.http_session_manager import (
      HTTPSessionManager,
  )
  from ..base import (
      JSONBaseProvider,
  )
  from .utils import (
      ExceptionRetryConfiguration,
      check_if_retry_on_failure,
  )
  
  if TYPE_CHECKING:
      from web3.middleware.base import (  # noqa: F401
          Middleware,
      )
  
  
  class HTTPProvider(JSONBaseProvider):
      logger = logging.getLogger("web3.providers.HTTPProvider")
      endpoint_uri = None
      _request_kwargs = None
  
      def __init__(
          self,
          endpoint_uri: URI | str | None = None,
          request_kwargs: Any | None = None,
          session: Any | None = None,
          exception_retry_configuration: None
          | (ExceptionRetryConfiguration | Empty) = empty,
          **kwargs: Any,
      ) -> None:
          super().__init__(**kwargs)
          self._request_session_manager = HTTPSessionManager()
  
          if endpoint_uri is None:
              self.endpoint_uri = (
                  self._request_session_manager.get_default_http_endpoint()
              )
          else:
              self.endpoint_uri = URI(endpoint_uri)
  
          self._request_kwargs = request_kwargs or {}
          self._exception_retry_configuration = exception_retry_configuration
  
          if session:
              self._request_session_manager.cache_and_return_session(
                  self.endpoint_uri, session
              )
  
      def __str__(self) -> str:
          return f"RPC connection {self.endpoint_uri}"
  
      @property
      def exception_retry_configuration(self) -> ExceptionRetryConfiguration:
          if isinstance(self._exception_retry_configuration, Empty):
              self._exception_retry_configuration = ExceptionRetryConfiguration(
                  errors=(
                      ConnectionError,
                      requests.HTTPError,
                      requests.Timeout,
                  )
              )
          return self._exception_retry_configuration
  
      @exception_retry_configuration.setter
      def exception_retry_configuration(
          self, value: ExceptionRetryConfiguration | Empty
      ) -> None:
          self._exception_retry_configuration = value
  
      @to_dict
      def get_request_kwargs(self) -> Iterable[tuple[str, Any]]:
          if "headers" not in self._request_kwargs:
              yield "headers", self.get_request_headers()
          yield from self._request_kwargs.items()
  
      @combomethod
      def get_request_headers(cls) -> dict[str, str]:
          if isinstance(cls, HTTPProvider):
              cls_name = cls.__class__.__name__
          else:
              cls_name = cls.__name__
  
          module = cls.__module__
  
          return {
              "Content-Type": "application/json",
              "User-Agent": construct_user_agent(module, cls_name),
          }
  
      def _make_request(self, method: RPCEndpoint, request_data: bytes) -> bytes:
          """
          If exception_retry_configuration is set, retry on failure; otherwise, make
          the request without retrying.
          """
          if (
              self.exception_retry_configuration is not None
              and check_if_retry_on_failure(
                  method, self.exception_retry_configuration.method_allowlist
              )
          ):
              for i in range(self.exception_retry_configuration.retries):
                  try:
                      return self._request_session_manager.make_post_request(
                          self.endpoint_uri, request_data, **self.get_request_kwargs()
                      )
                  except tuple(self.exception_retry_configuration.errors) as e:
                      if i < self.exception_retry_configuration.retries - 1:
                          time.sleep(
                              self.exception_retry_configuration.backoff_factor * 2**i
                          )
                          continue
                      else:
                          raise e
              return None
          else:
              return self._request_session_manager.make_post_request(
                  self.endpoint_uri, request_data, **self.get_request_kwargs()
              )
  
      @handle_request_caching
      def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
          self.logger.debug(
              "Making request HTTP. URI: %s, Method: %s", self.endpoint_uri, method
          )
          request_data = self.encode_rpc_request(method, params)
          raw_response = self._make_request(method, request_data)
          response = self.decode_rpc_response(raw_response)
          self.logger.debug(
              "Getting response HTTP. URI: %s, Method: %s, Response: %s",
              self.endpoint_uri,
              method,
              response,
          )
          return response
  
      def make_batch_request(
          self, batch_requests: list[tuple[RPCEndpoint, Any]]
      ) -> list[RPCResponse] | RPCResponse:
          self.logger.debug("Making batch request HTTP, uri: `%s`", self.endpoint_uri)
          request_data = self.encode_batch_rpc_request(batch_requests)
          raw_response = self._request_session_manager.make_post_request(
              self.endpoint_uri, request_data, **self.get_request_kwargs()
          )
          self.logger.debug("Received batch response HTTP.")
          response = self.decode_rpc_response(raw_response)
          if not isinstance(response, list):
              # RPC errors return only one response with the error object
              return response
          return sort_batch_response_by_response_ids(
              cast(list[RPCResponse], sort_batch_response_by_response_ids(response))
          )
  