Executors

RequestMethod

Bases: Enum

Enum for the request method.

Attributes:

Name	Type	Description
GET		The GET method.
POST		The POST method.

Source code in inference_sdk/http/utils/executors.py

class RequestMethod(Enum):
    """Enum for the request method.

    Attributes:
        GET: The GET method.
        POST: The POST method.
    """

    GET = "get"
    POST = "post"

execute_requests_packages(requests_data, request_method, max_concurrent_requests)

Execute a list of requests in parallel.

Parameters:

Name	Type	Description	Default
requests_data	List[RequestData]	The list of requests to execute.	required
request_method	RequestMethod	The method to use for the requests.	required
max_concurrent_requests	int	The maximum number of concurrent requests.	required

Returns:

Type	Description
List[Response]	The list of responses.

Source code in inference_sdk/http/utils/executors.py

def execute_requests_packages(
    requests_data: List[RequestData],
    request_method: RequestMethod,
    max_concurrent_requests: int,
) -> List[Response]:
    """Execute a list of requests in parallel.

    Args:
        requests_data: The list of requests to execute.
        request_method: The method to use for the requests.
        max_concurrent_requests: The maximum number of concurrent requests.

    Returns:
        The list of responses.
    """
    requests_data_packages = make_batches(
        iterable=requests_data,
        batch_size=max_concurrent_requests,
    )
    results = []
    all_request_data = []
    for requests_data_package in requests_data_packages:
        responses = make_parallel_requests(
            requests_data=requests_data_package,
            request_method=request_method,
        )
        results.extend(responses)
        all_request_data.extend(requests_data_package)
    _collect_remote_processing_times(results, all_request_data)
    for response in results:
        api_key_safe_raise_for_status(response=response)
    return results

execute_requests_packages_async(requests_data, request_method, max_concurrent_requests) async

Execute a list of requests in parallel asynchronously.

Parameters:

Name	Type	Description	Default
requests_data	List[RequestData]	The list of requests to execute.	required
request_method	RequestMethod	The method to use for the requests.	required
max_concurrent_requests	int	The maximum number of concurrent requests.	required

Returns:

Type	Description
List[Union[dict, bytes]]	The list of responses.

Source code in inference_sdk/http/utils/executors.py

async def execute_requests_packages_async(
    requests_data: List[RequestData],
    request_method: RequestMethod,
    max_concurrent_requests: int,
) -> List[Union[dict, bytes]]:
    """Execute a list of requests in parallel asynchronously.

    Args:
        requests_data: The list of requests to execute.
        request_method: The method to use for the requests.
        max_concurrent_requests: The maximum number of concurrent requests.

    Returns:
        The list of responses.
    """
    requests_data_packages = make_batches(
        iterable=requests_data,
        batch_size=max_concurrent_requests,
    )
    results = []
    for requests_data_package in requests_data_packages:
        responses = await make_parallel_requests_async(
            requests_data=requests_data_package,
            request_method=request_method,
        )
        results.extend(responses)
    return results

make_parallel_requests(requests_data, request_method)

Execute a list of requests in parallel.

Parameters:

Name	Type	Description	Default
requests_data	List[RequestData]	The list of requests to execute.	required
request_method	RequestMethod	The method to use for the requests.	required

Returns:

Type	Description
List[Response]	The list of responses.

Source code in inference_sdk/http/utils/executors.py

def make_parallel_requests(
    requests_data: List[RequestData],
    request_method: RequestMethod,
) -> List[Response]:
    """Execute a list of requests in parallel.

    Args:
        requests_data: The list of requests to execute.
        request_method: The method to use for the requests.

    Returns:
        The list of responses.
    """
    workers = len(requests_data)
    make_request_closure = partial(make_request, request_method=request_method)
    with ThreadPoolExecutor(max_workers=workers) as executor:
        return list(executor.map(make_request_closure, requests_data))

make_parallel_requests_async(requests_data, request_method) async

Execute a list of requests in parallel asynchronously.

Parameters:

Name	Type	Description	Default
requests_data	List[RequestData]	The list of requests to execute.	required
request_method	RequestMethod	The method to use for the requests.	required

Returns:

Type	Description
List[Union[dict, bytes]]	The list of responses.

Source code in inference_sdk/http/utils/executors.py

async def make_parallel_requests_async(
    requests_data: List[RequestData],
    request_method: RequestMethod,
) -> List[Union[dict, bytes]]:
    """Execute a list of requests in parallel asynchronously.

    Args:
        requests_data: The list of requests to execute.
        request_method: The method to use for the requests.

    Returns:
        The list of responses.
    """
    async with aiohttp.ClientSession() as session:
        make_request_closure = partial(
            make_request_async,
            request_method=request_method,
            session=session,
        )
        coroutines = [make_request_closure(data) for data in requests_data]
        responses = list(await asyncio.gather(*coroutines))
        return [r[1] for r in responses]

make_request(request_data, request_method)

Make a request to the API.

Parameters:

Name	Type	Description	Default
request_data	RequestData	The request data.	required
request_method	RequestMethod	The method to use for the request.	required

Returns:

Type	Description
Response	The response from the API.

Source code in inference_sdk/http/utils/executors.py

@backoff.on_predicate(
    backoff.constant,
    predicate=lambda r: r.status_code in RETRYABLE_STATUS_CODES,
    max_tries=3,
    interval=1,
    backoff_log_level=logging.DEBUG,
    giveup_log_level=logging.DEBUG,
)
@backoff.on_exception(
    backoff.constant,
    exception=ConnectionError,
    max_tries=3,
    interval=1,
    backoff_log_level=logging.DEBUG,
    giveup_log_level=logging.DEBUG,
)
def make_request(request_data: RequestData, request_method: RequestMethod) -> Response:
    """Make a request to the API.

    Args:
        request_data: The request data.
        request_method: The method to use for the request.

    Returns:
        The response from the API.
    """
    method = requests.get if request_method is RequestMethod.GET else requests.post
    return method(
        request_data.url,
        headers=request_data.headers,
        params=request_data.parameters,
        data=request_data.data,
        json=request_data.payload,
    )

make_request_async(request_data, request_method, session) async

Make a request to the API asynchronously.

Parameters:

Name	Type	Description	Default
request_data	RequestData	The request data.	required
request_method	RequestMethod	The method to use for the request.	required
session	ClientSession	The session to use for the request.	required

Returns:

Type	Description
Tuple[int, Union[bytes, dict]]	The response from the API.

Source code in inference_sdk/http/utils/executors.py

@backoff.on_predicate(
    backoff.constant,
    predicate=lambda r: r[0] in RETRYABLE_STATUS_CODES,
    max_tries=3,
    interval=1,
    on_giveup=raise_client_error,
    backoff_log_level=logging.DEBUG,
    giveup_log_level=logging.DEBUG,
)
@backoff.on_exception(
    backoff.constant,
    exception=ClientConnectionError,
    max_tries=3,
    interval=1,
    backoff_log_level=logging.DEBUG,
    giveup_log_level=logging.DEBUG,
)
async def make_request_async(
    request_data: RequestData,
    request_method: RequestMethod,
    session: aiohttp.ClientSession,
) -> Tuple[int, Union[bytes, dict]]:
    """Make a request to the API asynchronously.

    Args:
        request_data: The request data.
        request_method: The method to use for the request.
        session: The session to use for the request.

    Returns:
        The response from the API.
    """
    method = session.get if request_method is RequestMethod.GET else session.post
    parameters_serialised = None
    if request_data.parameters is not None:
        parameters_serialised = {
            name: (
                str(value)
                if not issubclass(type(value), list)
                else [str(e) for e in value]
            )
            for name, value in request_data.parameters.items()
        }
    async with method(
        request_data.url,
        headers=request_data.headers,
        params=parameters_serialised,
        data=request_data.data,
        json=request_data.payload,
    ) as response:
        try:
            response_data = await response.json()
        except:
            response_data = await response.read()
        if response_is_not_retryable_error(response=response):
            response.raise_for_status()
        return response.status, response_data

raise_client_error(details)

Raise a client error.

Parameters:

Name	Type	Description	Default
details	dict	The details of the error.	required

Source code in inference_sdk/http/utils/executors.py

def raise_client_error(details: dict) -> None:
    """Raise a client error.

    Args:
        details: The details of the error.
    """
    status_code = details["value"][0]
    request_data = details["kwargs"]["request_data"]
    raise ClientResponseError(
        request_info=RequestInfo(
            url=request_data.url,
            method="POST",
            headers={},
        ),
        history=(),
        status=status_code,
    )

response_is_not_retryable_error(response)

Check if the response is not a retryable error.

Parameters:

Name	Type	Description	Default
response	ClientResponse	The response to check.	required

Returns:

Type	Description
bool	True if the response is not a retryable error, False otherwise.

Source code in inference_sdk/http/utils/executors.py

def response_is_not_retryable_error(response: ClientResponse) -> bool:
    """Check if the response is not a retryable error.

    Args:
        response: The response to check.

    Returns:
        True if the response is not a retryable error, False otherwise.
    """
    return response.status != 200 and response.status not in RETRYABLE_STATUS_CODES