Base HTTP Client

psengine.base_http_client.BaseHTTPClient

BaseHTTPClient(
    http_proxy: str = None,
    https_proxy: str = None,
    verify: Union[str, bool] = SSL_VERIFY,
    auth: tuple[str, str] = None,
    cert: Union[str, tuple[str, str], None] = None,
    timeout: int = REQUEST_TIMEOUT,
    retries: int = RETRY_TOTAL,
    backoff_factor: int = BACKOFF_FACTOR,
    status_forcelist: int = STATUS_FORCELIST,
    pool_max_size: int = POOL_MAX_SIZE,
)

Generic HTTP client for making requests (requests wrapper).

PARAMETER	DESCRIPTION
http_proxy	An HTTP proxy URL. TYPE: str DEFAULT: None
https_proxy	An HTTPS proxy URL. TYPE: str DEFAULT: None
verify	An SSL verification flag or path to CA bundle. TYPE: Union[str, bool] DEFAULT: SSL_VERIFY
auth	Basic Auth credentials. TYPE: tuple[str, str] DEFAULT: None
cert	Client certificates. TYPE: Union[str, tuple[str, str], None] DEFAULT: None
timeout	A request timeout. TYPE: int DEFAULT: REQUEST_TIMEOUT
retries	A number of retries. TYPE: int DEFAULT: RETRY_TOTAL
backoff_factor	A backoff factor. TYPE: int DEFAULT: BACKOFF_FACTOR
status_forcelist	A list of status codes to force a retry. Defaults to [502, 503, 504]. TYPE: int DEFAULT: STATUS_FORCELIST
pool_max_size	The maximum number of connections in the pool. TYPE: int DEFAULT: POOL_MAX_SIZE

Source code in psengine/base_http_client.py

def __init__(
    self,
    http_proxy: Annotated[str, Doc('An HTTP proxy URL.')] = None,
    https_proxy: Annotated[str, Doc('An HTTPS proxy URL.')] = None,
    verify: Annotated[
        Union[str, bool], Doc('An SSL verification flag or path to CA bundle.')
    ] = SSL_VERIFY,
    auth: Annotated[tuple[str, str], Doc('Basic Auth credentials.')] = None,
    cert: Annotated[Union[str, tuple[str, str], None], Doc('Client certificates.')] = None,
    timeout: Annotated[int, Doc('A request timeout.')] = REQUEST_TIMEOUT,
    retries: Annotated[int, Doc('A number of retries.')] = RETRY_TOTAL,
    backoff_factor: Annotated[int, Doc('A backoff factor.')] = BACKOFF_FACTOR,
    status_forcelist: Annotated[
        int, Doc('A list of status codes to force a retry. Defaults to [502, 503, 504].')
    ] = STATUS_FORCELIST,
    pool_max_size: Annotated[
        int, Doc('The maximum number of connections in the pool.')
    ] = POOL_MAX_SIZE,
):
    """Generic HTTP client for making requests (`requests` wrapper)."""
    self.log = logging.getLogger(__name__)
    self.config = get_config()
    self.http_proxy = http_proxy if http_proxy is not None else self.config.http_proxy
    self.https_proxy = https_proxy if https_proxy is not None else self.config.https_proxy
    self.proxies = self._set_proxies()
    self.verify = verify if verify is not None else self.config.client_ssl_verify
    self.timeout = timeout if timeout is not None else self.config.client_timeout
    self.retries = retries if retries is not None else self.config.client_retries
    self.backoff_factor = (
        backoff_factor if backoff_factor is not None else self.config.client_backoff_factor
    )
    self.status_forcelist = (
        status_forcelist
        if status_forcelist is not None
        else self.config.client_status_forcelist
    )
    self.pool_max_size = (
        pool_max_size if pool_max_size is not None else self.config.client_pool_max_size
    )
    self.session = self._create_session()
    self.session.cert = cert if cert is not None else self.config.client_cert
    self.session.auth = auth if auth is not None else self.config.client_basic_auth

    self._set_retry_config()

backoff_factor instance-attribute

backoff_factor = (
    backoff_factor
    if backoff_factor is not None
    else client_backoff_factor
)

config instance-attribute

config = get_config()

http_proxy instance-attribute

http_proxy = (
    http_proxy if http_proxy is not None else http_proxy
)

https_proxy instance-attribute

https_proxy = (
    https_proxy if https_proxy is not None else https_proxy
)

log instance-attribute

log = getLogger(__name__)

pool_max_size instance-attribute

pool_max_size = (
    pool_max_size
    if pool_max_size is not None
    else client_pool_max_size
)

proxies instance-attribute

proxies = _set_proxies()

retries instance-attribute

retries = retries if retries is not None else client_retries

session instance-attribute

session = _create_session()

status_forcelist instance-attribute

status_forcelist = (
    status_forcelist
    if status_forcelist is not None
    else client_status_forcelist
)

timeout instance-attribute

timeout = timeout if timeout is not None else client_timeout

verify instance-attribute

verify = verify if verify is not None else client_ssl_verify

call

call(
    method: str,
    url: str,
    data: Union[dict, list[dict], None] = None,
    *,
    params: Union[dict, None] = None,
    headers: Union[dict, None] = None,
    **kwargs,
) -> Response

Invoke an HTTP request using the requests library.

PARAMETER	DESCRIPTION
method	An HTTP method. TYPE: str
url	A URL to make the request to. TYPE: str
data	A request body. TYPE: Union[dict, list[dict], None] DEFAULT: None
params	HTTP query parameters. TYPE: Union[dict, None] DEFAULT: None
headers	If specified, overrides default headers and does not set the token. TYPE: Union[dict, None] DEFAULT: None

RAISES	DESCRIPTION
ValueError	If method is not one of GET, PUT, POST, DELETE, HEAD, OPTIONS, PATCH.
HTTPError	If the response status is not 2xx.
JSONDecodeError	If the response contains malformed JSON.
ConnectTimeout	If the connection to the server times out.
ConnectionError	If the request fails before completing.
ReadTimeout	If the server did not send any data in time.

RETURNS	DESCRIPTION
Response	A requests.Response object.

Source code in psengine/base_http_client.py

@debug_call
@validate_call
def call(
    self,
    method: Annotated[str, Doc('An HTTP method.')],
    url: Annotated[str, Doc('A URL to make the request to.')],
    data: Annotated[Union[dict, list[dict], None], Doc('A request body.')] = None,
    *,
    params: Annotated[Union[dict, None], Doc('HTTP query parameters.')] = None,
    headers: Annotated[
        Union[dict, None],
        Doc('If specified, overrides default headers and does not set the token.'),
    ] = None,
    **kwargs,
) -> Annotated[Response, Doc('A requests.Response object.')]:
    """Invoke an HTTP request using the `requests` library.

    Raises:
        ValueError: If method is not one of GET, PUT, POST, DELETE, HEAD, OPTIONS, PATCH.
        HTTPError: If the response status is not 2xx.
        JSONDecodeError: If the response contains malformed JSON.
        ConnectTimeout: If the connection to the server times out.
        ConnectionError: If the request fails before completing.
        ReadTimeout: If the server did not send any data in time.
    """
    method_func = self._choose_method_type(method)

    if not headers:
        headers = {}

    if 'User-Agent' not in headers:
        headers['User-Agent'] = self._get_user_agent_header()

    data = json.dumps(data) if data is not None else None

    try:
        response = method_func(
            url=url,
            headers=headers,
            data=data,
            params=params,
            verify=self.verify,
            timeout=self.timeout,
            **kwargs,
        )
        self.log.debug(f'HTTP Status Code: {response.status_code}')

    except (ConnectionError, ConnectTimeout, ReadTimeout) as err:
        self.log.debug(f'GET request failed. Cause: {err}')
        raise

    try:
        response.raise_for_status()

    except HTTPError as err:
        msg = str(err)
        try:
            data = response.json()
        except JSONDecodeError:
            data = {}

        message = data.get('message') or data.get('error', {})
        if isinstance(message, dict):
            message = message.get('message')

        if message:
            msg += f', Cause: {message}'

        self.log.debug(f'{method} request failed. {msg}')

        raise HTTPError(msg, response=response) from err

    return response

can_connect

can_connect(
    method: str = 'get', url: str = BASE_URL
) -> bool

Check if the client can reach the specified API URL.

PARAMETER	DESCRIPTION
method	An HTTP method. TYPE: str DEFAULT: 'get'
url	A URL to test.recordedfuture.com. TYPE: str DEFAULT: BASE_URL

RETURNS	DESCRIPTION
bool	True if connection returns status 200, else False.

Source code in psengine/base_http_client.py

@debug_call
@validate_call
def can_connect(
    self,
    method: Annotated[str, Doc('An HTTP method.')] = 'get',
    url: Annotated[str, Doc('A URL to test.recordedfuture.com.')] = BASE_URL,
) -> Annotated[bool, Doc('True if connection returns status 200, else False.')]:
    """Check if the client can reach the specified API URL."""
    try:
        request = self.call(method=method, url=url)
        request.raise_for_status()
        return True
    except Exception as err:  # noqa: BLE001
        self.log.error(f'Error during connectivity test: {err}')
        return False

set_urllib_log_level

set_urllib_log_level(level: str) -> None

Set log level for urllib3 library.

PARAMETER	DESCRIPTION
level	A log level to be set: CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET. TYPE: str

Source code in psengine/base_http_client.py

@debug_call
@validate_call
def set_urllib_log_level(
    self,
    level: Annotated[
        str, Doc('A log level to be set: CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET.')
    ],
) -> None:
    """Set log level for urllib3 library."""
    if not level or level.upper() not in (
        'CRITICAL',
        'ERROR',
        'WARNING',
        'INFO',
        'DEBUG',
        'NOTSET',
    ):
        self.log.warning('Log level is empty or not valid')
        return
    logging.getLogger('urllib3').setLevel(level.upper())