ds_protocol_http_py_lib.utils.http.provider

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

Example

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

Attributes

logger

Classes

Http	Minimal synchronous HTTP client with:

Module Contents

ds_protocol_http_py_lib.utils.http.provider.logger

class ds_protocol_http_py_lib.utils.http.provider.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

_cfg

_bucket

_session

_build_session() → requests.Session

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

_response_info(response: requests.Response) → dict[str, Any]

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

Parameters:

response – The HTTP response object to extract info from.

Returns:

Dictionary containing the response information.

Return type:

dict[str, Any]

property session: requests.Session

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

Returns:

The requests session.

Return type:

requests.Session

Example

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

__enter__() → Http

__exit__(exc_type: Any, exc: Any, tb: Any) → None

close() → None

request(method: str, url: str, **kwargs: Any) → requests.Response

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

Parameters:

• method – HTTP method (GET, POST, PUT, DELETE, etc.).

• url – Target URL for the request.

• **kwargs – Additional keyword arguments passed to requests (timeout, headers, data, etc.).

Returns:

The HTTP response object.

Return type:

requests.Response

Raises:

• requests.HTTPError – If the response status code indicates an error.

• requests.RequestException – For other request-related errors.

Example

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

get(url: str, **kwargs: Any) → requests.Response

Send a GET request with enhanced logging.

Parameters:

• url – Target URL for the GET request.

• **kwargs – Additional keyword arguments passed to requests.

Returns:

The HTTP response object.

Return type:

requests.Response

Example

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

post(url: str, **kwargs: Any) → requests.Response

Send a POST request with enhanced logging.

Parameters:

• url – Target URL for the POST request.

• **kwargs – Additional keyword arguments passed to requests.

Returns:

The HTTP response object.

Return type:

requests.Response

Example

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

put(url: str, **kwargs: Any) → requests.Response

Send a PUT request with enhanced logging.

Parameters:

• url – Target URL for the PUT request.

• **kwargs – Additional keyword arguments passed to requests.

Returns:

The HTTP response object.

Return type:

requests.Response

Example

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

delete(url: str, **kwargs: Any) → requests.Response

Send a DELETE request with enhanced logging.

Parameters:

• url – Target URL for the DELETE request.

• **kwargs – Additional keyword arguments passed to requests.

Returns:

The HTTP response object.

Return type:

requests.Response

Example

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