Skip to content

HumeBatchClient

Bases: ClientBase

Batch API client.

Example
from hume import HumeBatchClient
from hume.models.config import FaceConfig

client = HumeBatchClient("<your-api-key>")
urls = ["<your-image-url>"]
config = FaceConfig(identify_faces=True)
job = client.submit_job(urls, [configs])

print(job)
print("Running...")

result = job.await_complete()
result.download_predictions("predictions.json")

print("Predictions downloaded!")
Source code in hume/_batch/hume_batch_client.py
class HumeBatchClient(ClientBase):
    """Batch API client.

    Example:
        ```python
        from hume import HumeBatchClient
        from hume.models.config import FaceConfig

        client = HumeBatchClient("<your-api-key>")
        urls = ["<your-image-url>"]
        config = FaceConfig(identify_faces=True)
        job = client.submit_job(urls, [configs])

        print(job)
        print("Running...")

        result = job.await_complete()
        result.download_predictions("predictions.json")

        print("Predictions downloaded!")
        ```
    """

    _DEFAULT_API_TIMEOUT = 10

    def __init__(self, api_key: str, *args: Any, **kwargs: Any):
        """Construct a HumeBatchClient.

        Args:
            api_key (str): Hume API key.
        """
        super().__init__(api_key, *args, **kwargs)

    @classmethod
    def get_api_type(cls) -> ApiType:
        """Get the ApiType of the client.

        Returns:
            ApiType: API type of the client.
        """
        return ApiType.BATCH

    def get_job_result(self, job_id: str) -> BatchJobResult:
        """Get the result of the batch job.

        Args:
            job_id (str): Job ID.

        Raises:
            HumeClientException: If the job result cannot be loaded.

        Returns:
            BatchJobResult: Batch job result.
        """
        endpoint = self._construct_endpoint(f"jobs/{job_id}")
        response = requests.get(
            endpoint,
            timeout=self._DEFAULT_API_TIMEOUT,
            headers=self._get_client_headers(),
        )

        try:
            body = response.json()
        except json.JSONDecodeError:
            # pylint: disable=raise-missing-from
            raise HumeClientException("Unexpected error when getting job result")

        if "message" in body and body["message"] == "job not found":
            raise HumeClientException(f"Could not find a job with ID {job_id}")

        return BatchJobResult.from_response(body)

    def get_job(self, job_id: str) -> BatchJob:
        """Rehydrate a job based on a Job ID.

        Args:
            job_id (str): ID of the job to rehydrate.

        Returns:
            BatchJob: Job associated with the given ID.
        """
        return BatchJob(self, job_id)

    def submit_job(self, urls: List[str], configs: List[ModelConfigBase]) -> BatchJob:
        """Submit a job for batch processing.

        Note: Only one config per model type should be passed.
            If more than one config is passed for a given model type, only the last config will be used.

        Args:
            urls (List[str]): List of URLs to media files to be processed.
            configs (List[ModelConfigBase]): List of model config objects to run on each media URL.

        Returns:
            BatchJob: The `BatchJob` representing the batch computation.
        """
        request = self._get_request(configs, urls)
        return self._submit_job_from_request(request)

    @classmethod
    def _get_request(cls, configs: List[ModelConfigBase], urls: List[str]) -> Dict[str, Any]:
        return {
            "urls": urls,
            "models": serialize_configs(configs),
        }

    def _submit_job_from_request(self, request_body: Any) -> BatchJob:
        """Start a job for batch processing by passing a JSON request body.

        This request body should match the request body used by the batch API,
        including both the list of URLs and the models configuration.

        Args:
            request_body (Any): JSON request body to be passed to the batch API.

        Raises:
            HumeClientException: If the batch job fails to start.

        Returns:
            BatchJob: A `BatchJob` that wraps the batch computation.
        """
        endpoint = self._construct_endpoint("jobs")
        response = requests.post(
            endpoint,
            json=request_body,
            timeout=self._DEFAULT_API_TIMEOUT,
            headers=self._get_client_headers(),
        )

        try:
            body = response.json()
        except json.decoder.JSONDecodeError:
            # pylint: disable=raise-missing-from
            raise HumeClientException(f"Failed batch request: {response.text}")

        if "job_id" not in body:
            if "fault" in body and "faultstring" in body["fault"]:
                fault = body["fault"]
                fault_string = fault["faultstring"]
                if "detail" in fault and "errorcode" in fault["detail"]:
                    detail = fault["detail"]
                    error_code = detail["errorcode"]
                    if "InvalidApiKey" in error_code:
                        raise HumeClientException("HumeBatchClient initialized with invalid API key.")
                    raise HumeClientException(f"Could not start batch job: {error_code}: {fault_string}")
                raise HumeClientException(f"Could not start batch job: {fault_string}")
            raise HumeClientException(f"Unexpected error when starting batch job: {body}")

        return BatchJob(self, body["job_id"])

__init__(api_key, *args, **kwargs)

Construct a HumeBatchClient.

Parameters:

Name Type Description Default
api_key str

Hume API key.

required
Source code in hume/_batch/hume_batch_client.py
def __init__(self, api_key: str, *args: Any, **kwargs: Any):
    """Construct a HumeBatchClient.

    Args:
        api_key (str): Hume API key.
    """
    super().__init__(api_key, *args, **kwargs)

get_api_type() classmethod

Get the ApiType of the client.

Returns:

Name Type Description
ApiType ApiType

API type of the client.

Source code in hume/_batch/hume_batch_client.py
@classmethod
def get_api_type(cls) -> ApiType:
    """Get the ApiType of the client.

    Returns:
        ApiType: API type of the client.
    """
    return ApiType.BATCH

get_job(job_id)

Rehydrate a job based on a Job ID.

Parameters:

Name Type Description Default
job_id str

ID of the job to rehydrate.

required

Returns:

Name Type Description
BatchJob BatchJob

Job associated with the given ID.

Source code in hume/_batch/hume_batch_client.py
def get_job(self, job_id: str) -> BatchJob:
    """Rehydrate a job based on a Job ID.

    Args:
        job_id (str): ID of the job to rehydrate.

    Returns:
        BatchJob: Job associated with the given ID.
    """
    return BatchJob(self, job_id)

get_job_result(job_id)

Get the result of the batch job.

Parameters:

Name Type Description Default
job_id str

Job ID.

required

Raises:

Type Description
HumeClientException

If the job result cannot be loaded.

Returns:

Name Type Description
BatchJobResult BatchJobResult

Batch job result.

Source code in hume/_batch/hume_batch_client.py
def get_job_result(self, job_id: str) -> BatchJobResult:
    """Get the result of the batch job.

    Args:
        job_id (str): Job ID.

    Raises:
        HumeClientException: If the job result cannot be loaded.

    Returns:
        BatchJobResult: Batch job result.
    """
    endpoint = self._construct_endpoint(f"jobs/{job_id}")
    response = requests.get(
        endpoint,
        timeout=self._DEFAULT_API_TIMEOUT,
        headers=self._get_client_headers(),
    )

    try:
        body = response.json()
    except json.JSONDecodeError:
        # pylint: disable=raise-missing-from
        raise HumeClientException("Unexpected error when getting job result")

    if "message" in body and body["message"] == "job not found":
        raise HumeClientException(f"Could not find a job with ID {job_id}")

    return BatchJobResult.from_response(body)

submit_job(urls, configs)

Submit a job for batch processing.

Only one config per model type should be passed.

If more than one config is passed for a given model type, only the last config will be used.

Parameters:

Name Type Description Default
urls List[str]

List of URLs to media files to be processed.

required
configs List[ModelConfigBase]

List of model config objects to run on each media URL.

required

Returns:

Name Type Description
BatchJob BatchJob

The BatchJob representing the batch computation.

Source code in hume/_batch/hume_batch_client.py
def submit_job(self, urls: List[str], configs: List[ModelConfigBase]) -> BatchJob:
    """Submit a job for batch processing.

    Note: Only one config per model type should be passed.
        If more than one config is passed for a given model type, only the last config will be used.

    Args:
        urls (List[str]): List of URLs to media files to be processed.
        configs (List[ModelConfigBase]): List of model config objects to run on each media URL.

    Returns:
        BatchJob: The `BatchJob` representing the batch computation.
    """
    request = self._get_request(configs, urls)
    return self._submit_job_from_request(request)