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)

    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:
        message = f'Request timeout for {method}'
        log.exception(message)

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

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

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

        raise HttpRequestError(message) from None

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:
        message = 'GET request timeout'
        log.exception(message)

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

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

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

        raise HttpRequestError(message) from None

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:
        message = 'POST request timeout'
        log.exception(message)

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

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

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

        raise HttpRequestError(message) from None

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:
        message = 'PUT request timeout'
        log.exception(message)

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

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

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

        raise HttpRequestError(message) from None

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:
        message = 'DELETE request timeout'
        log.exception(message)

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

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

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

        raise HttpRequestError(message) from None

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:
        message = 'PATCH request timeout'
        log.exception(message)

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

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

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

        raise HttpRequestError(message) from None

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:
        message = f'Stream request timeout for {method}'
        log.exception(message)

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

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

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

        raise HttpRequestError(message) from None