from collections.abc import Generator
from typing import Any
import requests
from requests import RequestException
from urllib3.exceptions import NewConnectionError
from ..__version__ import __version__
from ..client.response import Response
from .base_sync_client import BaseSyncClient
USER_AGENT = f"py-stellar-base/{__version__}/SimpleRequestsClient"
HEADERS = {
"X-Client-Name": "py-stellar-base",
"X-Client-Version": __version__,
"User-Agent": USER_AGENT,
}
__all__ = ["SimpleRequestsClient"]
[docs]
class SimpleRequestsClient(BaseSyncClient):
"""The :class:`SimpleRequestsClient` object is a synchronous http client,
which represents the interface for making requests to a server instance.
**This client is to guide you in writing a client that suits your needs.
I don't recommend that you actually use it.**
"""
[docs]
def get(
self,
url: str,
params: dict[str, str] | None = None,
max_content_size: int | None = None,
) -> Response:
"""Perform HTTP GET request.
:param url: the request url
:param params: the request params
:param max_content_size: **Not supported in SimpleRequestsClient**.
:return: the response from server
:raise: :exc:`ConnectionError <stellar_sdk.exceptions.ConnectionError>`
"""
if max_content_size is not None:
raise NotImplementedError(
"max_content_size is not supported in SimpleRequestsClient, "
"please use RequestsClient instead."
)
try:
resp = requests.get(url=url, params=params, headers=HEADERS)
except (RequestException, NewConnectionError) as err:
raise ConnectionError(err) from err
return Response(
status_code=resp.status_code,
text=resp.text,
headers=dict(resp.headers),
url=resp.url,
)
[docs]
def post(
self,
url: str,
data: dict[str, str] | None = None,
json_data: dict[str, Any] | None = None,
) -> Response:
"""Perform HTTP POST request.
:param url: the request url
:param data: the data send to server
:param json_data: the json data send to server
:return: the response from server
:raise: :exc:`ConnectionError <stellar_sdk.exceptions.ConnectionError>`
"""
try:
resp = requests.post(url=url, data=data, json=json_data, headers=HEADERS)
except (RequestException, NewConnectionError) as err:
raise ConnectionError(err) from err
return Response(
status_code=resp.status_code,
text=resp.text,
headers=dict(resp.headers),
url=resp.url,
)
[docs]
def stream(
self, url: str, params: dict[str, str] | None = None
) -> Generator[dict[str, Any], None, None]:
"""
**Not Implemented**
:param url: the request url
:param params: the request params
:return: None
"""
raise NotImplementedError
def close(self):
pass
def __repr__(self):
return "<SimpleRequestsClient>"