ds_protocol_http_py_lib.utils.http.provider
===========================================

.. py:module:: ds_protocol_http_py_lib.utils.http.provider

.. autoapi-nested-parse::

   **File:** ``provider.py``
   **Region:** ``ds_protocol_http_py_lib/utils/http/provider``

   HTTP Provider

   This module implements a synchronous HTTP client with:
   - requests.Session + urllib3.Retry (429/5xx, backoff, Retry-After)
   - optional TokenBucket for simple RPS throttling
   - context-managed lifetime
   - tiny API: request/get/post/close

   .. rubric:: Example

   >>> with Http() as client:
   ...     response = client.get("https://api.example.com/data")
   ...     data = response.json()



Attributes
----------

.. autoapisummary::

   ds_protocol_http_py_lib.utils.http.provider.logger


Classes
-------

.. autoapisummary::

   ds_protocol_http_py_lib.utils.http.provider.Http


Module Contents
---------------

.. py:data:: logger

.. py:class:: Http(*, config: ds_protocol_http_py_lib.utils.http.config.HttpConfig | None = None, bucket: ds_protocol_http_py_lib.utils.http.token_bucket.TokenBucket | None = None, session: requests.Session | None = None)

   Minimal synchronous HTTP client with:
     - requests.Session + urllib3.Retry (429/5xx, backoff, Retry-After)
     - optional TokenBucket for simple RPS throttling
     - context-managed lifetime
     - tiny API: request/get/post/close


   .. py:attribute:: _cfg


   .. py:attribute:: _bucket


   .. py:attribute:: _session


   .. py:method:: _build_session() -> requests.Session

      Build the session.
      :returns: The session.
      :rtype: requests.Session



   .. py:method:: _response_info(response: requests.Response) -> dict[str, Any]

      Get information about a response.
      Extracts safe metadata only (no request/response bodies).

      :param response: The HTTP response object to extract info from.

      :returns: Dictionary containing the response information.
      :rtype: dict[str, Any]



   .. py:property:: session
      :type: requests.Session


      Get the underlying requests session for direct use.
      Allows direct access to session properties like headers.

      :returns: The requests session.
      :rtype: requests.Session

      .. rubric:: Example

      >>> http = Http()
      >>> http.session.headers.update({"Authorization": "Bearer token"})


   .. py:method:: __enter__() -> Http


   .. py:method:: __exit__(exc_type: Any, exc: Any, tb: Any) -> None


   .. py:method:: close() -> None


   .. py:method:: request(method: str, url: str, **kwargs: Any) -> requests.Response

      Send an HTTP request with rate limiting, retry logic, and comprehensive logging.

      :param method: HTTP method (GET, POST, PUT, DELETE, etc.).
      :param url: Target URL for the request.
      :param \*\*kwargs: Additional keyword arguments passed to requests (timeout, headers, data, etc.).

      :returns: The HTTP response object.
      :rtype: requests.Response

      :raises requests.HTTPError: If the response status code indicates an error.
      :raises requests.RequestException: For other request-related errors.

      .. rubric:: Example

      >>> with Http() as client:
      ...     response = client.request('GET', 'https://api.example.com/data', timeout=30)
      ...     data = response.json()



   .. py:method:: get(url: str, **kwargs: Any) -> requests.Response

      Send a GET request with enhanced logging.

      :param url: Target URL for the GET request.
      :param \*\*kwargs: Additional keyword arguments passed to requests.

      :returns: The HTTP response object.
      :rtype: requests.Response

      .. rubric:: Example

      >>> with Http() as client:
      ...     response = client.get("https://api.example.com/data")



   .. py:method:: post(url: str, **kwargs: Any) -> requests.Response

      Send a POST request with enhanced logging.

      :param url: Target URL for the POST request.
      :param \*\*kwargs: Additional keyword arguments passed to requests.

      :returns: The HTTP response object.
      :rtype: requests.Response

      .. rubric:: Example

      >>> with Http() as client:
      ...     response = client.post("https://api.example.com/data", json={"key": "value"})



   .. py:method:: put(url: str, **kwargs: Any) -> requests.Response

      Send a PUT request with enhanced logging.

      :param url: Target URL for the PUT request.
      :param \*\*kwargs: Additional keyword arguments passed to requests.

      :returns: The HTTP response object.
      :rtype: requests.Response

      .. rubric:: Example

      >>> with Http() as client:
      ...     response = client.put("https://api.example.com/data/1", json={"key": "value"})



   .. py:method:: delete(url: str, **kwargs: Any) -> requests.Response

      Send a DELETE request with enhanced logging.

      :param url: Target URL for the DELETE request.
      :param \*\*kwargs: Additional keyword arguments passed to requests.

      :returns: The HTTP response object.
      :rtype: requests.Response

      .. rubric:: Example

      >>> with Http() as client:
      ...     response = client.delete("https://api.example.com/data/1")