Skip to content

Client

psengine.sandbox.client

SandboxClient 

SandboxClient(
    api_token: str | None = None,
    http_proxy: str = None,
    https_proxy: str = None,
    verify: str | bool = None,
    auth: tuple[str, str] = None,
    cert: str | tuple[str, str] | None = None,
    timeout: int = None,
    retries: int = None,
    backoff_factor: int = None,
    status_forcelist: list = None,
    pool_max_size: int = None,
)

Bases: BaseHTTPClient

Recorded Future Sandbox API client.

PARAMETER DESCRIPTION
api_token

The Sandbox API token. Defaults to RF_SANDBOX_TOKEN environment variable.

TYPE: str | None DEFAULT: None

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: str | bool DEFAULT: None

auth

Basic Auth credentials.

TYPE: tuple[str, str] DEFAULT: None

cert

Client certificates.

TYPE: str | tuple[str, str] | None DEFAULT: None

timeout

A request timeout. Defaults to 120.

TYPE: int DEFAULT: None

retries

A number of retries. Defaults to 5.

TYPE: int DEFAULT: None

backoff_factor

A backoff factor. Defaults to 1.

TYPE: int DEFAULT: None

status_forcelist

A list of status codes to force a retry. Defaults to [502, 503, 504].

TYPE: list DEFAULT: None

pool_max_size

The maximum number of connections in the pool. Defaults to 120.

TYPE: int DEFAULT: None

Source code in psengine/sandbox/client.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def __init__(
    self,
    api_token: Annotated[
        str | None,
        Doc('The Sandbox API token. Defaults to RF_SANDBOX_TOKEN environment variable.'),
    ] = None,
    http_proxy: Annotated[str, Doc('An HTTP proxy URL.')] = None,
    https_proxy: Annotated[str, Doc('An HTTPS proxy URL.')] = None,
    verify: Annotated[
        str | bool,
        Doc('An SSL verification flag or path to CA bundle.'),
    ] = None,
    auth: Annotated[tuple[str, str], Doc('Basic Auth credentials.')] = None,
    cert: Annotated[str | tuple[str, str] | None, Doc('Client certificates.')] = None,
    timeout: Annotated[int, Doc('A request timeout. Defaults to 120.')] = None,
    retries: Annotated[int, Doc('A number of retries. Defaults to 5.')] = None,
    backoff_factor: Annotated[int, Doc('A backoff factor. Defaults to 1.')] = None,
    status_forcelist: Annotated[
        list, Doc('A list of status codes to force a retry. Defaults to [502, 503, 504].')
    ] = None,
    pool_max_size: Annotated[
        int, Doc('The maximum number of connections in the pool. Defaults to 120.')
    ] = None,
):
    """Recorded Future HTTP API client."""
    super().__init__(
        http_proxy=http_proxy,
        https_proxy=https_proxy,
        verify=verify,
        auth=auth,
        cert=cert,
        timeout=timeout,
        retries=retries,
        backoff_factor=backoff_factor,
        status_forcelist=status_forcelist,
        pool_max_size=pool_max_size,
    )

    self._api_token = api_token or self.config.sandbox_token.get_secret_value()
    if not self._api_token:
        raise ValueError('Missing Recorded Future Sandbox API token.')
    if not is_api_token_format_valid(self._api_token):
        raise ValueError(
            f'Invalid Sandbox API token: must match regex {SANDBOX_TOKEN_VALIDATION_REGEX}'
        )

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: dict | list[dict] | bytes | None = None,
    *,
    params: dict | None = None,
    headers: 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: dict | list[dict] | bytes | None DEFAULT: None

params

HTTP query parameters.

TYPE: dict | None DEFAULT: None

headers

If specified, overrides default headers and does not set the token.

TYPE: 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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
@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[dict | list[dict] | bytes | None, Doc('A request body.')] = None,
    *,
    params: Annotated[dict | None, Doc('HTTP query parameters.')] = None,
    headers: Annotated[
        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()

    if not isinstance(data, bytes):
        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 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
@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

request 

request(
    method: str,
    url: str,
    data: dict | list[dict] | bytes | None = None,
    *,
    params: dict | None = None,
    headers: dict | None = None,
    content_type_header: str | None = 'application/json',
    **kwargs,
) -> Response

Perform an HTTP request.

PARAMETER DESCRIPTION
method

An HTTP method, one of GET, PUT, POST, DELETE, HEAD, OPTIONS, PATCH.

TYPE: str

url

A URL to make the request to.

TYPE: str

data

A request body.

TYPE: dict | list[dict] | bytes | None DEFAULT: None

params

HTTP query parameters.

TYPE: dict | None DEFAULT: None

headers

If specified, it overrides default headers and does not set the token.

TYPE: dict | None DEFAULT: None

content_type_header

Content-Type header value.

TYPE: str | None DEFAULT: 'application/json'

RAISES DESCRIPTION
ValidationError

If method is not one of GET, PUT, POST, DELETE, HEAD, OPTIONS, PATCH.
RETURNS DESCRIPTION
Response

A requests.Response object.
Source code in psengine/sandbox/client.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@debug_call
@validate_call
def request(
    self,
    method: Annotated[
        str, Doc('An HTTP method, one of GET, PUT, POST, DELETE, HEAD, OPTIONS, PATCH.')
    ],
    url: Annotated[str, Doc('A URL to make the request to.')],
    data: Annotated[dict | list[dict] | bytes | None, Doc('A request body.')] = None,
    *,
    params: Annotated[dict | None, Doc('HTTP query parameters.')] = None,
    headers: Annotated[
        dict | None,
        Doc('If specified, it overrides default headers and does not set the token.'),
    ] = None,
    content_type_header: Annotated[
        str | None, Doc('Content-Type header value.')
    ] = 'application/json',
    **kwargs,
) -> Annotated[Response, Doc('A requests.Response object.')]:
    """Perform an HTTP request.

    Raises:
        ValidationError: If method is not one of GET, PUT, POST, DELETE, HEAD, OPTIONS, PATCH.
    """
    headers = headers or self._prepare_headers(content_type_header)

    return self.call(
        method=method,
        url=url,
        headers=headers,
        data=data,
        params=params,
        **kwargs,
    )

request_paged 

request_paged(
    method: str,
    url: str,
    data: dict | list[dict] | bytes | None = None,
    *,
    params: dict | None = None,
    headers: dict | None = None,
    content_type_header: str | None = 'application/json',
    max_results: int = DEFAULT_PAGE_LIMIT,
    results_per_page: int | None = Field(
        ge=1, le=MAXIMUM_SAMPLES, default=SAMPLES_PER_PAGE
    ),
    **kwargs,
) -> list

Perform a paged HTTP request, following next offsets until max_results is reached.

PARAMETER DESCRIPTION
method

An HTTP method, one of GET, PUT, POST, DELETE, HEAD, OPTIONS, PATCH.

TYPE: str

url

A URL to make the request to.

TYPE: str

data

A request body.

TYPE: dict | list[dict] | bytes | None DEFAULT: None

params

HTTP query parameters.

TYPE: dict | None DEFAULT: None

headers

If specified, it overrides default headers and does not set the token.

TYPE: dict | None DEFAULT: None

content_type_header

Content-Type header value.

TYPE: str | None DEFAULT: 'application/json'

results_per_page

The number of samples per page for pagination.

TYPE: int | None DEFAULT: Field(ge=1, le=MAXIMUM_SAMPLES, default=SAMPLES_PER_PAGE)

RAISES DESCRIPTION
ValidationError

If method is not one of GET, PUT, POST, DELETE, HEAD, OPTIONS, PATCH.
RETURNS DESCRIPTION
list

Result rows accumulated across all fetched pages.
Source code in psengine/sandbox/client.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
@debug_call
@validate_call
def request_paged(
    self,
    method: Annotated[
        str, Doc('An HTTP method, one of GET, PUT, POST, DELETE, HEAD, OPTIONS, PATCH.')
    ],
    url: Annotated[str, Doc('A URL to make the request to.')],
    data: Annotated[dict | list[dict] | bytes | None, Doc('A request body.')] = None,
    *,
    params: Annotated[dict | None, Doc('HTTP query parameters.')] = None,
    headers: Annotated[
        dict | None,
        Doc('If specified, it overrides default headers and does not set the token.'),
    ] = None,
    content_type_header: Annotated[
        str | None, Doc('Content-Type header value.')
    ] = 'application/json',
    max_results: int = DEFAULT_PAGE_LIMIT,
    results_per_page: Annotated[
        int | None, Doc('The number of samples per page for pagination.')
    ] = Field(ge=1, le=MAXIMUM_SAMPLES, default=SAMPLES_PER_PAGE),
    **kwargs,
) -> Annotated[list, Doc('Result rows accumulated across all fetched pages.')]:
    """Perform a paged HTTP request, following `next` offsets until `max_results` is reached.

    Raises:
        ValidationError: If method is not one of GET, PUT, POST, DELETE, HEAD, OPTIONS, PATCH.
    """
    headers = headers or self._prepare_headers(content_type_header)
    # Copy so our `limit`/`offset` bookkeeping never mutates the caller's dict.
    params = dict(params or {})
    all_results: list = []

    # Trim `limit` to remaining headroom so the last page doesn't overfetch.
    params['limit'] = min(results_per_page, max_results - len(all_results))
    response = self.request(
        method=method,
        url=url,
        headers=headers,
        data=data,
        params=params,
        **kwargs,
    )

    try:
        json_response = response.json()
    except JSONDecodeError:
        self.log.debug(f'Paged request does not contain valid JSON:\n{response.text}')
        raise

    try:
        page = json_response['data'] or []
    except KeyError:
        self.log.debug(f'Paged request does not contain `data` JSON key:\n{response.text}')
        raise

    all_results.extend(page)
    offset = json_response.get('next')
    prev_offset = None

    while offset and offset != prev_offset and len(all_results) < max_results:
        params['offset'] = offset
        params['limit'] = min(results_per_page, max_results - len(all_results))
        response = self.request(
            method=method,
            url=url,
            headers=headers,
            data=data,
            params=params,
            **kwargs,
        )
        json_response = response.json()
        page = json_response.get('data', [])

        if not page:
            # The last page is always empty
            self.log.debug('Paged request returned an empty `data` page; stopping.')
            break

        all_results.extend(page)
        prev_offset = offset
        offset = json_response.get('next')

    return all_results[:max_results]

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 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
@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())

is_api_token_format_valid 

is_api_token_format_valid(token: str) -> bool

Check if the token format is valid.

The function performs a simple regex check but does not validate the token against the API.

PARAMETER DESCRIPTION
token

A Recorded Future API token.

TYPE: str

RETURNS DESCRIPTION
bool

True if the token format is valid, False otherwise.
Source code in psengine/sandbox/client.py 
28
29
30
31
32
33
34
35
36
@validate_call
def is_api_token_format_valid(
    token: Annotated[str, Doc('A Recorded Future API token.')],
) -> Annotated[bool, Doc('True if the token format is valid, False otherwise.')]:
    """Check if the token format is valid.

    The function performs a simple regex check but does not validate the token against the API.
    """
    return re.fullmatch(SANDBOX_TOKEN_VALIDATION_REGEX, token) is not None