request

dev_tool.services.request

__all__ = ['RequestService'] module-attribute

RequestService

A singleton service class for HTTP request operations.

This class provides methods for making HTTP requests with connection pooling, timeout handling, and automatic error management.

The constructor for the RequestService class.

Parameters:

• timeout (float, default: 360.0 ) –

  The request timeout in seconds.

• connect_timeout (float, default: 5.0 ) –

  The connection timeout in seconds.

• max_keepalive_connections (int, default: 5 ) –

  The maximum number of keepalive connections.

• max_connections (int, default: 10 ) –

  The maximum number of total connections.

Source code in dev_tool/services/request/service.py

def __init__(
    self,
    timeout: float = 360.0,
    connect_timeout: float = 5.0,
    max_keepalive_connections: int = 5,
    max_connections: int = 10
) -> None:
    """
    The constructor for the RequestService class.

    :param timeout: The request timeout in seconds.
    :param connect_timeout: The connection timeout in seconds.
    :param max_keepalive_connections: The maximum number of keepalive connections.
    :param max_connections: The maximum number of total connections.
    """

    if hasattr(self, 'client'):
        return

    self.client = httpx.Client(
        follow_redirects=True,
        limits=httpx.Limits(
            max_keepalive_connections=max_keepalive_connections,
            max_connections=max_connections
        ),
        timeout=httpx.Timeout(timeout, connect=connect_timeout)
    )

    atexit.register(self.close)

client = httpx.Client(follow_redirects=True, limits=(httpx.Limits(max_keepalive_connections=max_keepalive_connections, max_connections=max_connections)), timeout=(httpx.Timeout(timeout, connect=connect_timeout))) instance-attribute

__new__

A method that implements the singleton pattern for the RequestService.

Returns:

• Self –

  The singleton RequestService instance.

Source code in dev_tool/services/request/service.py

def __new__(cls, *_args, **_kwargs) -> Self:
    """
    A method that implements the singleton pattern for the RequestService.

    :return: The singleton RequestService instance.
    """

    if cls._instance is None:
        with cls._lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)

    assert isinstance(cls._instance, cls)
    return cls._instance

close

A method that closes the HTTP client and cleans up resources.

Source code in dev_tool/services/request/service.py

def close(self) -> None:
    """A method that closes the HTTP client and cleans up resources."""

    if hasattr(self, 'client'):
        self.client.close()

request

A method that makes a generic HTTP request.

Parameters:

• method (str) –

  The HTTP method to use.

• url (str) –

  The URL to request.

• kwargs –

  Additional parameters for the request.

Returns:

• Response –

  The HTTP response object.

Raises:

• HttpTimeoutError –

  If the request times out.

• HttpConnectionError –

  If the connection fails.

• HttpRequestError –

  If the request fails.

Source code in dev_tool/services/request/service.py

def request(self, method: str, url: str, **kwargs) -> httpx.Response:
    """
    A method that makes a generic HTTP request.

    :param method: The HTTP method to use.
    :param url: The URL to request.
    :param kwargs: Additional parameters for the request.
    :return: The HTTP response object.
    :raises HttpTimeoutError: If the request times out.
    :raises HttpConnectionError: If the connection fails.
    :raises HttpRequestError: If the request fails.
    """

    try:
        return self.client.request(method, url, **kwargs)
    except httpx.TimeoutException as exception:
        message = f'Request timeout for {method}'
        log.exception(message)

        raise HttpTimeoutError(message) from exception
    except httpx.ConnectError as exception:
        message = f'Connection error for {method}'
        log.exception(message)

        raise HttpConnectionError(message) from exception
    except httpx.HTTPError as exception:
        message = f'HTTP error for {method}'
        log.exception(message)

        raise HttpRequestError(message) from exception
    except Exception as exception:
        message = f'Unexpected error for {method}'
        log.exception(message)

        raise HttpRequestError(message) from exception

get

A method that makes a GET request.

Parameters:

• url (str) –

  The URL to request.

• kwargs –

  Additional parameters for the request.

Returns:

• Response –

  The HTTP response object.

Raises:

• HttpTimeoutError –

  If the request times out.

• HttpConnectionError –

  If the connection fails.

• HttpRequestError –

  If the request fails.

Source code in dev_tool/services/request/service.py

def get(self, url: str, **kwargs) -> httpx.Response:
    """
    A method that makes a GET request.

    :param url: The URL to request.
    :param kwargs: Additional parameters for the request.
    :return: The HTTP response object.
    :raises HttpTimeoutError: If the request times out.
    :raises HttpConnectionError: If the connection fails.
    :raises HttpRequestError: If the request fails.
    """

    try:
        return self.client.get(url, **kwargs)
    except httpx.TimeoutException as exception:
        message = 'GET request timeout'
        log.exception(message)

        raise HttpTimeoutError(message) from exception
    except httpx.ConnectError as exception:
        message = 'GET connection error'
        log.exception(message)

        raise HttpConnectionError(message) from exception
    except httpx.HTTPError as exception:
        message = 'GET HTTP error for'
        log.exception(message)

        raise HttpRequestError(message) from exception
    except Exception as exception:
        message = 'Unexpected GET error'
        log.exception(message)

        raise HttpRequestError(message) from exception

post

A method that makes a POST request.

Parameters:

• url (str) –

  The URL to request.

• kwargs –

  Additional parameters for the request.

Returns:

• Response –

  The HTTP response object.

Raises:

• HttpTimeoutError –

  If the request times out.

• HttpConnectionError –

  If the connection fails.

• HttpRequestError –

  If the request fails.

Source code in dev_tool/services/request/service.py

def post(self, url: str, **kwargs) -> httpx.Response:
    """
    A method that makes a POST request.

    :param url: The URL to request.
    :param kwargs: Additional parameters for the request.
    :return: The HTTP response object.
    :raises HttpTimeoutError: If the request times out.
    :raises HttpConnectionError: If the connection fails.
    :raises HttpRequestError: If the request fails.
    """

    try:
        return self.client.post(url, **kwargs)
    except httpx.TimeoutException as exception:
        message = 'POST request timeout'
        log.exception(message)

        raise HttpTimeoutError(message) from exception
    except httpx.ConnectError as exception:
        message = 'POST connection error'
        log.exception(message)

        raise HttpConnectionError(message) from exception
    except httpx.HTTPError as exception:
        message = 'POST HTTP error'
        log.exception(message)

        raise HttpRequestError(message) from exception
    except Exception as exception:
        message = 'Unexpected POST error'
        log.exception(message)

        raise HttpRequestError(message) from exception

put

A method that makes a PUT request.

Parameters:

• url (str) –

  The URL to request.

• kwargs –

  Additional parameters for the request.

Returns:

• Response –

  The HTTP response object.

Raises:

• HttpTimeoutError –

  If the request times out.

• HttpConnectionError –

  If the connection fails.

• HttpRequestError –

  If the request fails.

Source code in dev_tool/services/request/service.py

def put(self, url: str, **kwargs) -> httpx.Response:
    """
    A method that makes a PUT request.

    :param url: The URL to request.
    :param kwargs: Additional parameters for the request.
    :return: The HTTP response object.
    :raises HttpTimeoutError: If the request times out.
    :raises HttpConnectionError: If the connection fails.
    :raises HttpRequestError: If the request fails.
    """

    try:
        return self.client.put(url, **kwargs)
    except httpx.TimeoutException as exception:
        message = 'PUT request timeout'
        log.exception(message)

        raise HttpTimeoutError(message) from exception
    except httpx.ConnectError as exception:
        message = 'PUT connection error'
        log.exception(message)

        raise HttpConnectionError(message) from exception
    except httpx.HTTPError as exception:
        message = 'PUT HTTP error'
        log.exception(message)

        raise HttpRequestError(message) from exception
    except Exception as exception:
        message = 'Unexpected PUT error'
        log.exception(message)

        raise HttpRequestError(message) from exception

delete

A method that makes a DELETE request.

Parameters:

• url (str) –

  The URL to request.

• kwargs –

  Additional parameters for the request.

Returns:

• Response –

  The HTTP response object.

Raises:

• HttpTimeoutError –

  If the request times out.

• HttpConnectionError –

  If the connection fails.

• HttpRequestError –

  If the request fails.

Source code in dev_tool/services/request/service.py

def delete(self, url: str, **kwargs) -> httpx.Response:
    """
    A method that makes a DELETE request.

    :param url: The URL to request.
    :param kwargs: Additional parameters for the request.
    :return: The HTTP response object.
    :raises HttpTimeoutError: If the request times out.
    :raises HttpConnectionError: If the connection fails.
    :raises HttpRequestError: If the request fails.
    """

    try:
        return self.client.delete(url, **kwargs)
    except httpx.TimeoutException as exception:
        message = 'DELETE request timeout'
        log.exception(message)

        raise HttpTimeoutError(message) from exception
    except httpx.ConnectError as exception:
        message = 'DELETE connection error'
        log.exception(message)

        raise HttpConnectionError(message) from exception
    except httpx.HTTPError as exception:
        message = 'DELETE HTTP error'
        log.exception(message)

        raise HttpRequestError(message) from exception
    except Exception as exception:
        message = 'Unexpected DELETE error'
        log.exception(message)

        raise HttpRequestError(message) from exception

patch

A method that makes a PATCH request.

Parameters:

• url (str) –

  The URL to request.

• kwargs –

  Additional parameters for the request.

Returns:

• Response –

  The HTTP response object.

Raises:

• HttpTimeoutError –

  If the request times out.

• HttpConnectionError –

  If the connection fails.

• HttpRequestError –

  If the request fails.

Source code in dev_tool/services/request/service.py

def patch(self, url: str, **kwargs) -> httpx.Response:
    """
    A method that makes a PATCH request.

    :param url: The URL to request.
    :param kwargs: Additional parameters for the request.
    :return: The HTTP response object.
    :raises HttpTimeoutError: If the request times out.
    :raises HttpConnectionError: If the connection fails.
    :raises HttpRequestError: If the request fails.
    """

    try:
        return self.client.patch(url, **kwargs)
    except httpx.TimeoutException as exception:
        message = 'PATCH request timeout'
        log.exception(message)

        raise HttpTimeoutError(message) from exception
    except httpx.ConnectError as exception:
        message = 'PATCH connection error'
        log.exception(message)

        raise HttpConnectionError(message) from exception
    except httpx.HTTPError as exception:
        message = 'PATCH HTTP error'
        log.exception(message)

        raise HttpRequestError(message) from exception
    except Exception as exception:
        message = 'Unexpected PATCH error'
        log.exception(message)

        raise HttpRequestError(message) from exception

stream

A method that creates a streaming HTTP request.

Parameters:

• method (str) –

  The HTTP method to use.

• url (str) –

  The URL to request.

• kwargs –

  Additional parameters for the request.

Returns:

• ContextManager[Response] –

  A context manager for the streaming response.

Raises:

• HttpTimeoutError –

  If the request times out.

• HttpConnectionError –

  If the connection fails.

• HttpRequestError –

  If the request fails.

Source code in dev_tool/services/request/service.py

def stream(self, method: str, url: str, **kwargs) -> ContextManager[httpx.Response]:
    """
    A method that creates a streaming HTTP request.

    :param method: The HTTP method to use.
    :param url: The URL to request.
    :param kwargs: Additional parameters for the request.
    :return: A context manager for the streaming response.
    :raises HttpTimeoutError: If the request times out.
    :raises HttpConnectionError: If the connection fails.
    :raises HttpRequestError: If the request fails.
    """

    try:
        return self.client.stream(method, url, **kwargs)
    except httpx.TimeoutException as exception:
        message = f'Stream request timeout for {method}'
        log.exception(message)

        raise HttpTimeoutError(message) from exception
    except httpx.ConnectError as exception:
        message = f'Stream connection error for {method}'
        log.exception(message)

        raise HttpConnectionError(message) from exception
    except httpx.HTTPError as exception:
        message = f'Stream HTTP error for {method}'
        log.exception(message)

        raise HttpRequestError(message) from exception
    except Exception as exception:
        message = f'Unexpected stream error for {method}'
        log.exception(message)

        raise HttpRequestError(message) from exception