HTTP Core

HttpConnProperties(host, api_key=None, user=None, password=None, conn_timeout=30, request_timeout=300, verify_ssl=False, headers=(lambda: {'content-type': 'application/json', 'accept': 'application/json'})(), payload_kwargs=None, payload_kwargs_key='json') dataclass

Class that holds the properties used to establish a connection to an HTTP server.

Parameters:

• host

  (str) –

  The hostname of the database server.

• api_key

  (str | None, default: None ) –

  API key used to connect to the server. This will be the preferred method of authentication instead of user and password. By default None.

• user

  (str | None, default: None ) –

  User name. This will only be used if api_key is not provided. By default None.

• password

  (str | None, default: None ) –

  Password for the user. This will only be used if api_key is not provided. By default None.

• conn_timeout

  (int, default: 30 ) –

  Timeout in seconds for connection, by default 30

• request_timeout

  (int, default: 300 ) –

  Query timeout in seconds, by default 300

• verify_ssl

  (bool, default: False ) –

  If True, SSL certificates will be verified, by default False

• headers

  (dict | None, default: (lambda: {'content-type': 'application/json', 'accept': 'application/json'})() ) –

  Dictionary with headers to be added to the request.

  By default {"content-type": "application/json", "accept": "application/json"}.

• payload_kwargs

  (dict | None, default: None ) –

  Dictionary with key-word arguments to be added to the payload of every request made to the server.

  Mainly used in subclasses that require a token at every request.

  By default None.

• payload_kwargs_key

  (str, default: 'json' ) –

  Key to be used in the payload kwargs. By default "json".

  This defines the key to be used in the payload kwargs. Usually this is "json" but in some cases it can be "data" or "params".

  By default "json".

__post_init__()

Method that checks if inputs are valid after initialization.

Source code in echo_connhandler/http_core.py

def __post_init__(self) -> None:
    """Method that checks if inputs are valid after initialization."""
    if not isinstance(self.host, str):
        raise TypeError(f"host must be a string not {type(self.host)}")
    if not isinstance(self.api_key, str | type(None)):
        raise TypeError(f"api_key must be a string or None not {type(self.api_key)}")
    if not isinstance(self.user, str | type(None)):
        raise TypeError(f"user must be a string or None not {type(self.user)}")
    if not isinstance(self.password, str | type(None)):
        raise TypeError(f"password must be a string or None not {type(self.password)}")
    if not isinstance(self.conn_timeout, int):
        raise TypeError(f"conn_timeout must be an integer not {type(self.conn_timeout)}")
    if not isinstance(self.request_timeout, int):
        raise TypeError(f"request_timeout must be an integer not {type(self.request_timeout)}")
    if self.api_key and (self.user or self.password):
        raise ValueError("api_key and user/password cannot be provided at the same time.")
    if not isinstance(self.verify_ssl, bool):
        raise TypeError(f"verify_ssl must be a boolean not {type(self.verify_ssl)}")
    if not isinstance(self.payload_kwargs, dict | type(None)):
        raise TypeError(f"payload_kwargs must be a dictionary or None not {type(self.payload_kwargs)}")
    if not isinstance(self.payload_kwargs_key, str):
        raise TypeError(f"payload_kwargs_key must be a string not {type(self.payload_kwargs_key)}")

HttpHandler(connection_properties, max_retries=3, retry_wait_time=5, skip_connect=False, **kwargs)

Base abstract class for HTTP handlers.

This class should be used to handle connecting to HTTP servers and using APIs. This is an abstract class so it should not be instantiated directly.

This already connects to the HTTP server.

Parameters:

• connection_properties

  (HttpConnProperties) –

  Object containing connection parameters.

• max_retries

  (int, default: 3 ) –

  Number of retries that will be attempted when reconnecting or doing queries, by default 3

• retry_wait_time

  (float, default: 5 ) –

  Wait time in seconds between each connection or query retry, by default 5

• skip_connect

  (bool, default: False ) –

  If True, the connection will not be established when the object is created.

  If this is set toTrue, the user will need to manually call the reconnect() method when they want to connect to the server.

  By default False

• **kwargs

  (dict, default: {} ) –

  Just kept here for compatibility.

Source code in echo_connhandler/http_core.py

@validate_call
def __init__(
    self,
    connection_properties: HttpConnProperties,
    max_retries: int = 3,
    retry_wait_time: float = 5,
    skip_connect: bool = False,
    **kwargs,  # pylint: disable=unused-argument # noqa
) -> None:
    """Method that initializes the HTTP handler.

    This already connects to the HTTP server.

    Parameters
    ----------
    connection_properties : HttpConnProperties
        Object containing connection parameters.
    max_retries : int, optional
        Number of retries that will be attempted when reconnecting or doing queries, by default 3
    retry_wait_time : float, optional
        Wait time in seconds between each connection or query retry, by default 5
    skip_connect : bool, optional
        If True, the connection will not be established when the object is created.

        If this is set toTrue, the user will need to manually call the reconnect() method when they want to connect to the server.

        By default False
    **kwargs : dict, optional
        Just kept here for compatibility.

    """
    super().__init__(
        connection_properties=connection_properties,
        max_retries=max_retries,
        retry_wait_time=retry_wait_time,
        skip_connect=skip_connect,
    )

delete(endpoint, response_ok=200, **kwargs)

Method used to do a DELETE request in the desired endpoint.

Parameters:

• endpoint

  (str) –

  Endpoint of the API. Can be just the final part of the URL or can the full URL.

• response_ok

  (int | None, default: 200 ) –

  Response code that will be considered as a success. If None, no checks will be made on the response code.

  By default 200.

• **kwargs

  (dict, default: {} ) –

  Any key-word arguments to be passed to the request function.

Returns:

• Response –

  Response of the request.

Source code in echo_connhandler/http_core.py

@validate_call(config={"str_min_length": None})
def delete(self, endpoint: str, response_ok: int | None = 200, **kwargs) -> httpx.Response:
    """Method used to do a DELETE request in the desired endpoint.

    Parameters
    ----------
    endpoint : str
        Endpoint of the API. Can be just the final part of the URL or can the full URL.
    response_ok : int | None, optional
        Response code that will be considered as a success. If None, no checks will be made on the response code.

        By default 200.
    **kwargs : dict
        Any key-word arguments to be passed to the request function.

    Returns
    -------
    httpx.Response
        Response of the request.

    """
    return self._request(endpoint=endpoint, request_type="DELETE", response_ok=response_ok, **kwargs)

get(endpoint, response_ok=200, **kwargs)

Method used to do a GET request in the desired endpoint.

Parameters:

• endpoint

  (str) –

  Endpoint of the API. Can be just the final part of the URL or can the full URL.

• response_ok

  (int | None, default: 200 ) –

  Response code that will be considered as a success. If None, no checks will be made on the response code.

  By default 200.

• **kwargs

  (dict, default: {} ) –

  Any key-word arguments to be passed to the request function.

Returns:

• Response –

  Response of the request.

Source code in echo_connhandler/http_core.py

@validate_call(config={"str_min_length": None})
def get(self, endpoint: str, response_ok: int | None = 200, **kwargs) -> httpx.Response:
    """Method used to do a GET request in the desired endpoint.

    Parameters
    ----------
    endpoint : str
        Endpoint of the API. Can be just the final part of the URL or can the full URL.
    response_ok : int | None, optional
        Response code that will be considered as a success. If None, no checks will be made on the response code.

        By default 200.
    **kwargs : dict
        Any key-word arguments to be passed to the request function.

    Returns
    -------
    httpx.Response
        Response of the request.

    """
    return self._request(endpoint=endpoint, request_type="GET", response_ok=response_ok, **kwargs)

patch(endpoint, response_ok=200, **kwargs)

Method used to do a PATCH request in the desired endpoint.

Parameters:

• endpoint

  (str) –

  Endpoint of the API. Can be just the final part of the URL or can the full URL.

• response_ok

  (int | None, default: 200 ) –

  Response code that will be considered as a success. If None, no checks will be made on the response code.

  By default 200.

• **kwargs

  (dict, default: {} ) –

  Any key-word arguments to be passed to the request function.

Returns:

• Response –

  Response of the request.

Source code in echo_connhandler/http_core.py

@validate_call(config={"str_min_length": None})
def patch(self, endpoint: str, response_ok: int | None = 200, **kwargs) -> httpx.Response:
    """Method used to do a PATCH request in the desired endpoint.

    Parameters
    ----------
    endpoint : str
        Endpoint of the API. Can be just the final part of the URL or can the full URL.
    response_ok : int | None, optional
        Response code that will be considered as a success. If None, no checks will be made on the response code.

        By default 200.
    **kwargs : dict
        Any key-word arguments to be passed to the request function.

    Returns
    -------
    httpx.Response
        Response of the request.

    """
    return self._request(endpoint=endpoint, request_type="PATCH", response_ok=response_ok, **kwargs)

post(endpoint, response_ok=200, **kwargs)

Method used to do a GET request in the desired endpoint.

Parameters:

• endpoint

  (str) –

  Endpoint of the API. Can be just the final part of the URL or can the full URL.

• response_ok

  (int | None, default: 200 ) –

  Response code that will be considered as a success. If None, no checks will be made on the response code.

  By default 200.

• **kwargs

  (dict, default: {} ) –

  Any key-word arguments to be passed to the request function.

Returns:

• Response –

  Response of the request.

Source code in echo_connhandler/http_core.py

@validate_call(config={"str_min_length": None})
def post(self, endpoint: str, response_ok: int | None = 200, **kwargs) -> httpx.Response:
    """Method used to do a GET request in the desired endpoint.

    Parameters
    ----------
    endpoint : str
        Endpoint of the API. Can be just the final part of the URL or can the full URL.
    response_ok : int | None, optional
        Response code that will be considered as a success. If None, no checks will be made on the response code.

        By default 200.
    **kwargs : dict
        Any key-word arguments to be passed to the request function.

    Returns
    -------
    httpx.Response
        Response of the request.

    """
    return self._request(endpoint=endpoint, request_type="POST", response_ok=response_ok, **kwargs)

put(endpoint, response_ok=200, **kwargs)

Method used to do a PUT request in the desired endpoint.

Parameters:

• endpoint

  (str) –

  Endpoint of the API. Can be just the final part of the URL or can the full URL.

• response_ok

  (int | None, default: 200 ) –

  Response code that will be considered as a success. If None, no checks will be made on the response code.

  By default 200.

• **kwargs

  (dict, default: {} ) –

  Any key-word arguments to be passed to the request function.

Returns:

• Response –

  Response of the request.

Source code in echo_connhandler/http_core.py

@validate_call(config={"str_min_length": None})
def put(self, endpoint: str, response_ok: int | None = 200, **kwargs) -> httpx.Response:
    """Method used to do a PUT request in the desired endpoint.

    Parameters
    ----------
    endpoint : str
        Endpoint of the API. Can be just the final part of the URL or can the full URL.
    response_ok : int | None, optional
        Response code that will be considered as a success. If None, no checks will be made on the response code.

        By default 200.
    **kwargs : dict
        Any key-word arguments to be passed to the request function.

    Returns
    -------
    httpx.Response
        Response of the request.

    """
    return self._request(endpoint=endpoint, request_type="PUT", response_ok=response_ok, **kwargs)