parsl/dataflow/dflow.py
def sanitize_and_wrap(self, task_id, args, kwargs):
        """This function should be called **ONLY** when all the futures we track have been resolved.

        If the user hid futures a level below, we will not catch
        it, and will (most likely) result in a type error.

        Args:
             task_id (uuid str) : Task id
             func (Function) : App function
             args (List) : Positional args to app function
             kwargs (Dict) : Kwargs to app function

        Return:
             partial function evaluated with all dependencies in  args, kwargs and kwargs['inputs'] evaluated.

        """
        dep_failures = []

        # Replace item in args
        new_args = []
        for dep in args:
            if isinstance(dep, Future):
                try:
                    new_args.extend([dep.result()])
                except Exception as e:
                    if self.tasks[dep.tid]['status'] in FINAL_FAILURE_STATES:
                        dep_failures.extend([e])
            else:
                new_args.extend([dep])

        # Check for explicit kwargs ex, fu_1=<fut>
        for key in kwargs:
            dep = kwargs[key]
            if isinstance(dep, Future):
                try:
                    kwargs[key] = dep.result()
                except Exception as e:
                    if self.tasks[dep.tid]['status'] in FINAL_FAILURE_STATES:
                        dep_failures.extend([e])

        # Check for futures in inputs=[<fut>...]
        if 'inputs' in kwargs:
            new_inputs = []
            for dep in kwargs['inputs']:
                if isinstance(dep, Future):
                    try:
                        new_inputs.extend([dep.result()])
                    except Exception as e:
                        if self.tasks[dep.tid]['status'] in FINAL_FAILURE_STATES:
                            dep_failures.extend([e])

                else:
                    new_inputs.extend([dep])
            kwargs['inputs'] = new_inputs

        return new_args, kwargs, dep_failures
def sanitize_and_wrap(self, task_id, args, kwargs):
        """This function should be called **ONLY** when all the futures we track have been resolved.

        If the user hid futures a level below, we will not catch
        it, and will (most likely) result in a type error.

        Args:
             task_id (uuid str) : Task id
             func (Function) : App function
             args (List) : Positional args to app function
             kwargs (Dict) : Kwargs to app function

        Return:
             partial function evaluated with all dependencies in  args, kwargs and kwargs['inputs'] evaluated.

        """
        dep_failures = []

        # Replace item in args
        new_args = []
        for dep in args:
            if isinstance(dep, Future):
                try:
                    new_args.extend([dep.result()])
                except Exception as e:
                    if self.tasks[dep.tid]['status'] in FINAL_FAILURE_STATES:
                        dep_failures.extend([e])
            else:
                new_args.extend([dep])

        # Check for explicit kwargs ex, fu_1=<fut>
        for key in kwargs:
            dep = kwargs[key]
            if isinstance(dep, Future):
                try:
                    kwargs[key] = dep.result()
                except Exception as e:
                    if self.tasks[dep.tid]['status'] in FINAL_FAILURE_STATES:
                        dep_failures.extend([e])

        # Check for futures in inputs=[<fut>...]
        if 'inputs' in kwargs:
            new_inputs = []
            for dep in kwargs['inputs']:
                if isinstance(dep, Future):
                    try:
                        new_inputs.extend([dep.result()])
                    except Exception as e:
                        if self.tasks[dep.tid]['status'] in FINAL_FAILURE_STATES:
                            dep_failures.extend([e])

                else:
                    new_inputs.extend([dep])
            kwargs['inputs'] = new_inputs

        return new_args, kwargs, dep_failures
def sanitize_and_wrap ( self , task_id , args , kwargs ) : dep_failures = [ ] # Replace item in args new_args = [ ] for dep in args : if isinstance ( dep , Future ) : try : new_args . extend ( [ dep . result ( ) ] ) except Exception as e : if self . tasks [ dep . tid ] [ 'status' ] in FINAL_FAILURE_STATES : dep_failures . extend ( [ e ] ) else : new_args . extend ( [ dep ] ) # Check for explicit kwargs ex, fu_1=<fut> for key in kwargs : dep = kwargs [ key ] if isinstance ( dep , Future ) : try : kwargs [ key ] = dep . result ( ) except Exception as e : if self . tasks [ dep . tid ] [ 'status' ] in FINAL_FAILURE_STATES : dep_failures . extend ( [ e ] ) # Check for futures in inputs=[<fut>...] if 'inputs' in kwargs : new_inputs = [ ] for dep in kwargs [ 'inputs' ] : if isinstance ( dep , Future ) : try : new_inputs . extend ( [ dep . result ( ) ] ) except Exception as e : if self . tasks [ dep . tid ] [ 'status' ] in FINAL_FAILURE_STATES : dep_failures . extend ( [ e ] ) else : new_inputs . extend ( [ dep ] ) kwargs [ 'inputs' ] = new_inputs return new_args , kwargs , dep_failures
This function should be called **ONLY** when all the futures we track have been resolved.

        If the user hid futures a level below, we will not catch
        it, and will (most likely) result in a type error.

        Args:
             task_id (uuid str) : Task id
             func (Function) : App function
             args (List) : Positional args to app function
             kwargs (Dict) : Kwargs to app function

        Return:
             partial function evaluated with all dependencies in  args, kwargs and kwargs['inputs'] evaluated.
This function should be called ** ONLY ** when all the futures we track have been resolved .