def apply()

in pythonflow/core.py [0:0]

• 13 lines of code
• 6 McCabe index (conditional complexity)

    def apply(self, fetches, context=None, *, callback=None, **kwargs):
        """
        Evaluate one or more operations given a context.

        .. note::
           This function modifies the context in place. Use :code:`context=context.copy()` to avoid
           the context being modified.

        Parameters
        ----------
        fetches : list[str or Operation] or str or Operation
            One or more `Operation` instances or names to evaluate.
        context : dict or None
            Context in which to evaluate the operations.
        callback : callable or None
            Callback to be evaluated when an operation is evaluated.
        kwargs : dict
            Additional context information keyed by variable name.

        Returns
        -------
        values : tuple[object]
            Output of the operations given the context.

        Raises
        ------
        ValueError
            If `fetches` is not an `Operation` instance, operation name, or a sequence thereof.
        ValueError
            If `context` is not a mapping.
        """
        if isinstance(fetches, (str, Operation)):
            fetches = [fetches]
            single = True
        elif isinstance(fetches, collections.abc.Sequence):
            single = False
        else:
            raise ValueError("`fetches` must be an `Operation` instance, operation name, or a "
                             "sequence thereof.")

        fetches = [self.normalize_operation(operation) for operation in fetches]
        context = self.normalize_context(context, **kwargs)
        values = [fetch.evaluate_operation(fetch, context, callback=callback) for fetch in fetches]
        return values[0] if single else tuple(values)