Limiting parallel creation of resources in a loop with the Python bindings

iris.csv

parallel.py

import bigml

from bigml.api import BigML, get_resource_type


inprogress = [] # list of resources that are being created
NUMBER_OF_RESOURCES = 10 # the total number of resources to be created
MAX_PARALLEL_RESOURCES = 2 # max. number of resources to be created in parallel
SOURCE_FILE = 'iris.csv'

def main():
    """
    """
    api = BigML(dev_mode=True)
    project = api.create_project({"name": "Sources in parallel"})
    for i in range(0, NUMBER_OF_RESOURCES):
        # this function checks whether the list accepts a new resource for
        # creation
        wait_for_available_tasks(inprogress, MAX_PARALLEL_RESOURCES,
                                 api)
        # now you can create a new resource
        source_args = {"project": project['resource']}
        source = api.create_source(SOURCE_FILE, args=source_args)
        # checking for errors
        source_id = check_resource_error(source,
                                         "Failed to create source.")
        # storing the source_id in the inprogress list
        inprogress.append(source_id)


# this is an auxiiliary function which waits for a resource to be finished
def wait_for_available_tasks(inprogress, max_parallel, api, wait_step=2):
    """According to the max_parallel number of parallel resources to be
       created, when the number of in progress resources reaches the limit,
       it checks the ones in inprogress to see if there's a
       FINISHED or FAULTY resource. If found, it is removed from the
       inprogress list and returns to allow another one to be created.

    """

    check_kwargs = {"retries": 0, "query_string": "full=false", "api": api}
    while len(inprogress) == max_parallel:
        for j in range(0, len(inprogress)):
            try:
                resource_type = get_resource_type(inprogress[j])
                ready = check_resource(inprogress[j], **check_kwargs)
                status = bigml.api.get_status(ready)
                if status['code'] == bigml.api.FINISHED:
                    del inprogress[j]
                    return
                elif status['code'] == bigml.api.FAULTY:
                    raise ValueError(status['message'])
            except ValueError, exception:
                sys.exit("Failed to get a finished %s: %s" %
                         (resource_type, str(exception)))
        time.sleep(max_parallel * wait_step)


def check_resource_error(resource, message):
    """If a given resource is faulty, or some other error has occured, exits.
       Returns the resource id otherwise.

    """
    if ('error' in resource and resource['error'] or
            bigml.api.get_status(resource)['code'] == bigml.api.FAULTY):
        if ('status' in resource['error'] and
                'message' in resource['error']['status']):
            error_message = resource['error']['status']['message']
        sys.exit("%s%s" % (message, error_message))
    return bigml.api.get_resource_id(resource)

def check_resource(*args, **kwargs):
    """Wrapper to catch errors in resource retrieval

    """
    try:
        result = bigml.api.check_resource(*args, **kwargs)
        return result
    except ValueError, exc:
        sys.exit("\nFailed to obtain a finished resource:\n%s." % str(exc))


if __name__ == "__main__":
    main()