API Reference

Auto-generated from the source docstrings. For task-writing guidance start with the Guide; this page is the exhaustive symbol reference.

Top-level API

The public API re-exported from oryxflow — run, preview, Workflow, WorkflowMulti, the requires / inherits decorators, the Parameter types, and the invalidate_* / enable_* helpers.

Engine — oryxflow.core

oryxflow.core

Mini-engine: a small, self-contained execution engine for oryxflow. Provides the Task base class, target bases, flatten/getpaths, deterministic task ids, the inherits/requires decorators, find_deps and a sequential build.

The execution engine is sequential: the DAG is run in dependency order in-process. The workers argument is accepted for API compatibility but ignored.

Register

Bases: type

Minimal metaclass providing a class-level task_family property and instance memoization.

A property defined directly on class:Task is only invoked for instances; reading SomeTaskClass.task_family needs this metaclass property (used eg by WorkflowMulti).

__call__ memoizes instances so two Cls(**same_params) calls return the identical object (and __init__ runs only on the first), preserving the instance identity that Workflow's path/flow propagation depends on.

Source code in oryxflow/core.py

class Register(type):
    """
    Minimal metaclass providing a class-level ``task_family`` property and instance memoization.

    A property defined directly on :py:class:`Task` is only invoked for instances; reading
    ``SomeTaskClass.task_family`` needs this metaclass property (used eg by ``WorkflowMulti``).

    ``__call__`` memoizes instances so two ``Cls(**same_params)`` calls return the identical
    object (and ``__init__`` runs only on the first), preserving the instance identity that
    ``Workflow``'s path/flow propagation depends on.
    """

    # Names oryxflow already uses around a task, which a Parameter of the same name shadows --
    # silently, in both cases, which is why the name is rejected where it is WRITTEN rather than
    # at the far-away call that breaks:
    #   path/flows  the engine sets on instances itself (see TaskData.__init__, which takes both
    #               as keyword-only args), so the Parameter never receives the value you pass --
    #               it keeps its default, and that default becomes the data directory.
    #   cls/derive  are clone()/requires_grid() arguments, so clone(cls=Other) or a fan-out over
    #               a parameter of that name binds to the argument instead of the parameter.
    RESERVED_PARAM_NAMES = {                       # name: (what it is reserved for, rename to)
        'path': ("the flow's data directory", 'file'),
        'flows': ('flows attached with attach_flow()', 'flow_name'),
        'cls': ('the target class in clone() / requires_grid()', 'model_cls'),
        'derive': ("requires_grid()'s per-branch derived parameters", 'derive_features'),
    }

    def __init__(cls, name, bases, namespace):
        super().__init__(name, bases, namespace)
        for attr, (what, suggestion) in Register.RESERVED_PARAM_NAMES.items():
            if isinstance(namespace.get(attr), Parameter):
                raise ValueError(
                    "{}: '{}' cannot be a Parameter -- the name is reserved for {}. "
                    "Rename it (eg '{}').".format(name, attr, what, suggestion))

    @property
    def task_family(cls):
        return cls.__name__

    def __call__(cls, *args, **kwargs):
        try:
            params = cls.get_params()
            param_values = cls.get_param_values(params, args, kwargs)
            param_objs = dict(params)
            key = (cls, tuple((n, param_objs[n].serialize(v)) for n, v in param_values))
            hash(key)
        except Exception:
            # On any trouble building a stable key, fall back to a fresh (uncached) instance.
            return super().__call__(*args, **kwargs)

        inst = _instance_cache.get(key)
        if inst is None:
            inst = super().__call__(*args, **kwargs)
            _instance_cache[key] = inst
        return inst

Task

Base class for all oryxflow tasks.

Subclasses declare class:~oryxflow.parameter.Parameter members and override meth:run, meth:requires and meth:output.

Source code in oryxflow/core.py

class Task(metaclass=Register):
    """
    Base class for all oryxflow tasks.

    Subclasses declare :py:class:`~oryxflow.parameter.Parameter` members and override
    :py:meth:`run`, :py:meth:`requires` and :py:meth:`output`.
    """

    # Explicit code identity token (str or int), OPT-IN: when set it is the authority
    # for this task's own logic -- bump it to rerun this task and everything downstream
    # (code edits without a bump only warn). None (the default) with
    # settings.code_version_auto (the default) derives the token automatically from the
    # AST hash of the task's module + repo-local imports, so logic edits rerun with no
    # attribute to maintain; with code_version_auto=False, None leaves caching purely
    # parameter-based.
    code_version = None

    @classmethod
    def get_params(cls):
        """Return all ``(name, Parameter)`` pairs for this task, in declaration order."""
        params = []
        for param_name in dir(cls):
            param_obj = getattr(cls, param_name)
            if not isinstance(param_obj, Parameter):
                continue
            params.append((param_name, param_obj))

        params.sort(key=lambda t: t[1]._counter)
        return params

    @classmethod
    def get_param_names(cls, include_significant=False):
        """Return parameter names. ``include_significant=True`` returns all params."""
        return [name for name, p in cls.get_params() if include_significant or p.significant]

    @classmethod
    def get_param_values(cls, params, args, kwargs):
        """
        Resolve parameter values from positional args, kwargs and defaults.

        :param params: list of ``(param_name, Parameter)``
        :param args: positional arguments
        :param kwargs: keyword arguments
        :returns: list of ``(name, value)`` tuples, one per parameter
        """
        result = {}
        params_dict = dict(params)
        task_family = cls.get_task_family()
        exc_desc = '%s[args=%s, kwargs=%s]' % (task_family, args, kwargs)

        # Positional arguments.
        positional_params = [(n, p) for n, p in params if p.positional]
        for i, arg in enumerate(args):
            if i >= len(positional_params):
                raise UnknownParameterException(
                    '%s: takes at most %d parameters (%d given)' % (exc_desc, len(positional_params), len(args)))
            param_name, param_obj = positional_params[i]
            result[param_name] = param_obj.normalize(arg)

        # Keyword arguments.
        for param_name, arg in kwargs.items():
            if param_name in result:
                raise DuplicateParameterException(
                    '%s: parameter %s was already set as a positional parameter' % (exc_desc, param_name))
            if param_name not in params_dict:
                raise UnknownParameterException('%s: unknown parameter %s' % (exc_desc, param_name))
            result[param_name] = params_dict[param_name].normalize(arg)

        # Defaults for anything still unset.
        for param_name, param_obj in params:
            if param_name not in result:
                if not param_obj.has_task_value():
                    raise MissingParameterException(
                        "%s: requires the '%s' parameter to be set" % (exc_desc, param_name))
                result[param_name] = param_obj.task_value()

        return [(param_name, result[param_name]) for param_name, param_obj in params]

    def __init__(self, *args, **kwargs):
        params = self.get_params()
        param_values = self.get_param_values(params, args, kwargs)

        for key, value in param_values:
            setattr(self, key, value)

        self.param_kwargs = dict(param_values)

        self.task_id = task_id_str(self.get_task_family(), self.to_str_params(only_significant=True))
        self.__hash = hash(self.task_id)

    @property
    def task_family(self):
        """The task family (class name) of this task instance."""
        return self.__class__.__name__

    @property
    def logger(self):
        """Contextual logger for task authors; auto-tagged with task identity.

        Lives in the ``oryxflow`` namespace, so it is silent until
        ``oryxflow.enable_logging()`` is called.
        """
        if getattr(self, "_logger", None) is None:
            self._logger = TaskLogger(task_id=self.task_id, task_family=self.task_family)
        return self._logger

    @classmethod
    def get_task_family(cls):
        """The task family (class name) for this class."""
        return cls.__name__

    def to_str_params(self, only_significant=False, only_public=False):
        """
        Convert parameters to a ``{name: serialized_value}`` dict.

        :param bool only_significant: only include parameters marked significant.
        :param bool only_public: accepted for API compatibility; visibility is not modeled.
        """
        params_str = {}
        params = dict(self.get_params())
        for param_name, param_value in self.param_kwargs.items():
            if (not only_significant) or params[param_name].significant:
                params_str[param_name] = params[param_name].serialize(param_value)
        return params_str

    def clone(self, cls=None, **kwargs):
        """
        Create a new task instance from this one, overriding some args.

        Parameters common to this task and ``cls`` are carried over; ``kwargs`` take precedence.
        """
        if cls is None:
            cls = self.__class__

        new_k = {}
        for param_name, param_class in cls.get_params():
            if param_name in kwargs:
                new_k[param_name] = kwargs[param_name]
            elif hasattr(self, param_name):
                new_k[param_name] = getattr(self, param_name)

        return cls(**new_k)

    def requires_grid(self, cls=None, derive=None, **grid):
        """
        Build the ``requires()`` dict for depending on ``cls`` once per value -- the runtime
        form of the ``@requires_each`` decorator, for when the values are only known here
        (computed from this task's own parameters).

        Each keyword is a parameter name mapped to the **list** of values to fan out over;
        the cartesian product of those lists is taken. Every task returned is a
        :py:meth:`clone`, so the parameters this task and ``cls`` have in common -- everything
        the flow passed down -- reach every branch without being listed here.

        Dict keys are the value itself for a single parameter, or ``name_value`` pairs joined
        with ``_`` for several; they are what ``inputLoad(task=key)`` selects on.

        A value may also be a callable taking this task and returning the list, for a grid
        computed from this task's own **parameters**. It may not read this task's inputs --
        ``input()`` is defined as ``getpaths(self.requires())``, so that recurses forever.

        ``derive={'name': fn}`` sets a FURTHER parameter per branch, computed from that branch's
        fanned values: ``derive={'source': lambda v: URLS[v['region']]}``. It lands in the
        branch's parameters, and therefore in its ``task_id`` -- so editing the lookup
        invalidates exactly the branches it changes. The derived name stays OUT of the dict key
        (the key remains the fanned value, which the derived value is a function of).

        ::

            def requires(self):
                return self.requires_grid(ProcessState, state=STATES[self.country])
        """
        if cls is None:
            cls = self.__class__
        names = sorted(grid)
        if not names:
            raise ValueError(
                '{}: requires_grid needs at least one parameter to fan out over'.format(self.task_family))
        resolved = {}
        for name in names:
            values = grid[name]
            if callable(values):
                values = values(self)
            if not isinstance(values, (list, tuple)):
                raise ValueError(
                    '{}: requires_grid values must be a list of values to fan out over, got {} for {}'.format(
                        self.task_family, repr(values), name))
            resolved[name] = values
        derive = _check_grid(cls, set(names), derive, self.task_family)

        out, seen = {}, {}
        for combo in itertools.product(*(resolved[n] for n in names)):
            values = dict(zip(names, combo))
            if len(names) == 1:
                key = values[names[0]]
            else:
                key = '_'.join('{}_{}'.format(n, values[n]) for n in names)
            # Two combos can produce one key -- values joined with '_' that themselves contain
            # '_' ({'x': ['a_y_b', 'a'], 'y': ['c', 'b_y_c']}), or values that merely hash
            # equal (1 and True). Assigning would DROP a branch: it would never run and never
            # show in preview(). Raise instead.
            if key in seen:
                raise ValueError(
                    '{}: fan-out values {} and {} both produce the dependency key {!r}, so one '
                    'branch would be silently dropped. Rename a value so the keys differ (a '
                    "value containing '_' can collide once several parameters are joined).".format(
                        self.task_family, seen[key], values, key))
            seen[key] = values
            # a copy per callable: one that mutates its argument must not corrupt the key
            # bookkeeping above or another branch's values
            extra = {n: fn(dict(values)) for n, fn in derive.items()}
            out[key] = self.clone(cls, **values, **extra)
        return out

    def __hash__(self):
        return self.__hash

    def __repr__(self):
        params = dict(self.get_params())
        repr_parts = []
        for param_name, param_value in self.param_kwargs.items():
            if params[param_name].significant:
                repr_parts.append('%s=%s' % (param_name, params[param_name].serialize(param_value)))
        return '{}({})'.format(self.get_task_family(), ', '.join(repr_parts))

    def __eq__(self, other):
        return self.__class__ == other.__class__ and self.task_id == other.task_id

    def complete(self):
        """``True`` if all of this task's outputs exist.

        Note: ``TaskData`` OVERRIDES this to ALSO require the stored code
        fingerprint to match the current one (``TaskData._code_ok``), so a
        ``code_version`` bump makes a task incomplete and forces a rerun -- the
        fingerprint is authoritative here, not merely advisory. The AST
        source-hash is a SEPARATE, warn-only advisory (fires when code changed
        but ``code_version`` did not); it does not gate completeness.
        """
        import warnings
        outputs = flatten(self.output())
        if len(outputs) == 0:
            warnings.warn(
                "Task %r without outputs has no custom complete() method" % self,
                stacklevel=2,
            )
            return False
        return all(output.exists() for output in outputs)

    def output(self):
        """The output Target(s) this task produces. Default: none."""
        return []

    def requires(self):
        """The Task(s) this task depends on. Default: none."""
        return []

    def input(self):
        """The outputs of the tasks returned by :py:meth:`requires`."""
        return getpaths(self.requires())

    def deps(self):
        """Flattened list of required tasks."""
        return flatten(self.requires())

    @property
    def _code_fingerprint(self):
        """Recursive code identity, compared against the state store by
        TaskData.complete() -- a mismatch forces a rerun (authoritative). The own
        token is the explicit ``code_version`` when declared; otherwise, with
        ``settings.code_version_auto`` (the default), the AST hash of the task's own
        class plus the repo-local symbols it transitively references, so logic edits
        rerun automatically -- and edits to unrelated siblings in the same file don't.
        None when neither applies here or upstream (feature inert).
        For tasks WITH an explicit code_version the AST source-hash stays a
        warn-only advisory and never gates completeness.

        Memoized per TRAVERSAL only (``traversal_scope``), never per instance: instances are
        process-long-lived via the Register cache, so a cached fingerprint would go stale on a
        runtime ``code_version`` bump between runs. Within one traversal it cannot change --
        ``codehash.freeze()`` already brackets ``build()`` on that premise -- and recomputing it
        is O(paths): 8,439 recursive evaluations for one no-op run over 75 tasks.
        """
        return fingerprint_cached(self, self._code_fingerprint_compute)

    def _code_fingerprint_compute(self):
        dep_fps = [d._code_fingerprint for d in self.deps()]
        own = self.code_version
        if own is None:
            from oryxflow import settings as _settings_
            if _settings_.code_version_auto:
                from oryxflow import codehash as _codehash_
                h = _codehash_.task_code_hash(self)
                own = 'auto:{}'.format(h) if h is not None else None
        if own is None and all(f is None for f in dep_fps):
            return None
        parts = [self.task_family, str(own)] + sorted(f or '' for f in dep_fps)
        return hashlib.md5('|'.join(parts).encode('utf-8')).hexdigest()[:16]

    def run(self):
        """The task computation, to be overridden in a subclass."""
        pass

task_family property

task_family

The task family (class name) of this task instance.

logger property

logger

Contextual logger for task authors; auto-tagged with task identity.

Lives in the oryxflow namespace, so it is silent until oryxflow.enable_logging() is called.

get_params classmethod

get_params()

Return all (name, Parameter) pairs for this task, in declaration order.

Source code in oryxflow/core.py

@classmethod
def get_params(cls):
    """Return all ``(name, Parameter)`` pairs for this task, in declaration order."""
    params = []
    for param_name in dir(cls):
        param_obj = getattr(cls, param_name)
        if not isinstance(param_obj, Parameter):
            continue
        params.append((param_name, param_obj))

    params.sort(key=lambda t: t[1]._counter)
    return params

get_param_names classmethod

get_param_names(include_significant=False)

Return parameter names. include_significant=True returns all params.

Source code in oryxflow/core.py

@classmethod
def get_param_names(cls, include_significant=False):
    """Return parameter names. ``include_significant=True`` returns all params."""
    return [name for name, p in cls.get_params() if include_significant or p.significant]

get_param_values classmethod

get_param_values(params, args, kwargs)

Resolve parameter values from positional args, kwargs and defaults.

:param params: list of (param_name, Parameter) :param args: positional arguments :param kwargs: keyword arguments :returns: list of (name, value) tuples, one per parameter

Source code in oryxflow/core.py

@classmethod
def get_param_values(cls, params, args, kwargs):
    """
    Resolve parameter values from positional args, kwargs and defaults.

    :param params: list of ``(param_name, Parameter)``
    :param args: positional arguments
    :param kwargs: keyword arguments
    :returns: list of ``(name, value)`` tuples, one per parameter
    """
    result = {}
    params_dict = dict(params)
    task_family = cls.get_task_family()
    exc_desc = '%s[args=%s, kwargs=%s]' % (task_family, args, kwargs)

    # Positional arguments.
    positional_params = [(n, p) for n, p in params if p.positional]
    for i, arg in enumerate(args):
        if i >= len(positional_params):
            raise UnknownParameterException(
                '%s: takes at most %d parameters (%d given)' % (exc_desc, len(positional_params), len(args)))
        param_name, param_obj = positional_params[i]
        result[param_name] = param_obj.normalize(arg)

    # Keyword arguments.
    for param_name, arg in kwargs.items():
        if param_name in result:
            raise DuplicateParameterException(
                '%s: parameter %s was already set as a positional parameter' % (exc_desc, param_name))
        if param_name not in params_dict:
            raise UnknownParameterException('%s: unknown parameter %s' % (exc_desc, param_name))
        result[param_name] = params_dict[param_name].normalize(arg)

    # Defaults for anything still unset.
    for param_name, param_obj in params:
        if param_name not in result:
            if not param_obj.has_task_value():
                raise MissingParameterException(
                    "%s: requires the '%s' parameter to be set" % (exc_desc, param_name))
            result[param_name] = param_obj.task_value()

    return [(param_name, result[param_name]) for param_name, param_obj in params]

get_task_family classmethod

get_task_family()

The task family (class name) for this class.

Source code in oryxflow/core.py

@classmethod
def get_task_family(cls):
    """The task family (class name) for this class."""
    return cls.__name__

to_str_params

to_str_params(only_significant=False, only_public=False)

Convert parameters to a {name: serialized_value} dict.

:param bool only_significant: only include parameters marked significant. :param bool only_public: accepted for API compatibility; visibility is not modeled.

Source code in oryxflow/core.py

def to_str_params(self, only_significant=False, only_public=False):
    """
    Convert parameters to a ``{name: serialized_value}`` dict.

    :param bool only_significant: only include parameters marked significant.
    :param bool only_public: accepted for API compatibility; visibility is not modeled.
    """
    params_str = {}
    params = dict(self.get_params())
    for param_name, param_value in self.param_kwargs.items():
        if (not only_significant) or params[param_name].significant:
            params_str[param_name] = params[param_name].serialize(param_value)
    return params_str

clone

clone(cls=None, **kwargs)

Create a new task instance from this one, overriding some args.

Parameters common to this task and cls are carried over; kwargs take precedence.

Source code in oryxflow/core.py

def clone(self, cls=None, **kwargs):
    """
    Create a new task instance from this one, overriding some args.

    Parameters common to this task and ``cls`` are carried over; ``kwargs`` take precedence.
    """
    if cls is None:
        cls = self.__class__

    new_k = {}
    for param_name, param_class in cls.get_params():
        if param_name in kwargs:
            new_k[param_name] = kwargs[param_name]
        elif hasattr(self, param_name):
            new_k[param_name] = getattr(self, param_name)

    return cls(**new_k)

requires_grid

requires_grid(cls=None, derive=None, **grid)

Build the requires() dict for depending on cls once per value -- the runtime form of the @requires_each decorator, for when the values are only known here (computed from this task's own parameters).

Each keyword is a parameter name mapped to the list of values to fan out over; the cartesian product of those lists is taken. Every task returned is a meth:clone, so the parameters this task and cls have in common -- everything the flow passed down -- reach every branch without being listed here.

Dict keys are the value itself for a single parameter, or name_value pairs joined with _ for several; they are what inputLoad(task=key) selects on.

A value may also be a callable taking this task and returning the list, for a grid computed from this task's own parameters. It may not read this task's inputs -- input() is defined as getpaths(self.requires()), so that recurses forever.

derive={'name': fn} sets a FURTHER parameter per branch, computed from that branch's fanned values: derive={'source': lambda v: URLS[v['region']]}. It lands in the branch's parameters, and therefore in its task_id -- so editing the lookup invalidates exactly the branches it changes. The derived name stays OUT of the dict key (the key remains the fanned value, which the derived value is a function of).

::

def requires(self):
    return self.requires_grid(ProcessState, state=STATES[self.country])

Source code in oryxflow/core.py

def requires_grid(self, cls=None, derive=None, **grid):
    """
    Build the ``requires()`` dict for depending on ``cls`` once per value -- the runtime
    form of the ``@requires_each`` decorator, for when the values are only known here
    (computed from this task's own parameters).

    Each keyword is a parameter name mapped to the **list** of values to fan out over;
    the cartesian product of those lists is taken. Every task returned is a
    :py:meth:`clone`, so the parameters this task and ``cls`` have in common -- everything
    the flow passed down -- reach every branch without being listed here.

    Dict keys are the value itself for a single parameter, or ``name_value`` pairs joined
    with ``_`` for several; they are what ``inputLoad(task=key)`` selects on.

    A value may also be a callable taking this task and returning the list, for a grid
    computed from this task's own **parameters**. It may not read this task's inputs --
    ``input()`` is defined as ``getpaths(self.requires())``, so that recurses forever.

    ``derive={'name': fn}`` sets a FURTHER parameter per branch, computed from that branch's
    fanned values: ``derive={'source': lambda v: URLS[v['region']]}``. It lands in the
    branch's parameters, and therefore in its ``task_id`` -- so editing the lookup
    invalidates exactly the branches it changes. The derived name stays OUT of the dict key
    (the key remains the fanned value, which the derived value is a function of).

    ::

        def requires(self):
            return self.requires_grid(ProcessState, state=STATES[self.country])
    """
    if cls is None:
        cls = self.__class__
    names = sorted(grid)
    if not names:
        raise ValueError(
            '{}: requires_grid needs at least one parameter to fan out over'.format(self.task_family))
    resolved = {}
    for name in names:
        values = grid[name]
        if callable(values):
            values = values(self)
        if not isinstance(values, (list, tuple)):
            raise ValueError(
                '{}: requires_grid values must be a list of values to fan out over, got {} for {}'.format(
                    self.task_family, repr(values), name))
        resolved[name] = values
    derive = _check_grid(cls, set(names), derive, self.task_family)

    out, seen = {}, {}
    for combo in itertools.product(*(resolved[n] for n in names)):
        values = dict(zip(names, combo))
        if len(names) == 1:
            key = values[names[0]]
        else:
            key = '_'.join('{}_{}'.format(n, values[n]) for n in names)
        # Two combos can produce one key -- values joined with '_' that themselves contain
        # '_' ({'x': ['a_y_b', 'a'], 'y': ['c', 'b_y_c']}), or values that merely hash
        # equal (1 and True). Assigning would DROP a branch: it would never run and never
        # show in preview(). Raise instead.
        if key in seen:
            raise ValueError(
                '{}: fan-out values {} and {} both produce the dependency key {!r}, so one '
                'branch would be silently dropped. Rename a value so the keys differ (a '
                "value containing '_' can collide once several parameters are joined).".format(
                    self.task_family, seen[key], values, key))
        seen[key] = values
        # a copy per callable: one that mutates its argument must not corrupt the key
        # bookkeeping above or another branch's values
        extra = {n: fn(dict(values)) for n, fn in derive.items()}
        out[key] = self.clone(cls, **values, **extra)
    return out

complete

complete()

True if all of this task's outputs exist.

Note: TaskData OVERRIDES this to ALSO require the stored code fingerprint to match the current one (TaskData._code_ok), so a code_version bump makes a task incomplete and forces a rerun -- the fingerprint is authoritative here, not merely advisory. The AST source-hash is a SEPARATE, warn-only advisory (fires when code changed but code_version did not); it does not gate completeness.

Source code in oryxflow/core.py

def complete(self):
    """``True`` if all of this task's outputs exist.

    Note: ``TaskData`` OVERRIDES this to ALSO require the stored code
    fingerprint to match the current one (``TaskData._code_ok``), so a
    ``code_version`` bump makes a task incomplete and forces a rerun -- the
    fingerprint is authoritative here, not merely advisory. The AST
    source-hash is a SEPARATE, warn-only advisory (fires when code changed
    but ``code_version`` did not); it does not gate completeness.
    """
    import warnings
    outputs = flatten(self.output())
    if len(outputs) == 0:
        warnings.warn(
            "Task %r without outputs has no custom complete() method" % self,
            stacklevel=2,
        )
        return False
    return all(output.exists() for output in outputs)

output

output()

The output Target(s) this task produces. Default: none.

Source code in oryxflow/core.py

def output(self):
    """The output Target(s) this task produces. Default: none."""
    return []

requires

requires()

The Task(s) this task depends on. Default: none.

Source code in oryxflow/core.py

def requires(self):
    """The Task(s) this task depends on. Default: none."""
    return []

input

input()

The outputs of the tasks returned by meth:requires.

Source code in oryxflow/core.py

def input(self):
    """The outputs of the tasks returned by :py:meth:`requires`."""
    return getpaths(self.requires())

deps

deps()

Flattened list of required tasks.

Source code in oryxflow/core.py

def deps(self):
    """Flattened list of required tasks."""
    return flatten(self.requires())

run

run()

The task computation, to be overridden in a subclass.

Source code in oryxflow/core.py

def run(self):
    """The task computation, to be overridden in a subclass."""
    pass

Target

Minimal abstract target base.

Source code in oryxflow/core.py

class Target:
    """Minimal abstract target base."""

    def exists(self):
        raise NotImplementedError

LocalTarget

Bases: Target

Tiny local target base.

self.path is stored as-is (NOT coerced to str); subclasses (CacheTarget/_LocalPathTarget) override the rest and normalize the path.

Source code in oryxflow/core.py

class LocalTarget(Target):
    """
    Tiny local target base.

    ``self.path`` is stored as-is (NOT coerced to ``str``); subclasses
    (``CacheTarget``/``_LocalPathTarget``) override the rest and normalize the path.
    """

    def __init__(self, path=None):
        self.path = path

    def exists(self):
        return False

inherits

Copy parameters (and nothing else) from one or more task classes onto the decorated task, and add clone_parent/clone_parents helpers. Avoids pythonic inheritance.

Supports positional tasks (clone_parents returns a list) or named tasks via keyword arguments (clone_parents returns a dict).

Source code in oryxflow/core.py

class inherits:
    """
    Copy parameters (and nothing else) from one or more task classes onto the decorated task,
    and add ``clone_parent``/``clone_parents`` helpers. Avoids pythonic inheritance.

    Supports positional tasks (``clone_parents`` returns a list) or named tasks via keyword
    arguments (``clone_parents`` returns a dict).
    """

    def __init__(self, *tasks_to_inherit, **kw_tasks_to_inherit):
        super(inherits, self).__init__()
        if not tasks_to_inherit and not kw_tasks_to_inherit:
            raise TypeError("tasks_to_inherit or kw_tasks_to_inherit must contain at least one task")
        if tasks_to_inherit and kw_tasks_to_inherit:
            raise TypeError("Only one of tasks_to_inherit or kw_tasks_to_inherit may be present")
        self.tasks_to_inherit = tasks_to_inherit
        self.kw_tasks_to_inherit = kw_tasks_to_inherit

    def __call__(self, task_that_inherits):
        if self.tasks_to_inherit:
            def clone_parent(_self, **kwargs):
                return _self.clone(cls=self.tasks_to_inherit[0], **kwargs)
            task_that_inherits.clone_parent = clone_parent

            def clone_parents(_self, **kwargs):
                return [
                    _self.clone(cls=task_to_inherit, **kwargs)
                    for task_to_inherit in self.tasks_to_inherit
                ]
            task_that_inherits.clone_parents = clone_parents
        elif self.kw_tasks_to_inherit:
            def clone_parents(_self, **kwargs):
                return {
                    task_name: _self.clone(cls=task_to_inherit, **kwargs)
                    for task_name, task_to_inherit in self.kw_tasks_to_inherit.items()
                }
            task_that_inherits.clone_parents = clone_parents

        return _add_spec(task_that_inherits, {
            'kind': 'inherit',
            'tasks': self.tasks_to_inherit or self.kw_tasks_to_inherit,
        })

requires_each

Fan out over ONE task: copies its parameters except the ones being fanned out, and defines requires() as a meth:Task.requires_grid over them.

The decorated task is the point the branches converge into, so it must NOT carry the fanned-out parameter itself -- that is why those names are skipped when the parameters are copied across.

Stacks with @requires/@inherits/other @requires_each decorators. Pass a single-entry {name: Cls} dict to name the group; it defaults to the dependency's task family. A grid value may be a callable taking the task, resolved at requires() time against its parameters.

derive={'name': fn} sets a further parameter per branch from that branch's fanned values (see meth:Task.requires_grid); derived names are excluded from the decorated task's parameters for the same reason fanned ones are.

Source code in oryxflow/core.py

class requires_each:
    """Fan out over ONE task: copies its parameters **except** the ones being fanned out, and
    defines ``requires()`` as a :py:meth:`Task.requires_grid` over them.

    The decorated task is the point the branches converge into, so it must NOT carry the
    fanned-out parameter itself -- that is why those names are skipped when the parameters are
    copied across.

    Stacks with ``@requires``/``@inherits``/other ``@requires_each`` decorators. Pass a
    single-entry ``{name: Cls}`` dict to name the group; it defaults to the dependency's
    task family. A grid value may be a callable taking the task, resolved at ``requires()``
    time against its parameters.

    ``derive={'name': fn}`` sets a further parameter per branch from that branch's fanned values
    (see :py:meth:`Task.requires_grid`); derived names are excluded from the decorated task's
    parameters for the same reason fanned ones are.
    """

    def __init__(self, task_to_require, /, *extra, derive=None, **grid):
        super(requires_each, self).__init__()
        if extra:
            raise TypeError(
                'requires_each takes exactly one task (or one {name: task} dict). For two '
                'fan-outs, stack two @requires_each decorators.')
        group = None
        if isinstance(task_to_require, dict):
            if len(task_to_require) != 1:
                raise TypeError(
                    'requires_each takes one named task, e.g. '
                    "@requires_each({'narrative': RegionNarrative}, region=[...])")
            group, task_to_require = list(task_to_require.items())[0]
        if not grid:
            raise TypeError(
                'requires_each needs at least one parameter to fan out over, '
                'e.g. @requires_each(ModelTrain, model=MODELS)')
        for name, values in grid.items():
            if not callable(values) and not isinstance(values, (list, tuple)):
                raise TypeError(
                    'requires_each values must be a list of values to fan out over (or a '
                    'callable returning one), got {} for {}'.format(repr(values), name))
        self.task_to_require = task_to_require
        self.group = group
        self.grid = grid
        # the dependency class is fully defined by the time it is decorated with, so the whole
        # spec is checked HERE -- at class definition, not at the first requires() call
        self.derive = _check_grid(task_to_require, set(grid), derive,
                                  '@requires_each({})'.format(task_to_require.__name__))

    def __call__(self, task_that_requires):
        _check_requires_not_overwritten(task_that_requires, 'requires_each')
        return _add_spec(task_that_requires, {'kind': 'each', 'cls': self.task_to_require,
                                              'group': self.group, 'grid': self.grid,
                                              'derive': self.derive})

requires

Same as class:inherits, but also auto-defines the requires method.

Source code in oryxflow/core.py

class requires:
    """Same as :py:class:`inherits`, but also auto-defines the ``requires`` method."""

    def __init__(self, *tasks_to_require, **kw_tasks_to_require):
        super(requires, self).__init__()
        self.tasks_to_require = tasks_to_require
        self.kw_tasks_to_require = kw_tasks_to_require

    def __call__(self, task_that_requires):
        _check_requires_not_overwritten(task_that_requires, 'requires')
        inherits(*self.tasks_to_require, **self.kw_tasks_to_require)(task_that_requires)
        # the inherits call above appended an 'inherit' entry -- upgrade it in place, so the
        # dependency comes from the spec rather than from whichever decorator's
        # clone_parent()/clone_parents() happened to survive
        _spec_entries(task_that_requires)[-1]['kind'] = 'fixed'
        _apply_spec(task_that_requires)
        return task_that_requires

TaskFailure

One failed task: the task instance plus why it failed.

Source code in oryxflow/core.py

class TaskFailure:
    """One failed task: the task instance plus why it failed."""
    def __init__(self, task, exception=None, traceback=None, reason=None):
        self.task = task
        self.exception = exception      # the real run() exception, or None
        self.traceback = traceback      # formatted traceback string, or None
        self.reason = reason            # 'run error' | 'dependency failed' | 'external missing'
    def __str__(self):
        if self.exception is not None:
            return "{}: {}: {}".format(_task_label(self.task),
                                       type(self.exception).__name__, self.exception)
        return "{}: {}".format(_task_label(self.task), self.reason or "failed")

RunResult

What happened to the DAG in one build: identities, status, failure context.

Source code in oryxflow/core.py

class RunResult:
    """What happened to the DAG in one build: identities, status, failure context."""
    def __init__(self, success, ran, complete, failed, run_id=None, reasons=None,
                 warnings=None):
        self.success = success          # bool: no failures
        self.ran = ran                  # list[Task]  actually recomputed
        self.complete = complete        # list[Task]  cache hits (skipped)
        self.failed = failed            # list[TaskFailure]
        self.run_id = run_id            # id shared by this build's events
        self.reasons = reasons or {}    # {task_id: why it ran} for tasks in `ran`
        self.warnings = warnings or []  # unacknowledged code-change warnings, deduped messages

    # --- back-compat aliases ---
    @property
    def scheduling_succeeded(self):
        return self.success
    @property
    def first_exception(self):
        return next((f.exception for f in self.failed if f.exception is not None), None)
    def summary(self):
        """Return the execution summary text (``print(result.summary())``)."""
        return str(self)

    def __bool__(self):
        return self.success

    # --- queries (by task CLASS) ---
    def did_run(self, task_cls):
        return any(isinstance(t, task_cls) for t in self.ran)
    def ran_of(self, task_cls):
        return [t for t in self.ran if isinstance(t, task_cls)]
    def failure_of(self, task_cls):
        return next((f for f in self.failed if isinstance(f.task, task_cls)), None)

    def __str__(self):
        # luigi-compatible wording so the plugin SKILL.md guidance stays accurate.
        n = len(self.complete) + len(self.ran) + len(self.failed)
        def block(label, rendered, cap=None):
            lines = ["* {} {}".format(len(rendered), label) + (":" if rendered else "")]
            shown = rendered if cap is None else rendered[:cap]
            lines += ["    - {}".format(s) for s in shown]
            if cap is not None and len(rendered) > cap:
                lines.append("    ... and {} more".format(len(rendered) - cap))
            return lines
        parts = ["Scheduled {} tasks of which:".format(n)]
        parts += block("complete ones were encountered",
                       [_task_label(t) for t in self.complete], cap=10)
        parts += block("ran successfully", [_task_label(t) for t in self.ran])
        parts += block("failed", [str(f) for f in self.failed])   # TaskFailure.__str__
        smiley = ":)" if self.success else ":("
        why = ("there were no failed tasks or missing dependencies" if self.success
               else "there were failed tasks")
        parts.append("This progress looks {} because {}".format(smiley, why))
        return "\n".join(parts)

summary

summary()

Return the execution summary text (print(result.summary())).

Source code in oryxflow/core.py

def summary(self):
    """Return the execution summary text (``print(result.summary())``)."""
    return str(self)

MultiRunResult

Bases: dict

Return value of WorkflowMulti.run(): a {flow_name: RunResult} dict that also carries .summary()/.success so print(result.summary()) works the same as for a single Workflow (whose run() returns a plain class:RunResult).

Source code in oryxflow/core.py

class MultiRunResult(dict):
    """Return value of ``WorkflowMulti.run()``: a ``{flow_name: RunResult}`` dict that also
    carries ``.summary()``/``.success`` so ``print(result.summary())`` works the same as for a
    single ``Workflow`` (whose ``run()`` returns a plain :py:class:`RunResult`)."""
    @property
    def success(self):
        return all(bool(r) for r in self.values())
    def __bool__(self):
        return self.success

    # --- aggregates across flows, so callers never hand-roll them ---
    @property
    def ran(self):
        return [t for r in self.values() for t in r.ran]
    @property
    def complete(self):
        return [t for r in self.values() for t in r.complete]
    @property
    def failed(self):
        return [f for r in self.values() for f in r.failed]
    @property
    def reasons(self):
        merged = {}
        for r in self.values():
            merged.update(r.reasons)
        return merged
    @property
    def warnings(self):
        # deduped across flows: each per-flow build re-warns for shared upstreams, and
        # "how many pending conditions" must not scale with the flow count
        merged = []
        for r in self.values():
            for w in r.warnings:
                if w not in merged:
                    merged.append(w)
        return merged

    def summary(self):
        """Per-flow execution summaries, each under a ``===== <flow> =====`` header."""
        return "\n\n".join(
            "===== {} =====\n{}".format(name, res.summary()) for name, res in self.items())
    def __str__(self):
        return self.summary()

summary

summary()

Per-flow execution summaries, each under a ===== <flow> ===== header.

Source code in oryxflow/core.py

def summary(self):
    """Per-flow execution summaries, each under a ``===== <flow> =====`` header."""
    return "\n\n".join(
        "===== {} =====\n{}".format(name, res.summary()) for name, res in self.items())

flatten

flatten(struct)

Create a flat list of all items in a structured object (dicts, lists, items)::

>>> sorted(flatten({'a': 'foo', 'b': 'bar'}))
['bar', 'foo']
>>> sorted(flatten(['foo', ['bar', 'troll']]))
['bar', 'foo', 'troll']
>>> flatten('foo')
['foo']
>>> flatten(42)
[42]

Source code in oryxflow/core.py

def flatten(struct):
    """
    Create a flat list of all items in a structured object (dicts, lists, items)::

        >>> sorted(flatten({'a': 'foo', 'b': 'bar'}))
        ['bar', 'foo']
        >>> sorted(flatten(['foo', ['bar', 'troll']]))
        ['bar', 'foo', 'troll']
        >>> flatten('foo')
        ['foo']
        >>> flatten(42)
        [42]
    """
    if struct is None:
        return []
    flat = []
    if isinstance(struct, dict):
        for _, result in struct.items():
            flat += flatten(result)
        return flat
    if isinstance(struct, str):
        return [struct]

    try:
        iterator = iter(struct)
    except TypeError:
        return [struct]

    for result in iterator:
        flat += flatten(result)
    return flat

getpaths

getpaths(struct)

Map all Tasks in a structured object to their .output().

Source code in oryxflow/core.py

def getpaths(struct):
    """Map all Tasks in a structured object to their ``.output()``."""
    if isinstance(struct, Task):
        return struct.output()
    elif isinstance(struct, dict):
        return struct.__class__((k, getpaths(v)) for k, v in struct.items())
    elif isinstance(struct, (list, tuple)):
        return struct.__class__(getpaths(r) for r in struct)
    else:
        try:
            return [getpaths(r) for r in struct]
        except TypeError:
            raise Exception('Cannot map %s to Task/dict/list' % str(struct))

traversal_scope

traversal_scope()

Memoize completeness, resolved dependencies and code fingerprints for the duration of one traversal -- a build, a preview, a Workflow.complete(), accept_code(), the upstream/downstream walks behind the invalidate helpers.

Nested scopes share one memo and only the outermost drops it, so a task's run() calling oryxflow.run() (flow-within-a-flow) needs no special case: that nested build clears the shared volatile memo whenever it materializes something.

Source code in oryxflow/core.py

@contextlib.contextmanager
def traversal_scope():
    """Memoize completeness, resolved dependencies and code fingerprints for the duration of
    one traversal -- a build, a preview, a ``Workflow.complete()``, ``accept_code()``, the
    upstream/downstream walks behind the invalidate helpers.

    Nested scopes share one memo and only the outermost drops it, so a task's ``run()`` calling
    ``oryxflow.run()`` (flow-within-a-flow) needs no special case: that nested build clears the
    shared volatile memo whenever it materializes something.
    """
    st = _traversal
    if not _traversal_memo_enabled:
        yield
        return
    if st.depth == 0:
        st.complete, st.requires, st.fingerprint = {}, {}, {}
        st.stats = _new_stats()
    st.depth += 1
    try:
        yield
    finally:
        st.depth -= 1
        if st.depth == 0:
            s = st.last_stats = st.stats
            logger.debug(
                "traversal memo: complete {}/{}, requires {}/{}, fingerprint {}/{} hit, "
                "{} invalidation(s)",
                s['complete_hit'], s['complete_hit'] + s['complete_miss'],
                s['requires_hit'], s['requires_hit'] + s['requires_miss'],
                s['fingerprint_hit'], s['fingerprint_hit'] + s['fingerprint_miss'],
                s['cleared'])
            st.complete = st.requires = st.fingerprint = st.stats = None

traversal_stats

traversal_stats()

Memo counters for the traversal in flight, else the most recent one.

Internal. The regression tests assert on the MISS counts: one miss per unique task is the invariant this whole section exists to hold, and unlike wall-clock it is deterministic.

Source code in oryxflow/core.py

def traversal_stats():
    """Memo counters for the traversal in flight, else the most recent one.

    Internal. The regression tests assert on the MISS counts: one miss per unique task is the
    invariant this whole section exists to hold, and unlike wall-clock it is deterministic.
    """
    st = _traversal
    return dict(st.stats or st.last_stats or _new_stats())

traversal_memo_clear

traversal_memo_clear()

Drop the volatile memo: something changed what is complete (a task materialized, an output was invalidated). The structural memos are untouched -- neither the code nor the DAG shape changes mid-traversal.

Source code in oryxflow/core.py

def traversal_memo_clear():
    """Drop the volatile memo: something changed what is complete (a task materialized, an
    output was invalidated). The structural memos are untouched -- neither the code nor the DAG
    shape changes mid-traversal."""
    st = _traversal
    if st.complete is not None:
        st.complete.clear()
        st.gen += 1
        st.stats['cleared'] += 1

complete_cached

complete_cached(task, cascade, compute)

Evaluate a cascading complete() at most once per task per traversal.

cascade=False is never memoized: it does no recursion, so there is nothing to save, and it is what the load paths ask right after a save.

Source code in oryxflow/core.py

def complete_cached(task, cascade, compute):
    """Evaluate a cascading ``complete()`` at most once per task per traversal.

    ``cascade=False`` is never memoized: it does no recursion, so there is nothing to save, and
    it is what the load paths ask right after a save.
    """
    if not cascade:
        return compute()
    return _memoized(_traversal.complete, 'complete', task, compute)

task_id_str

task_id_str(task_family, params)

Return a canonical, deterministic string identifying a task.

The id is {family}_{param_summary}_{md5(sorted_json)[:10]} so that task_id.split('_')[0] yields the task family (the directory convention).

:param task_family: the task family (class name) :param params: dict mapping parameter names to serialized (str) values

Source code in oryxflow/core.py

def task_id_str(task_family, params):
    """
    Return a canonical, deterministic string identifying a task.

    The id is ``{family}_{param_summary}_{md5(sorted_json)[:10]}`` so that
    ``task_id.split('_')[0]`` yields the task family (the directory convention).

    :param task_family: the task family (class name)
    :param params: dict mapping parameter names to serialized (str) values
    """
    param_str = json.dumps(params, separators=(',', ':'), sort_keys=True)
    param_hash = hashlib.md5(param_str.encode('utf-8')).hexdigest()

    param_summary = '_'.join(p[:TASK_ID_TRUNCATE_PARAMS]
                             for p in (params[p] for p in sorted(params)[:TASK_ID_INCLUDE_PARAMS]))
    param_summary = TASK_ID_INVALID_CHAR_REGEX.sub('_', param_summary)

    return '{}_{}_{}'.format(task_family, param_summary, param_hash[:TASK_ID_TRUNCATE_HASH])

dfs_paths

dfs_paths(start_task, goal_task_family, path=None)

Back-compat generator: yield tasks on paths from start_task to goal_task_family (goal_task_family=None yields the whole upstream DAG). Superseded by find_paths (ordered paths) and find_deps (deduped set); both are memoized.

Source code in oryxflow/core.py

def dfs_paths(start_task, goal_task_family, path=None):
    """Back-compat generator: yield tasks on paths from ``start_task`` to ``goal_task_family``
    (``goal_task_family=None`` yields the whole upstream DAG). Superseded by ``find_paths``
    (ordered paths) and ``find_deps`` (deduped set); both are memoized."""
    if goal_task_family is None:
        for t in _all_upstream(start_task):
            yield t
        return
    for p in _suffix_paths(start_task, goal_task_family, {}):
        for t in p:
            yield t

find_deps

find_deps(task, upstream_task_family)

Set of all tasks on all paths between task (the downstream root) and upstream_task_family. upstream_task_family=None returns the whole upstream DAG.

Source code in oryxflow/core.py

def find_deps(task, upstream_task_family):
    """Set of all tasks on all paths between ``task`` (the downstream root) and
    ``upstream_task_family``. ``upstream_task_family=None`` returns the whole upstream DAG."""
    if upstream_task_family is None:
        return _all_upstream(task)
    return _reachable_to(task, upstream_task_family, {})

find_paths

find_paths(task, upstream_task_family)

Ordered dependency paths (task -> ... -> upstream_task_family), deduped, root first. Returns a list of lists of task instances; empty if the family is unreachable.

Source code in oryxflow/core.py

def find_paths(task, upstream_task_family):
    """Ordered dependency paths (``task`` -> ... -> ``upstream_task_family``), deduped, root
    first. Returns a list of lists of task instances; empty if the family is unreachable."""
    seen = set()
    out = []
    for p in _suffix_paths(task, upstream_task_family, {}):
        key = tuple(t.task_id for t in p)
        if key not in seen:
            seen.add(key)
            out.append(p)
    return out

build

build(tasks, workers=1, detailed_summary=False, flow=None, **ignored)

Run tasks and their dependencies sequentially, in dependency order.

All state is local to this call, so a task's run() may itself call oryxflow.run / flow.run (the flow-within-a-flow pattern) without corrupting the outer build.

External tasks (external=True or run is None) are never executed; if still incomplete after their dependencies, they are marked failed.

:param flow: flow name when launched via Workflow/WorkflowMulti; stamped into this build's event envelopes. :returns: a class:RunResult with .ran/.complete/.failed task identities, .success, .run_id, per-task .reasons and code-change .warnings.

Source code in oryxflow/core.py

def build(tasks, workers=1, detailed_summary=False, flow=None, **ignored):
    """
    Run ``tasks`` and their dependencies sequentially, in dependency order.

    All state is local to this call, so a task's ``run()`` may itself call ``oryxflow.run`` /
    ``flow.run`` (the flow-within-a-flow pattern) without corrupting the outer build.

    External tasks (``external=True`` or ``run is None``) are never executed; if still
    incomplete after their dependencies, they are marked failed.

    :param flow: flow name when launched via ``Workflow``/``WorkflowMulti``; stamped into
        this build's event envelopes.
    :returns: a :py:class:`RunResult` with ``.ran``/``.complete``/``.failed`` task identities,
        ``.success``, ``.run_id``, per-task ``.reasons`` and code-change ``.warnings``.
    """
    import uuid
    from oryxflow import events as _events
    from oryxflow import state as _state
    from oryxflow import codehash as _codehash
    from oryxflow import codecheck as _codecheck
    from oryxflow import log as _log
    from oryxflow import settings as _settings

    if not isinstance(tasks, (list,)):
        tasks = [tasks]

    visited = {}          # task_id -> bool (success)
    ran = []              # tasks executed this build
    already_complete = []  # tasks already complete
    failed = []           # TaskFailure for tasks that failed (run error, failed dep, missing external)
    reasons = {}          # task_id -> why it ran

    run_id = uuid.uuid4().hex
    git_sha, git_dirty = _git_info()

    def _emit(event_type, payload, msg=None, level='info'):
        # single call site per engine fact: the event and the human loguru line
        # come from the same data and can't disagree
        _events.append(event_type, payload, run_id=run_id, flow=flow)
        if msg is not None:
            getattr(logger, level)("{}", msg)

    _advisor = _codecheck.Advisor(_emit)
    warned = _advisor.warned   # unacknowledged code-change warning strings (RunResult)

    def _emit_failed(task, error, tb, duration):
        _emit('task_failed',
              {'task_id': task.task_id, 'family': task.task_family,
               'params': task.to_str_params(only_significant=False),
               'error': error, 'traceback': tb[-4096:] if tb else None,
               'duration_s': duration})

    def _process(task):
        tid = task.task_id
        if tid in visited:
            return visited[tid]

        if task.complete():
            _advisor.advise(task)
            logger.debug("task skipped (already complete): {}", tid)
            already_complete.append(task)
            visited[tid] = True
            return True

        # Process dependencies first.
        dep_ok = True
        for dep in flatten(task.requires()):
            if not _process(dep):
                dep_ok = False
        if not dep_ok:
            logger.error("task failed (dependency failed): {}", tid)
            failed.append(TaskFailure(task, reason="dependency failed"))
            _emit_failed(task, 'dependency failed', None, None)
            visited[tid] = False
            return False

        # External tasks are never executed; their output must come from elsewhere.
        is_external = getattr(task, 'external', False) or getattr(task, 'run', None) is None
        if is_external:
            if task.complete():
                already_complete.append(task)
                visited[tid] = True
                return True
            logger.warning("external task missing output: {}", tid)
            failed.append(TaskFailure(task, reason="external missing output"))
            _emit_failed(task, 'external missing output', None, None)
            visited[tid] = False
            return False

        reason = _advisor.reason_for(task, ran, reasons)
        logger.info("task start: {} ({})", task.task_family, tid)
        t0 = time.perf_counter()
        try:
            if not _run_task(task):
                failed.append(TaskFailure(task, reason="dependency failed"))
                _emit_failed(task, 'dependency failed', None,
                             time.perf_counter() - t0)
                visited[tid] = False
                return False
        except Exception as e:
            logger.opt(exception=True).error("task failed: {}", tid)
            tb = traceback.format_exc()
            failed.append(TaskFailure(task, exception=e, traceback=tb,
                                      reason="run error"))
            _emit_failed(task, '{}: {}'.format(type(e).__name__, e), tb,
                         time.perf_counter() - t0)
            visited[tid] = False
            return False

        duration = time.perf_counter() - t0
        fp, dirpath = _codecheck.code_state(task)
        hashes = _codecheck.hashes_of(task) if (fp is not None
                                                or getattr(_settings, 'events', True)) else {}
        if fp is not None:
            try:
                # fresh output_id: this task actually rematerialized, downstream
                # dep_state folds must move
                _state.put_record(dirpath, tid,
                                  _codecheck.make_record(task, hashes, uuid.uuid4().hex[:16],
                                                         duration=duration))
            except Exception as e:
                logger.debug("state record write failed for {}: {}", tid, e)
        try:
            from oryxflow import __version__ as _version
        except Exception:
            _version = None
        _emit('task_ran',
              {'task_id': tid, 'family': task.task_family,
               'params': task.to_str_params(only_significant=False),
               'code_version': task.code_version, 'fingerprint': fp,
               'auto': task.code_version is None and _settings.code_version_auto,
               'source_hashes': hashes, 'reason': reason, 'duration_s': duration,
               'git_sha': git_sha, 'git_dirty': git_dirty,
               'oryxflow_version': _version,
               'dirpath': str(dirpath if dirpath is not None else _settings.dirpath)},
              msg="task complete: {} in {:.3f}s".format(tid, duration))
        _code_warned.pop(tid, None)   # condition resolved; a future recurrence re-warns
        ran.append(task)
        reasons[tid] = reason
        visited[tid] = True
        return True

    def _run_task(task):
        # A task that materializes changes what is complete, so no volatile answer may outlive
        # it -- and code inside run() (inputLoad, a flow-within-a-flow oryxflow.run(), a manual
        # complete()) must not read an answer computed before it either. Hence both sides.
        # TaskData.save() clears too; this bracket also covers tasks that materialize without
        # save() and runs that fail part-way.
        traversal_memo_clear()
        try:
            result = task.run()
            if inspect.isgenerator(result):
                return _drive_generator(result)
            return True
        finally:
            traversal_memo_clear()

    def _drive_generator(gen):
        # Drive a generator-style run() that yields tasks (eg TaskAggregator) or dynamic
        # requirements, processing each yielded batch before resuming.
        try:
            next_requires = next(gen)
            while True:
                logger.debug("generator yielded {} requires", len(flatten(next_requires)))
                ok = True
                for req in flatten(next_requires):
                    if not _process(req):
                        ok = False
                if not ok:
                    gen.close()
                    return False
                next_requires = gen.send(getpaths(next_requires))
        except StopIteration:
            return True

    _emit('run_started',
          {'tasks': sorted({t.task_family for t in tasks})},
          msg="run started: {} (run_id={})".format(
              ', '.join(sorted({t.task_family for t in tasks})), run_id))

    def _capture_task_log(level, message, context):
        extra = {}
        for k, v in list(context.items())[:20]:
            extra[k] = v if isinstance(v, (int, float, bool, type(None))) else str(v)[:512]
        _events.append('task_log',
                       {'task_id': context.get('task_id'),
                        'family': context.get('task_family'),
                        'level': level, 'message': str(message)[:4096], 'extra': extra},
                       run_id=run_id, flow=flow)

    previous_capture = _log.set_task_log_capture(_capture_task_log)
    _codehash.freeze()   # code can't change mid-build: skip per-complete() mtime re-stats
    try:
        with traversal_scope():
            for task in tasks:
                _process(task)
    finally:
        _codehash.unfreeze()
        _log.set_task_log_capture(previous_capture)

    success = len(failed) == 0
    _emit('run_finished',
          {'counts': {'ran': len(ran), 'complete': len(already_complete),
                      'failed': len(failed)}, 'success': success},
          msg="run finished: {} ran, {} complete, {} failed, success={}".format(
              len(ran), len(already_complete), len(failed), success))
    try:
        _events.flush()   # events are written async; the run's story is durable on return
    except Exception:
        pass
    result = RunResult(success, ran, already_complete, failed,
                       run_id=run_id, reasons=reasons, warnings=warned)
    if detailed_summary:                       # gated by execution_summary
        logger.info("run summary:\n{}", result.summary())
    return result

Parameters — oryxflow.parameter

oryxflow.parameter

Self-contained, trimmed set of parameter types used by oryxflow.

There is no command-line / config-file value resolution, parameter visibility, date_interval, freezing (FrozenOrderedDict) or jsonschema support. Values are resolved from the constructor argument or the default only.

Only the parameter types oryxflow actually uses are kept: Parameter, IntParameter, FloatParameter, BoolParameter, DateParameter, DictParameter, ListParameter, ChoiceParameter and EnumParameter.

ParameterException

Bases: Exception

Base parameter exception.

Source code in oryxflow/parameter.py

class ParameterException(Exception):
    """Base parameter exception."""
    pass

MissingParameterException

Bases: ParameterException

Raised when a required parameter has no value and no default.

Source code in oryxflow/parameter.py

class MissingParameterException(ParameterException):
    """Raised when a required parameter has no value and no default."""
    pass

UnknownParameterException

Bases: ParameterException

Raised when an unknown parameter is supplied to a task.

Source code in oryxflow/parameter.py

class UnknownParameterException(ParameterException):
    """Raised when an unknown parameter is supplied to a task."""
    pass

DuplicateParameterException

Bases: ParameterException

Raised when a parameter is supplied both positionally and as a keyword.

Source code in oryxflow/parameter.py

class DuplicateParameterException(ParameterException):
    """Raised when a parameter is supplied both positionally and as a keyword."""
    pass

Parameter

Parameter whose value is a str, and the base class for the other parameter types.

Parameters are set on the Task class to parameterize tasks::

class MyTask(oryxflow.tasks.TaskData):
    foo = oryxflow.Parameter()

When a value is not provided at instantiation, the default is used.

Source code in oryxflow/parameter.py

class Parameter:
    """
    Parameter whose value is a ``str``, and the base class for the other parameter types.

    Parameters are set on the Task class to parameterize tasks::

        class MyTask(oryxflow.tasks.TaskData):
            foo = oryxflow.Parameter()

    When a value is not provided at instantiation, the ``default`` is used.
    """

    # Non-atomically increasing counter used to preserve declaration order (see Task.get_params).
    _counter = 0

    def __init__(self, default=_no_value, significant=True, description=None, positional=True):
        """
        :param default: the default value for this parameter. If not given, a value must be
                        specified when the task is instantiated.
        :param bool significant: ``False`` if the parameter should not be part of a task's unique
                                 identifier (eg passwords, environment markers). Default ``True``.
        :param str description: human-readable description of the parameter.
        :param bool positional: if ``True`` the parameter may be set positionally. Default ``True``.
        """
        self._default = default
        self.significant = significant
        self.description = description
        self.positional = positional

        self._counter = Parameter._counter
        Parameter._counter += 1

    def has_task_value(self):
        """Whether a default value is available."""
        return self._default != _no_value

    def task_value(self):
        """The normalized default value. Raises if no default is set."""
        if self._default == _no_value:
            raise MissingParameterException("No default specified")
        return self.normalize(self._default)

    def parse(self, x):
        """Parse an individual value from a string. Identity by default."""
        return x

    def serialize(self, x):
        """Convert the value ``x`` to a string. Opposite of :py:meth:`parse`."""
        return str(x)

    def normalize(self, x):
        """Normalize a parsed/default/constructor value. Identity by default."""
        return x

has_task_value

has_task_value()

Whether a default value is available.

Source code in oryxflow/parameter.py

def has_task_value(self):
    """Whether a default value is available."""
    return self._default != _no_value

task_value

task_value()

The normalized default value. Raises if no default is set.

Source code in oryxflow/parameter.py

def task_value(self):
    """The normalized default value. Raises if no default is set."""
    if self._default == _no_value:
        raise MissingParameterException("No default specified")
    return self.normalize(self._default)

parse

parse(x)

Parse an individual value from a string. Identity by default.

Source code in oryxflow/parameter.py

def parse(self, x):
    """Parse an individual value from a string. Identity by default."""
    return x

serialize

serialize(x)

Convert the value x to a string. Opposite of meth:parse.

Source code in oryxflow/parameter.py

def serialize(self, x):
    """Convert the value ``x`` to a string. Opposite of :py:meth:`parse`."""
    return str(x)

normalize

normalize(x)

Normalize a parsed/default/constructor value. Identity by default.

Source code in oryxflow/parameter.py

def normalize(self, x):
    """Normalize a parsed/default/constructor value. Identity by default."""
    return x

IntParameter

Bases: Parameter

Parameter whose value is an int.

Source code in oryxflow/parameter.py

class IntParameter(Parameter):
    """Parameter whose value is an ``int``."""

    def parse(self, s):
        return int(s)

FloatParameter

Bases: Parameter

Parameter whose value is a float.

Source code in oryxflow/parameter.py

class FloatParameter(Parameter):
    """Parameter whose value is a ``float``."""

    def parse(self, s):
        return float(s)

BoolParameter

Bases: Parameter

A Parameter whose value is a bool. Has an implicit default of False.

Source code in oryxflow/parameter.py

class BoolParameter(Parameter):
    """
    A Parameter whose value is a ``bool``. Has an implicit default of ``False``.
    """

    def __init__(self, *args, **kwargs):
        super(BoolParameter, self).__init__(*args, **kwargs)
        if self._default == _no_value:
            self._default = False

    def parse(self, val):
        """Parse a ``bool`` from the string, matching 'true'/'false' case-insensitively."""
        s = str(val).lower()
        if s == "true":
            return True
        elif s == "false":
            return False
        else:
            raise ValueError("cannot interpret '{}' as boolean".format(val))

    def normalize(self, value):
        try:
            return self.parse(value)
        except ValueError:
            return None

parse

parse(val)

Parse a bool from the string, matching 'true'/'false' case-insensitively.

Source code in oryxflow/parameter.py

def parse(self, val):
    """Parse a ``bool`` from the string, matching 'true'/'false' case-insensitively."""
    s = str(val).lower()
    if s == "true":
        return True
    elif s == "false":
        return False
    else:
        raise ValueError("cannot interpret '{}' as boolean".format(val))

DateParameter

Bases: Parameter

Parameter whose value is a class:~datetime.date, formatted YYYY-MM-DD.

Source code in oryxflow/parameter.py

class DateParameter(Parameter):
    """
    Parameter whose value is a :py:class:`~datetime.date`, formatted ``YYYY-MM-DD``.
    """

    date_format = '%Y-%m-%d'

    def parse(self, s):
        return datetime.datetime.strptime(s, self.date_format).date()

    def serialize(self, dt):
        if dt is None:
            return str(dt)
        return dt.strftime(self.date_format)

    def normalize(self, value):
        if value is None:
            return None
        if isinstance(value, datetime.datetime):
            value = value.date()
        return value

DictParameter

Bases: Parameter

Parameter whose value is a dict.

The value is stored as-is (no freezing); it is serialized with sorted keys so that the task id stays deterministic.

Source code in oryxflow/parameter.py

class DictParameter(Parameter):
    """
    Parameter whose value is a ``dict``.

    The value is stored as-is (no freezing); it is serialized with sorted keys so that the
    task id stays deterministic.
    """

    def parse(self, source):
        if not isinstance(source, str):
            return source
        return json.loads(source)

    def serialize(self, x):
        return json.dumps(x, sort_keys=True)

ListParameter

Bases: Parameter

Parameter whose value is a list.

The value is stored as-is (no freezing); it is serialized as JSON so that the task id stays deterministic.

Source code in oryxflow/parameter.py

class ListParameter(Parameter):
    """
    Parameter whose value is a ``list``.

    The value is stored as-is (no freezing); it is serialized as JSON so that the task id
    stays deterministic.
    """

    def parse(self, x):
        if not isinstance(x, str):
            return x
        i = json.loads(x)
        if i is None:
            return None
        return list(i)

    def serialize(self, x):
        return json.dumps(x)

ChoiceParameter

Bases: Parameter

A string-valued parameter restricted to a fixed set of choices::

class MyTask(oryxflow.tasks.TaskData):
    model = oryxflow.ChoiceParameter(choices=['rf', 'lgbm'], default='rf')

Values stay plain strings (no enum.Enum ceremony, unlike :class:EnumParameter); a value outside choices raises immediately at task construction (or when the default is resolved), so typos fail fast instead of dying deep in downstream code.

Source code in oryxflow/parameter.py

class ChoiceParameter(Parameter):
    """
    A string-valued parameter restricted to a fixed set of ``choices``::

        class MyTask(oryxflow.tasks.TaskData):
            model = oryxflow.ChoiceParameter(choices=['rf', 'lgbm'], default='rf')

    Values stay plain strings (no ``enum.Enum`` ceremony, unlike
    :class:`EnumParameter`); a value outside ``choices`` raises immediately at task
    construction (or when the default is resolved), so typos fail fast instead of
    dying deep in downstream code.
    """

    def __init__(self, *args, **kwargs):
        if 'choices' not in kwargs:
            raise ParameterException('A choices list must be specified.')
        choices = kwargs.pop('choices')
        try:
            self._choices = list(choices)
        except TypeError:
            raise ParameterException('choices must be an iterable of values.')
        if not self._choices:
            raise ParameterException('choices must not be empty.')
        super(ChoiceParameter, self).__init__(*args, **kwargs)

    def normalize(self, x):
        if x not in self._choices:
            raise ValueError(
                "'{}' is not a valid choice - must be one of {}".format(x, self._choices))
        return x

EnumParameter

Bases: Parameter

A parameter whose value is an :class:~enum.Enum. Pass the enum class via enum=::

class Model(enum.Enum):
    Honda = 1
    Volvo = 2

class MyTask(oryxflow.tasks.TaskData):
    my_param = oryxflow.EnumParameter(enum=Model)

Source code in oryxflow/parameter.py

class EnumParameter(Parameter):
    """
    A parameter whose value is an :class:`~enum.Enum`. Pass the enum class via ``enum=``::

        class Model(enum.Enum):
            Honda = 1
            Volvo = 2

        class MyTask(oryxflow.tasks.TaskData):
            my_param = oryxflow.EnumParameter(enum=Model)
    """

    def __init__(self, *args, **kwargs):
        if 'enum' not in kwargs:
            raise ParameterException('An enum class must be specified.')
        self._enum = kwargs.pop('enum')
        super(EnumParameter, self).__init__(*args, **kwargs)

    def parse(self, s):
        try:
            return self._enum[s]
        except KeyError:
            raise ValueError('Invalid enum value - could not be parsed')

    def serialize(self, e):
        return e.name

Tasks — oryxflow.tasks

oryxflow.tasks

TaskData

Bases: Task

Task which has data as input and output

Attributes:

Name	Type	Description
target_class	obj	target data format
target_ext	str	file extension
persists	list	list of strings naming the outputs this task saves. Declare it on your task class, e.g. persists = ['x', 'y']. persist (singular) is a backwards-compatible alias for the same thing; prefer persists.
data	dict	data container for all outputs

Source code in oryxflow/tasks/__init__.py

class TaskData(core.Task):
    """
    Task which has data as input and output

    Attributes:
        target_class (obj): target data format
        target_ext (str): file extension
        persists (list): list of strings naming the outputs this task saves.
            Declare it on your task class, e.g. ``persists = ['x', 'y']``.
            ``persist`` (singular) is a backwards-compatible alias for the same
            thing; prefer ``persists``.
        data (dict): data container for all outputs

    """
    target_class = oryxflow.targets.DataTarget
    target_ext = 'ext'
    # canonical internal attribute; users declare it as `persists` (see __init__)
    persist = ['data']
    metadata = None
    # keep outputs of previous code_versions at readable paths (.../Task/v1/...) instead
    # of overwriting in place; only takes effect when code_version is set
    keep_versions = False

    def __init__(self, *args, path=None, flows=None, **kwargs):
        kwargs_ = {k: v for k, v in kwargs.items(
        ) if k in self.get_param_names(include_significant=True)}
        super().__init__(*args, **kwargs_)

        # Check if Child Has Path Var
        self.path = getattr(self, 'path', path)

        # `persists` is the user-facing name; fold it into the internal `persist`.
        # (engine code reads `self.persist` throughout)
        self.persist = getattr(self, 'persists', self.persist)

        # Flow
        self.flows = flows

    @classmethod
    def get_param_values(cls, params, args, kwargs):
        kwargs_ = {k: v for k, v in kwargs.items(
        ) if k in cls.get_param_names(include_significant=True)}
        return super(TaskData, cls).get_param_values(params, args, kwargs_)

    def reset(self, confirm=False):
        """
        Reset a task, eg by deleting output file
        """
        return self.invalidate(confirm)

    def invalidate(self, confirm=False):
        """
        Reset a task, eg by deleting output file
        """
        if confirm:
            c = input(
                'Confirm invalidating task: {} (y/n). PS You can disable this message by passing confirm=False'.format(
                    self.__class__.__qualname__))
        else:
            c = 'y'
        if c == 'y':  # and self.complete():
            if self.persist == ['data']:  # 1 data shortcut
                self.output().invalidate()
            else:
                [t.invalidate() for t in self.output().values()]
            self._invalidate_meta()
            # this task's outputs changed: an open traversal's completeness answers are stale
            core.traversal_memo_clear()
            logger.debug("invalidated {}", self.task_id)
        return True

    def _invalidate_meta(self):
        # Metadata (saveMeta/saveMetaJson) lives outside output(), so delete it here too.
        meta_base = self._getpath('meta')
        for ext in ('.pickle', '.json'):
            path = self._make_path_cloud_compatible(meta_base.with_suffix(ext))
            try:
                path.unlink()
                logger.debug("invalidated meta {}", path)
            except FileNotFoundError:
                pass  # no metadata was saved for this format


    def complete(self, cascade=True):
        """
        Check if a task is complete: output exists AND the stored code fingerprint
        matches the current one (``_code_ok`` -- a ``code_version`` bump makes the
        task incomplete and forces a rerun; authoritative, unlike the warn-only AST
        source-hash advisory). With ``check_dependencies``, cascades upstream.

        The cascading form is evaluated once per task per engine traversal (see
        ``core.traversal_scope``); ``cascade=False`` is never memoized.
        """
        return core.complete_cached(self, cascade, lambda: self._complete_check(cascade))

    def _complete_check(self, cascade):
        complete = super().complete()
        if complete and not getattr(self, 'external', False):
            complete = self._code_ok()
        if oryxflow.settings.check_dependencies and cascade and not getattr(self, 'external', False):
            complete = complete and all(
                [t.complete() for t in core.flatten(self.requires())])
        return complete

    def _code_ok(self):
        # record-based, two-dimensional completeness: outputs count as complete only while
        # (a) the task's OWN code identity is unchanged -- the explicit code_version when
        # pinned, the stored source hashes when auto (mode-aware, see _own_code_ok) -- and
        # (b) no dependency has rematerialized since this record (_dep_state folds dep
        # output_ids). Both dimensions live in every record, so pinning/unpinning a task
        # with genuinely unchanged code "just resumes" instead of recomputing, and never
        # ripples downstream. Inert (True) when no code identity applies here or upstream,
        # and grandfathering (no record yet) also passes -- build() stamps the baseline.
        fp = self._code_fingerprint
        if fp is None:
            return True
        from oryxflow import state, codehash
        rec = state.get_record(self._resolved_dirpath(), self.task_id)
        if rec is None:
            return True          # grandfathered; build() stamps it
        if rec.get('v') != state.RECORD_V or rec.get('py') != codehash.PY_TAG:
            # record schema or interpreter changed -> not comparable; treat as complete
            # (grandfather trust level), build()'s advisory sweep re-stamps
            return True
        if not self._own_code_ok(rec):
            return False
        return rec.get('dep_state') == self._dep_state()

    def _own_code_ok(self, rec):
        # own dimension, mode-aware: a pin (code_version) is THE version while present;
        # auto compares the source hashes as of the last materialization. Because the
        # hashes are stored on every stamp regardless of mode, removing a pin "just
        # resumes" (no recompute when the code is genuinely unchanged) yet an edit made
        # while pinned-and-unbumped is caught the moment the pin comes off -- and pinning
        # a task in the same change that edits its code forces a rerun instead of
        # blessing the stale output.
        from oryxflow import codehash
        if self.code_version is not None:
            if rec.get('code_version') == self.code_version:
                return True
            if rec.get('code_version') is None and settings.code_version_auto:
                # opting in: free iff the code really is what produced the output
                return rec.get('source_hashes') == codehash.task_hashes(self)
            return False         # pin bump (or unverifiable flip) -> recompute
        if not settings.code_version_auto:
            return True          # own logic untracked in explicit-only mode
        if rec.get('source_hashes') == codehash.task_hashes(self):
            return True
        # expensive-recompute guard: don't silently burn a long run on a code change --
        # stay complete; build()'s advisory warns with the exits (reset/accept/pin)
        thresh = getattr(settings, 'code_version_auto_expensive_s', None)
        if thresh and (rec.get('duration_s') or 0) > thresh:
            return True
        return False

    def _dep_state(self):
        # folded output identity of the direct deps: each dep's record output_id, a fresh
        # id per actual materialization that re-stamps and accept_code PRESERVE. Downstream
        # therefore reruns when an upstream actually recomputed -- and only then: mode
        # toggles and accepts upstream don't ripple, while a reset+rerun upstream
        # propagates even across separate builds.
        from oryxflow import state
        ids = []
        for d in self.deps():
            dirpath = d._resolved_dirpath() if hasattr(d, '_resolved_dirpath') \
                else settings.dirpath
            rec = state.get_record(dirpath, d.task_id)
            ids.append((rec or {}).get('output_id') or '')
        return hashlib.md5('|'.join(sorted(ids)).encode('utf-8')).hexdigest()[:16]

    def _resolved_dirpath(self):
        # the data directory this task's artifacts (and code-invalidation records) live in
        if self.path is not None:
            return pathlib.Path(self.path)
        return settings.dirpath

    # Private Get Path Function
    def _getpath(self, k, subdir=True):
        # Get Output dir
        dirpath = self._resolved_dirpath()

        # Add Group
        if hasattr(self, 'task_group'):
            dirpath = dirpath / f"/group={getattr(self, 'task_group')}"

        # Get Path
        tidroot = getattr(self, 'target_dir', self.task_id.split('_')[0])
        if getattr(self, 'keep_versions', False) and self.code_version is not None:
            tidroot = '{}/v{}'.format(
                tidroot, core.TASK_ID_INVALID_CHAR_REGEX.sub('_', str(self.code_version)))
        fname = '{}-{}'.format(self.task_id, k) if (settings.save_with_param and getattr(
            self, 'save_attrib', True)) else '{}'.format(k)
        fname += '.{}'.format(self.target_ext)
        if subdir:
            path = dirpath / tidroot / fname
        else:
            path = dirpath / fname

        # use cloud storage
        if settings.cloud_fs_enabled:
            from pathlib import PurePosixPath  # needed on windows
            path = f'{settings.cloud_fs_prefix}/{PurePosixPath(path)}'

        return path

    def output(self):
        """
        Output target(s) this task produces
        """
        save_ = getattr(self, 'persist', [])
        output = dict([(k, self.target_class(self._getpath(k)))
                       for k in save_])
        if self.persist == ['data']:  # 1 data shortcut
            output = output['data']
        return output

    def inputLoad(self, keys=None, task=None, cached=False, as_dict=False, flatten=True):
        """
        Load all or several outputs from task

        Args:
            keys (list): list of data to load
            task (str, class): if requires multiple tasks load that task 'input1' for eg `def requires: {'input1':Task1(), 'input2':Task2()}`.
                For a fan-out, name the group to get just its branches — pass the dependency
                **class** (`task=RegionNarrative`) rather than its name and a rename can't
                break it.
            cached (bool): cache data in memory
            as_dict (bool): if the inputs were saved as a dictionary. use this to return them as dictionary.
            flatten (bool): False groups a fan-out's branches under one key, so a task that
                mixes a fan-out with shared inputs can tell them apart
        Returns: list or dict of all task output
        """
        groups = core._requires_groups(self)
        task = core._group_key(task, groups)
        if task is not None and task in groups:
            # a whole fan-out group, keyed by branch
            return {label: self.inputLoad(keys=keys, task=key, cached=cached, as_dict=as_dict)
                    for key, label in groups[task].items()}

        if task is not None:
            input = self.input()[task]
        else:
            input = self.input()

        requires = self.requires()
        type_of_requires = type(requires)

        if isinstance(input, dict):
            keys = input.keys() if keys is None else keys
            data = {}
            for k, v in input.items():
                if k in keys:
                    if type(v) == dict:
                        if as_dict:
                            data[k] = {k: v.load(cached) for k, v in v.items()}
                        else:
                            data[k] = [v.load(cached) for k, v in v.items()]
                    else:
                        data[k] = v.load(cached)
            # Return DF if Single Key
            if isinstance(keys, str) and not as_dict:
                return data[keys]
            # Convert to list if dependecy is Single
            if (type_of_requires != dict or task is not None) and not as_dict:
                data = list(data.values())
        elif isinstance(input, list):
            data = []
            for _target in input:
                if isinstance(_target, dict):
                    if as_dict:
                        data.append({k: v.load(cached)
                                     for k, v in _target.items()})
                    else:
                        data.append([v.load(cached)
                                     for _, v in _target.items()])
                else:
                    data.append(_target.load(cached))
        else:
            data = input.load()

        logger.debug("loaded input for {} keys={}", self.task_id,
                     list(keys) if keys is not None else None)
        if not flatten and isinstance(data, dict) and groups:
            data = _regroup(data, groups)
        return data

    def inputLoadConcat(self, keys=None, tag=True, tagkeys=None, as_dict=False,
                        concat_fn=None, cached=False, task=None, flatten=True):
        """Load every dependency and concatenate into one DataFrame. Works for the dict form of
        requires() ({key: Task(...)}) and the list/positional form. By default each dependency's
        significant params are added as columns. concat_fn(identifier, params, df)->df overrides.

        task: concatenate only one fan-out group (the group name, or the dependency class --
        pass the class and a rename can't break it), or only one dependency.
        flatten: False returns {name: DataFrame} -- each fan-out group concatenated within
        itself, every other dependency its own entry."""
        requires = self.requires()
        if isinstance(requires, dict):
            items = list(requires.items())        # (key, task)
        elif isinstance(requires, (list, tuple)):
            items = list(enumerate(requires))     # (index, task)
        else:
            items = [(None, requires)]            # single dep
        groups = core._requires_groups(self)
        task = core._group_key(task, groups)

        def _concat(subitems):
            def _gen():
                for ident, dep in subitems:
                    data = self.inputLoad(keys=keys, task=ident, as_dict=as_dict, cached=cached)
                    params = {n: getattr(dep, n) for n in dep.get_param_names()} if tag else {}
                    yield ident, params, data
            import oryxflow.utils
            return oryxflow.utils.concat_iter(_gen(), concat_fn=concat_fn, keys=tagkeys)

        if task is not None:
            if task in groups:
                return _concat([(k, requires[k]) for k in groups[task]])
            return _concat([(task, requires[task])])

        if not flatten and groups:
            grouped = {key: name for name, labels in groups.items() for key in labels}
            out = {}
            for ident, dep in items:
                name = grouped.get(ident, ident)
                out.setdefault(name, []).append((ident, dep))
            return {name: _concat(sub) for name, sub in out.items()}

        # a shared dependency alongside a fan-out would be row-stacked in with the branches,
        # which is a union frame nobody asked for -- an exact check off the spec, not a guess
        if groups:
            branches = {key for labels in groups.values() for key in labels}
            shared = [ident for ident, _ in items if ident not in branches]
            if shared:
                import warnings
                warnings.warn(
                    "inputLoadConcat() is stacking the fan-out branches together with {} shared "
                    "dependency/dependencies ({}). Pass task='{}' to concatenate just the "
                    "fan-out, or flatten=False to get one frame per group.".format(
                        len(shared), ', '.join(map(repr, shared)), list(groups)[0]),
                    UserWarning, stacklevel=2)
        return _concat(items)

    def outputLoad(self, keys=None, as_dict=False, cached=False):
        """
        Load all or several outputs from task

        Args:
            keys (list): list of data to load
            as_dict (bool): cache data in memory
            cached (bool): cache data in memory

        Returns: list or dict of all task output
        """
        if not self.complete(cascade=False):
            raise RuntimeError(
                f'Cannot load {self.__class__}, task not complete, run flow first')

        # Check Keys is not empty
        keys = self.persist if keys is None else keys
        # Not List
        if type(keys) is not list:
            if not keys in self.persist:
                raise IndexError('Key name does not match')
        else:
            for key in keys:
                if not key in self.persist:
                    raise IndexError('Key name does not match')

        logger.debug("loaded output for {} keys={}", self.task_id,
                     keys if isinstance(keys, list) else [keys])

        if self.persist == ['data']:  # 1 data shortcut
            persist_data = self.output().load()
            return persist_data

        # Get Data
        data = {k: v.load(cached)
                for k, v in self.output().items() if k in keys}

        # Return As List
        if not as_dict:
            data = list(data.values())
        # If Keys is not a list
        if type(keys) is not list:
            data = data[0]

        # Return
        return data

    def save(self, data, from_list=False, **kwargs):
        """
        Persist data to target

        Args:
            data (dict): data to save. keys are the self.persist keys and values is data

        """

        if self.persist == ['data']:  # 1 data shortcut
            self.output().save(data, **kwargs)
        else:
            targets = self.output()
            if from_list:
                data = dict(zip(self.persist, data))
            if not set(data.keys()) == set(targets.keys()):
                raise ValueError(
                    'Save dictionary needs to consistent with Task.persist')
            for k, v in data.items():
                targets[k].save(v, **kwargs)
        logger.debug("saved {} keys={}", self.task_id, list(self.persist))
        # this task's outputs changed: an open traversal's completeness answers are stale
        core.traversal_memo_clear()

    def _get_meta_path_with_format(self, task, format='pickle'):
        """Get metadata path for a given task and format"""
        if format == 'pickle':
            return self._get_meta_path(task)
        else:  # json
            return task._getpath('meta').with_suffix('.json')

    def _make_path_cloud_compatible(self, path):
        """Convert path to cloud-compatible path if cloud storage is enabled"""
        if settings.cloud_fs_enabled:
            import upath
            return upath.UPath(path)
        return pathlib.Path(path)

    def _save_meta_internal(self, data, format='pickle'):
        """Internal method to save metadata in specified format"""
        self.metadata = data
        if format == 'pickle':
            path = self._get_meta_path(self)
            path = self._make_path_cloud_compatible(path)
            with path.open("wb") as fh:
                pickle.dump(data, fh)
        else:  # json
            path = self._getpath('meta').with_suffix('.json')
            path = self._make_path_cloud_compatible(path)
            path.parent.mkdir(exist_ok=True, parents=True)
            with path.open("w") as fh:
                json.dump(data, fh)

    def _load_meta_from_task(self, task, format='pickle'):
        """Load metadata from a single task"""
        if format == 'pickle':
            path = self._get_meta_path(task)
            path = self._make_path_cloud_compatible(path)
            with path.open("rb") as fh:
                return pickle.load(fh)
        else:  # json
            path = task._getpath('meta').with_suffix('.json')
            path = self._make_path_cloud_compatible(path)
            with path.open("r") as fh:
                return json.load(fh)

    def _input_load_meta_internal(self, key=None, format='pickle'):
        """Internal method to load metadata from input tasks"""
        inputs = self.requires()

        if key is not None:
            return self._load_meta_from_task(inputs[key], format)
        elif isinstance(inputs, dict):
            return {k: self._load_meta_from_task(v, format) for k, v in inputs.items()}
        elif isinstance(inputs, list):
            return [self._load_meta_from_task(task, format) for task in inputs]
        else:
            return self._load_meta_from_task(inputs, format)

    def metaSave(self, data):
        self._save_meta_internal(data, format='pickle')

    def saveMeta(self, data):
        self.metaSave(data)

    def saveMetaJson(self, data):
        self._save_meta_internal(data, format='json')

    def metaLoad(self, key=None):
        return self._input_load_meta_internal(key, format='pickle')

    def inputLoadMetaJson(self, key=None):
        return self._input_load_meta_internal(key, format='json')

    def outputLoadMeta(self):
        if not self.complete(cascade=False):
            raise RuntimeError(
                'Cannot load, task not complete, run flow first')
        try:
            return self._load_meta_from_task(self, format='pickle')
        except FileNotFoundError:
            raise RuntimeError(
                f"No metadata to load for task {self.task_family}")

    def outputLoadMetaJson(self):
        if not self.complete(cascade=False):
            raise RuntimeError(
                'Cannot load, task not complete, run flow first')
        try:
            return self._load_meta_from_task(self, format='json')
        except FileNotFoundError:
            raise RuntimeError(
                f"No metadata to load for task {self.task_family}")

    def outputLoadAllMeta(self):
        if not self.complete(cascade=False):
            raise RuntimeError(
                'Cannot load, task not complete, run flow first')
        tasks = oryxflow.taskflow_upstream(self, only_complete=True)
        meta = []
        for task in tasks:
            try:
                meta.append(task.outputLoadMeta())
            except:
                tasks.remove(task)
        tasks = [task.task_family for task in tasks]
        return dict(zip(tasks, meta))

    def _get_meta_path(self, task):
        # Get Meta Path
        meta_path = task._getpath('meta').with_suffix('.pickle')
        meta_path.parent.mkdir(exist_ok=True, parents=True)
        return meta_path

reset

reset(confirm=False)

Reset a task, eg by deleting output file

Source code in oryxflow/tasks/__init__.py

def reset(self, confirm=False):
    """
    Reset a task, eg by deleting output file
    """
    return self.invalidate(confirm)

invalidate

invalidate(confirm=False)

Reset a task, eg by deleting output file

Source code in oryxflow/tasks/__init__.py

def invalidate(self, confirm=False):
    """
    Reset a task, eg by deleting output file
    """
    if confirm:
        c = input(
            'Confirm invalidating task: {} (y/n). PS You can disable this message by passing confirm=False'.format(
                self.__class__.__qualname__))
    else:
        c = 'y'
    if c == 'y':  # and self.complete():
        if self.persist == ['data']:  # 1 data shortcut
            self.output().invalidate()
        else:
            [t.invalidate() for t in self.output().values()]
        self._invalidate_meta()
        # this task's outputs changed: an open traversal's completeness answers are stale
        core.traversal_memo_clear()
        logger.debug("invalidated {}", self.task_id)
    return True

complete

complete(cascade=True)

Check if a task is complete: output exists AND the stored code fingerprint matches the current one (_code_ok -- a code_version bump makes the task incomplete and forces a rerun; authoritative, unlike the warn-only AST source-hash advisory). With check_dependencies, cascades upstream.

The cascading form is evaluated once per task per engine traversal (see core.traversal_scope); cascade=False is never memoized.

Source code in oryxflow/tasks/__init__.py

def complete(self, cascade=True):
    """
    Check if a task is complete: output exists AND the stored code fingerprint
    matches the current one (``_code_ok`` -- a ``code_version`` bump makes the
    task incomplete and forces a rerun; authoritative, unlike the warn-only AST
    source-hash advisory). With ``check_dependencies``, cascades upstream.

    The cascading form is evaluated once per task per engine traversal (see
    ``core.traversal_scope``); ``cascade=False`` is never memoized.
    """
    return core.complete_cached(self, cascade, lambda: self._complete_check(cascade))

output

output()

Output target(s) this task produces

Source code in oryxflow/tasks/__init__.py

def output(self):
    """
    Output target(s) this task produces
    """
    save_ = getattr(self, 'persist', [])
    output = dict([(k, self.target_class(self._getpath(k)))
                   for k in save_])
    if self.persist == ['data']:  # 1 data shortcut
        output = output['data']
    return output

inputLoad

inputLoad(keys=None, task=None, cached=False, as_dict=False, flatten=True)

Load all or several outputs from task

Parameters:

Name	Type	Description	Default
keys	list	list of data to load	None
task	str, class	if requires multiple tasks load that task 'input1' for eg def requires: {'input1':Task1(), 'input2':Task2()}. For a fan-out, name the group to get just its branches — pass the dependency class (task=RegionNarrative) rather than its name and a rename can't break it.	None
cached	bool	cache data in memory	False
as_dict	bool	if the inputs were saved as a dictionary. use this to return them as dictionary.	False
flatten	bool	False groups a fan-out's branches under one key, so a task that mixes a fan-out with shared inputs can tell them apart	True

Returns: list or dict of all task output

Source code in oryxflow/tasks/__init__.py

def inputLoad(self, keys=None, task=None, cached=False, as_dict=False, flatten=True):
    """
    Load all or several outputs from task

    Args:
        keys (list): list of data to load
        task (str, class): if requires multiple tasks load that task 'input1' for eg `def requires: {'input1':Task1(), 'input2':Task2()}`.
            For a fan-out, name the group to get just its branches — pass the dependency
            **class** (`task=RegionNarrative`) rather than its name and a rename can't
            break it.
        cached (bool): cache data in memory
        as_dict (bool): if the inputs were saved as a dictionary. use this to return them as dictionary.
        flatten (bool): False groups a fan-out's branches under one key, so a task that
            mixes a fan-out with shared inputs can tell them apart
    Returns: list or dict of all task output
    """
    groups = core._requires_groups(self)
    task = core._group_key(task, groups)
    if task is not None and task in groups:
        # a whole fan-out group, keyed by branch
        return {label: self.inputLoad(keys=keys, task=key, cached=cached, as_dict=as_dict)
                for key, label in groups[task].items()}

    if task is not None:
        input = self.input()[task]
    else:
        input = self.input()

    requires = self.requires()
    type_of_requires = type(requires)

    if isinstance(input, dict):
        keys = input.keys() if keys is None else keys
        data = {}
        for k, v in input.items():
            if k in keys:
                if type(v) == dict:
                    if as_dict:
                        data[k] = {k: v.load(cached) for k, v in v.items()}
                    else:
                        data[k] = [v.load(cached) for k, v in v.items()]
                else:
                    data[k] = v.load(cached)
        # Return DF if Single Key
        if isinstance(keys, str) and not as_dict:
            return data[keys]
        # Convert to list if dependecy is Single
        if (type_of_requires != dict or task is not None) and not as_dict:
            data = list(data.values())
    elif isinstance(input, list):
        data = []
        for _target in input:
            if isinstance(_target, dict):
                if as_dict:
                    data.append({k: v.load(cached)
                                 for k, v in _target.items()})
                else:
                    data.append([v.load(cached)
                                 for _, v in _target.items()])
            else:
                data.append(_target.load(cached))
    else:
        data = input.load()

    logger.debug("loaded input for {} keys={}", self.task_id,
                 list(keys) if keys is not None else None)
    if not flatten and isinstance(data, dict) and groups:
        data = _regroup(data, groups)
    return data

inputLoadConcat

inputLoadConcat(keys=None, tag=True, tagkeys=None, as_dict=False, concat_fn=None, cached=False, task=None, flatten=True)

Load every dependency and concatenate into one DataFrame. Works for the dict form of requires() ({key: Task(...)}) and the list/positional form. By default each dependency's significant params are added as columns. concat_fn(identifier, params, df)->df overrides.

task: concatenate only one fan-out group (the group name, or the dependency class -- pass the class and a rename can't break it), or only one dependency. flatten: False returns {name: DataFrame} -- each fan-out group concatenated within itself, every other dependency its own entry.

Source code in oryxflow/tasks/__init__.py

def inputLoadConcat(self, keys=None, tag=True, tagkeys=None, as_dict=False,
                    concat_fn=None, cached=False, task=None, flatten=True):
    """Load every dependency and concatenate into one DataFrame. Works for the dict form of
    requires() ({key: Task(...)}) and the list/positional form. By default each dependency's
    significant params are added as columns. concat_fn(identifier, params, df)->df overrides.

    task: concatenate only one fan-out group (the group name, or the dependency class --
    pass the class and a rename can't break it), or only one dependency.
    flatten: False returns {name: DataFrame} -- each fan-out group concatenated within
    itself, every other dependency its own entry."""
    requires = self.requires()
    if isinstance(requires, dict):
        items = list(requires.items())        # (key, task)
    elif isinstance(requires, (list, tuple)):
        items = list(enumerate(requires))     # (index, task)
    else:
        items = [(None, requires)]            # single dep
    groups = core._requires_groups(self)
    task = core._group_key(task, groups)

    def _concat(subitems):
        def _gen():
            for ident, dep in subitems:
                data = self.inputLoad(keys=keys, task=ident, as_dict=as_dict, cached=cached)
                params = {n: getattr(dep, n) for n in dep.get_param_names()} if tag else {}
                yield ident, params, data
        import oryxflow.utils
        return oryxflow.utils.concat_iter(_gen(), concat_fn=concat_fn, keys=tagkeys)

    if task is not None:
        if task in groups:
            return _concat([(k, requires[k]) for k in groups[task]])
        return _concat([(task, requires[task])])

    if not flatten and groups:
        grouped = {key: name for name, labels in groups.items() for key in labels}
        out = {}
        for ident, dep in items:
            name = grouped.get(ident, ident)
            out.setdefault(name, []).append((ident, dep))
        return {name: _concat(sub) for name, sub in out.items()}

    # a shared dependency alongside a fan-out would be row-stacked in with the branches,
    # which is a union frame nobody asked for -- an exact check off the spec, not a guess
    if groups:
        branches = {key for labels in groups.values() for key in labels}
        shared = [ident for ident, _ in items if ident not in branches]
        if shared:
            import warnings
            warnings.warn(
                "inputLoadConcat() is stacking the fan-out branches together with {} shared "
                "dependency/dependencies ({}). Pass task='{}' to concatenate just the "
                "fan-out, or flatten=False to get one frame per group.".format(
                    len(shared), ', '.join(map(repr, shared)), list(groups)[0]),
                UserWarning, stacklevel=2)
    return _concat(items)

outputLoad

outputLoad(keys=None, as_dict=False, cached=False)

Load all or several outputs from task

Parameters:

Name	Type	Description	Default
keys	list	list of data to load	None
as_dict	bool	cache data in memory	False
cached	bool	cache data in memory	False

Returns: list or dict of all task output

Source code in oryxflow/tasks/__init__.py

def outputLoad(self, keys=None, as_dict=False, cached=False):
    """
    Load all or several outputs from task

    Args:
        keys (list): list of data to load
        as_dict (bool): cache data in memory
        cached (bool): cache data in memory

    Returns: list or dict of all task output
    """
    if not self.complete(cascade=False):
        raise RuntimeError(
            f'Cannot load {self.__class__}, task not complete, run flow first')

    # Check Keys is not empty
    keys = self.persist if keys is None else keys
    # Not List
    if type(keys) is not list:
        if not keys in self.persist:
            raise IndexError('Key name does not match')
    else:
        for key in keys:
            if not key in self.persist:
                raise IndexError('Key name does not match')

    logger.debug("loaded output for {} keys={}", self.task_id,
                 keys if isinstance(keys, list) else [keys])

    if self.persist == ['data']:  # 1 data shortcut
        persist_data = self.output().load()
        return persist_data

    # Get Data
    data = {k: v.load(cached)
            for k, v in self.output().items() if k in keys}

    # Return As List
    if not as_dict:
        data = list(data.values())
    # If Keys is not a list
    if type(keys) is not list:
        data = data[0]

    # Return
    return data

save

save(data, from_list=False, **kwargs)

Persist data to target

Parameters:

Name	Type	Description	Default
data	dict	data to save. keys are the self.persist keys and values is data	required

Source code in oryxflow/tasks/__init__.py

def save(self, data, from_list=False, **kwargs):
    """
    Persist data to target

    Args:
        data (dict): data to save. keys are the self.persist keys and values is data

    """

    if self.persist == ['data']:  # 1 data shortcut
        self.output().save(data, **kwargs)
    else:
        targets = self.output()
        if from_list:
            data = dict(zip(self.persist, data))
        if not set(data.keys()) == set(targets.keys()):
            raise ValueError(
                'Save dictionary needs to consistent with Task.persist')
        for k, v in data.items():
            targets[k].save(v, **kwargs)
    logger.debug("saved {} keys={}", self.task_id, list(self.persist))
    # this task's outputs changed: an open traversal's completeness answers are stale
    core.traversal_memo_clear()

TaskCache

Bases: TaskData

Task which saves to cache

Source code in oryxflow/tasks/__init__.py

class TaskCache(TaskData):
    """
    Task which saves to cache
    """
    target_class = oryxflow.targets.CacheTarget
    target_ext = 'cache'

TaskCachePandas

Bases: TaskData

Task which saves to cache pandas dataframes

Source code in oryxflow/tasks/__init__.py

class TaskCachePandas(TaskData):
    """
    Task which saves to cache pandas dataframes
    """
    target_class = oryxflow.targets.PdCacheTarget
    target_ext = 'cache'

TaskJson

Bases: TaskData

Task which saves to json

Source code in oryxflow/tasks/__init__.py

class TaskJson(TaskData):
    """
    Task which saves to json
    """
    target_class = oryxflow.targets.JsonTarget
    target_ext = 'json'

TaskPickle

Bases: TaskData

Task which saves to pickle

Source code in oryxflow/tasks/__init__.py

class TaskPickle(TaskData):
    """
    Task which saves to pickle
    """
    target_class = oryxflow.targets.PickleTarget
    target_ext = 'pkl'

TaskCSVPandas

Bases: TaskData

Task which saves to CSV

Source code in oryxflow/tasks/__init__.py

class TaskCSVPandas(TaskData):
    """
    Task which saves to CSV
    """
    target_class = oryxflow.targets.CSVPandasTarget
    target_ext = 'csv'

TaskCSVGZPandas

Bases: TaskData

Task which saves to CSV

Source code in oryxflow/tasks/__init__.py

class TaskCSVGZPandas(TaskData):
    """
    Task which saves to CSV
    """
    target_class = oryxflow.targets.CSVGZPandasTarget
    target_ext = 'csv.gz'

TaskExcelPandasSingle

Bases: TaskData

Task which saves each persist key as a separate Excel file

Source code in oryxflow/tasks/__init__.py

class TaskExcelPandasSingle(TaskData):
    """
    Task which saves each persist key as a separate Excel file
    """
    target_class = oryxflow.targets.ExcelPandasTarget
    target_ext = 'xlsx'

TaskExcelPandas

Bases: TaskData

Task which saves multiple dataframes as sheets in a single Excel file

Source code in oryxflow/tasks/__init__.py

class TaskExcelPandas(TaskData):
    """
    Task which saves multiple dataframes as sheets in a single Excel file
    """
    target_class = oryxflow.targets.ExcelPandasSheetsTarget
    target_ext = 'xlsx'

    def output(self):
        return self.target_class(self._getpath('data'))

    def save(self, data, from_list=False, **kwargs):
        if self.persist == ['data']:
            data = {'data': data}
        else:
            if from_list:
                data = dict(zip(self.persist, data))
            if not set(data.keys()) == set(self.persist):
                raise ValueError(
                    'Save dictionary needs to be consistent with Task.persist')
        self.output().save(data, **kwargs)
        # this task's outputs changed: an open traversal's completeness answers are stale
        core.traversal_memo_clear()

    def outputLoad(self, keys=None, as_dict=False, cached=False):
        if not self.complete(cascade=False):
            raise RuntimeError(
                f'Cannot load {self.__class__}, task not complete, run flow first')

        if self.persist == ['data']:
            return self.output().load(keys='data', cached=cached)

        if keys is not None:
            # Validate keys
            check_keys = [keys] if isinstance(keys, str) else keys
            for key in check_keys:
                if key not in self.persist:
                    raise IndexError('Key name does not match')

        data = self.output().load(keys=keys, cached=cached)

        # keys=str: target returns single df directly
        if isinstance(keys, str):
            return data

        # keys=None or keys=list: target returns dict
        if as_dict:
            return data
        return list(data.values())

    def invalidate(self, confirm=False):
        if confirm:
            c = input(
                'Confirm invalidating task: {} (y/n). PS You can disable this message by passing confirm=False'.format(
                    self.__class__.__qualname__))
        else:
            c = 'y'
        if c == 'y':
            self.output().invalidate()
            # this task's outputs changed: an open traversal's completeness answers are stale
            core.traversal_memo_clear()
        return True

TaskPqPandas

Bases: TaskData

Task which saves to parquet

Source code in oryxflow/tasks/__init__.py

class TaskPqPandas(TaskData):
    """
    Task which saves to parquet
    """
    target_class = oryxflow.targets.PqPandasTarget
    target_ext = 'parquet'

TaskMarkdown

Bases: TaskData

Task which saves to markdown and HTML

Source code in oryxflow/tasks/__init__.py

class TaskMarkdown(TaskData):
    """
    Task which saves to markdown and HTML
    """
    target_class = oryxflow.targets.MarkdownTarget
    target_ext = 'md'

TaskAggregator

Bases: Task

Task which groups other tasks, without saving an output of its own

Declare the group in requires() (or with @oryxflow.requires) and leave run() empty. The group is complete when every task it requires is complete.

example::

@oryxflow.requires({'ols': TaskTrainOLS, 'gbm': TaskTrainGBM})
class TaskTrainAll(oryxflow.tasks.TaskAggregator):
    pass

oryxflow.Workflow(TaskTrainAll).run()

For a one-off group that needs no task of its own, pass a list instead: oryxflow.run([TaskTrainOLS(), TaskTrainGBM()]).

Source code in oryxflow/tasks/__init__.py

class TaskAggregator(core.Task):
    """
    Task which groups other tasks, without saving an output of its own

    Declare the group in ``requires()`` (or with ``@oryxflow.requires``) and leave
    ``run()`` empty. The group is complete when every task it requires is complete.

    example::

        @oryxflow.requires({'ols': TaskTrainOLS, 'gbm': TaskTrainGBM})
        class TaskTrainAll(oryxflow.tasks.TaskAggregator):
            pass

        oryxflow.Workflow(TaskTrainAll).run()

    For a one-off group that needs no task of its own, pass a list instead:
    ``oryxflow.run([TaskTrainOLS(), TaskTrainGBM()])``.
    """

    def __init__(self, *args, path=None, flows=None, **kwargs):
        # `path`/`flows` are set by Workflow, are not Parameters, and must not reach
        # the parameter machinery -- same absorption TaskData does.
        kwargs_ = {k: v for k, v in kwargs.items(
        ) if k in self.get_param_names(include_significant=True)}
        super().__init__(*args, **kwargs_)
        if inspect.isgeneratorfunction(type(self).run):
            raise RuntimeError(
                '{}: TaskAggregator no longer yields tasks from run(). Declare the group in '
                'requires() (or @oryxflow.requires) and leave run() empty.'.format(self.task_family))
        self.path = getattr(self, 'path', path)
        self.flows = flows

    @classmethod
    def get_param_values(cls, params, args, kwargs):
        kwargs_ = {k: v for k, v in kwargs.items(
        ) if k in cls.get_param_names(include_significant=True)}
        return super(TaskAggregator, cls).get_param_values(params, args, kwargs_)

    def run(self):
        pass

    def reset(self, confirm=False):
        return self.invalidate(confirm=confirm)

    def invalidate(self, confirm=False):
        for t in self.deps():
            t.invalidate(confirm)
        return True

    def complete(self, cascade=True):
        return core.complete_cached(
            self, cascade, lambda: all(t.complete(cascade) for t in self.deps()))

    def output(self):
        return [t.output() for t in self.deps()]

    def outputLoad(self, keys=None, as_dict=False, cached=False):
        return [t.outputLoad(keys, as_dict, cached) for t in self.deps()]

Targets — oryxflow.targets

oryxflow.targets

CacheTarget

Bases: LocalTarget

Saves to in-memory cache, loads to python object

Source code in oryxflow/targets/__init__.py

class CacheTarget(core.LocalTarget):
    """
    Saves to in-memory cache, loads to python object

    """
    def __init__(self, path=None):
        super().__init__(path)
        # store as pathlib.Path so cache keys / outputPath match file-based targets
        self.path = pathlib.Path(path)

    def exists(self):
        return self.path in cache

    def invalidate(self):
        if self.path in cache:
            cache.pop(self.path)

    def load(self, cached=True):
        """
        Load from in-memory cache

        Returns: python object

        """
        if self.exists():
            return cache.get(self.path)
        else:
            raise RuntimeError('Target does not exist, make sure task is complete')


    def save(self, df):
        """
        Save dataframe to in-memory cache

        Args:
            df (obj): pandas dataframe

        Returns: filename

        """
        cache[self.path] = df
        return self.path

load

load(cached=True)

Load from in-memory cache

Returns: python object

Source code in oryxflow/targets/__init__.py

def load(self, cached=True):
    """
    Load from in-memory cache

    Returns: python object

    """
    if self.exists():
        return cache.get(self.path)
    else:
        raise RuntimeError('Target does not exist, make sure task is complete')

save

save(df)

Save dataframe to in-memory cache

Parameters:

Name	Type	Description	Default
df	obj	pandas dataframe	required

Returns: filename

Source code in oryxflow/targets/__init__.py

def save(self, df):
    """
    Save dataframe to in-memory cache

    Args:
        df (obj): pandas dataframe

    Returns: filename

    """
    cache[self.path] = df
    return self.path

DataTarget

Bases: _LocalPathTarget

Local target which saves in-memory data (eg dataframes) to persistent storage (eg files) and loads from storage to memory

This is an abstract class that you should extend.

Source code in oryxflow/targets/__init__.py

class DataTarget(_LocalPathTarget):
    """
    Local target which saves in-memory data (eg dataframes) to persistent storage (eg files) and loads from storage to memory

    This is an abstract class that you should extend.

    """
    def load(self, fun, cached=False, **kwargs):
        """
        Runs a function to load data from storage into memory

        Args:
            fun (function): loading function
            cached (bool): keep data cached in memory
            **kwargs: arguments to pass to `fun`

        Returns: data object

        """
        if self.exists():
            if not cached or not settings.cached or self.path not in cache:
                opts = {**{},**kwargs}
                df = fun(self.path, **opts)
                if cached or settings.cached:
                    cache[self.path] = df
                return df
            else:
                return cache.get(self.path)
        else:
            raise RuntimeError('Target does not exist, make sure task is complete')

    def save(self, df, fun, **kwargs):
        """
        Runs a function to save data from memory into storage

        Args:
            df (obj): data to save
            fun (function): saving function
            **kwargs: arguments to pass to `fun`

        Returns: filename

        """
        fun = getattr(df, fun)
        (self.path).parent.mkdir(parents=True, exist_ok=True)
        fun(self.path, **kwargs)
        return self.path

load

load(fun, cached=False, **kwargs)

Runs a function to load data from storage into memory

Parameters:

Name	Type	Description	Default
fun	function	loading function	required
cached	bool	keep data cached in memory	False
**kwargs		arguments to pass to fun	{}

Returns: data object

Source code in oryxflow/targets/__init__.py

def load(self, fun, cached=False, **kwargs):
    """
    Runs a function to load data from storage into memory

    Args:
        fun (function): loading function
        cached (bool): keep data cached in memory
        **kwargs: arguments to pass to `fun`

    Returns: data object

    """
    if self.exists():
        if not cached or not settings.cached or self.path not in cache:
            opts = {**{},**kwargs}
            df = fun(self.path, **opts)
            if cached or settings.cached:
                cache[self.path] = df
            return df
        else:
            return cache.get(self.path)
    else:
        raise RuntimeError('Target does not exist, make sure task is complete')

save

save(df, fun, **kwargs)

Runs a function to save data from memory into storage

Parameters:

Name	Type	Description	Default
df	obj	data to save	required
fun	function	saving function	required
**kwargs		arguments to pass to fun	{}

Returns: filename

Source code in oryxflow/targets/__init__.py

def save(self, df, fun, **kwargs):
    """
    Runs a function to save data from memory into storage

    Args:
        df (obj): data to save
        fun (function): saving function
        **kwargs: arguments to pass to `fun`

    Returns: filename

    """
    fun = getattr(df, fun)
    (self.path).parent.mkdir(parents=True, exist_ok=True)
    fun(self.path, **kwargs)
    return self.path

CSVPandasTarget

Bases: DataTarget

Saves to CSV, loads to pandas dataframe

Source code in oryxflow/targets/__init__.py

class CSVPandasTarget(DataTarget):
    """
    Saves to CSV, loads to pandas dataframe

    """
    def load(self, cached=False, **kwargs):
        """
        Load from csv to pandas dataframe

        Args:
            cached (bool): keep data cached in memory
            **kwargs: arguments to pass to pd.read_csv

        Returns: pandas dataframe

        """
        return super().load(pd.read_csv, cached, **kwargs)


    def save(self, df, **kwargs):
        """
        Save dataframe to csv

        Args:
            df (obj): pandas dataframe
            **kwargs (dict): additional arguments to pass to df.to_csv

        Returns: filename

        """
        opts = {**{'index':False},**kwargs}
        return super().save(df, 'to_csv', **opts)

load

load(cached=False, **kwargs)

Load from csv to pandas dataframe

Parameters:

Name	Type	Description	Default
cached	bool	keep data cached in memory	False
**kwargs		arguments to pass to pd.read_csv	{}

Returns: pandas dataframe

Source code in oryxflow/targets/__init__.py

def load(self, cached=False, **kwargs):
    """
    Load from csv to pandas dataframe

    Args:
        cached (bool): keep data cached in memory
        **kwargs: arguments to pass to pd.read_csv

    Returns: pandas dataframe

    """
    return super().load(pd.read_csv, cached, **kwargs)

save

save(df, **kwargs)

Save dataframe to csv

Parameters:

Name	Type	Description	Default
df	obj	pandas dataframe	required
**kwargs	dict	additional arguments to pass to df.to_csv	{}

Returns: filename

Source code in oryxflow/targets/__init__.py

def save(self, df, **kwargs):
    """
    Save dataframe to csv

    Args:
        df (obj): pandas dataframe
        **kwargs (dict): additional arguments to pass to df.to_csv

    Returns: filename

    """
    opts = {**{'index':False},**kwargs}
    return super().save(df, 'to_csv', **opts)

CSVGZPandasTarget

Bases: CSVPandasTarget

Saves to CSV gzip, loads to pandas dataframe

Source code in oryxflow/targets/__init__.py

class CSVGZPandasTarget(CSVPandasTarget):
    """
    Saves to CSV gzip, loads to pandas dataframe

    """


    def save(self, df, **kwargs):
        """
        Save dataframe to csv gzip

        Args:
            df (obj): pandas dataframe
            **kwargs (dict): additional arguments to pass to df.to_csv

        Returns: filename

        """
        opts = {**{'index':False, 'compression':'gzip'},**kwargs}
        return super().save(df, 'to_csv', **opts)

save

save(df, **kwargs)

Save dataframe to csv gzip

Parameters:

Name	Type	Description	Default
df	obj	pandas dataframe	required
**kwargs	dict	additional arguments to pass to df.to_csv	{}

Returns: filename

Source code in oryxflow/targets/__init__.py

def save(self, df, **kwargs):
    """
    Save dataframe to csv gzip

    Args:
        df (obj): pandas dataframe
        **kwargs (dict): additional arguments to pass to df.to_csv

    Returns: filename

    """
    opts = {**{'index':False, 'compression':'gzip'},**kwargs}
    return super().save(df, 'to_csv', **opts)

ExcelPandasTarget

Bases: DataTarget

Saves to Excel, loads to pandas dataframe

Source code in oryxflow/targets/__init__.py

class ExcelPandasTarget(DataTarget):
    """
    Saves to Excel, loads to pandas dataframe

    """
    def load(self, cached=False, **kwargs):
        """
        Load from Excel to pandas dataframe

        Args:
            cached (bool): keep data cached in memory
            **kwargs: arguments to pass to pd.read_csv

        Returns: pandas dataframe

        """
        return super().load(pd.read_excel, cached, **kwargs)


    def save(self, df, **kwargs):
        """
        Save dataframe to Excel

        Args:
            df (obj): pandas dataframe
            **kwargs (dict): additional arguments to pass to df.to_csv

        Returns: filename

        """
        opts = {**{'index':False},**kwargs}
        return super().save(df, 'to_excel', **opts)

load

load(cached=False, **kwargs)

Load from Excel to pandas dataframe

Parameters:

Name	Type	Description	Default
cached	bool	keep data cached in memory	False
**kwargs		arguments to pass to pd.read_csv	{}

Returns: pandas dataframe

Source code in oryxflow/targets/__init__.py

def load(self, cached=False, **kwargs):
    """
    Load from Excel to pandas dataframe

    Args:
        cached (bool): keep data cached in memory
        **kwargs: arguments to pass to pd.read_csv

    Returns: pandas dataframe

    """
    return super().load(pd.read_excel, cached, **kwargs)

save

save(df, **kwargs)

Save dataframe to Excel

Parameters:

Name	Type	Description	Default
df	obj	pandas dataframe	required
**kwargs	dict	additional arguments to pass to df.to_csv	{}

Returns: filename

Source code in oryxflow/targets/__init__.py

def save(self, df, **kwargs):
    """
    Save dataframe to Excel

    Args:
        df (obj): pandas dataframe
        **kwargs (dict): additional arguments to pass to df.to_csv

    Returns: filename

    """
    opts = {**{'index':False},**kwargs}
    return super().save(df, 'to_excel', **opts)

ExcelPandasSheetsTarget

Bases: _LocalPathTarget

Saves dict of dataframes as sheets in a single Excel file, loads selectively by sheet

Source code in oryxflow/targets/__init__.py

class ExcelPandasSheetsTarget(_LocalPathTarget):
    """
    Saves dict of dataframes as sheets in a single Excel file, loads selectively by sheet

    """
    def load(self, keys=None, cached=False, **kwargs):
        """
        Load sheets from Excel file

        Args:
            keys (str/list): sheet name(s) to load. None loads all sheets
            cached (bool): keep data cached in memory
            **kwargs: arguments to pass to pd.read_excel

        Returns: dict of dataframes, single dataframe, or filtered dict

        """
        if self.exists():
            if not cached or not settings.cached or self.path not in cache:
                sheet_name = keys if keys is not None else None
                data = pd.read_excel(self.path, sheet_name=sheet_name, **kwargs)
                if cached or settings.cached:
                    cache[self.path] = data
                return data
            else:
                data = cache.get(self.path)
                if keys is None:
                    return data
                if isinstance(keys, str):
                    return data[keys]
                return {k: v for k, v in data.items() if k in keys}
        else:
            raise RuntimeError('Target does not exist, make sure task is complete')

    def save(self, data, **kwargs):
        """
        Save dict of dataframes as sheets in a single Excel file

        Args:
            data (dict): {sheet_name: dataframe}
            kwargs: additional arguments to pass to df.to_excel

        Returns: filename

        """
        opts = {**{'index': False}, **kwargs}
        (self.path).parent.mkdir(parents=True, exist_ok=True)
        with pd.ExcelWriter(self.path, engine='openpyxl') as writer:
            for sheet_name, df in data.items():
                df.to_excel(writer, sheet_name=sheet_name, **opts)
        return self.path

load

load(keys=None, cached=False, **kwargs)

Load sheets from Excel file

Parameters:

Name	Type	Description	Default
keys	str / list	sheet name(s) to load. None loads all sheets	None
cached	bool	keep data cached in memory	False
**kwargs		arguments to pass to pd.read_excel	{}

Returns: dict of dataframes, single dataframe, or filtered dict

Source code in oryxflow/targets/__init__.py

def load(self, keys=None, cached=False, **kwargs):
    """
    Load sheets from Excel file

    Args:
        keys (str/list): sheet name(s) to load. None loads all sheets
        cached (bool): keep data cached in memory
        **kwargs: arguments to pass to pd.read_excel

    Returns: dict of dataframes, single dataframe, or filtered dict

    """
    if self.exists():
        if not cached or not settings.cached or self.path not in cache:
            sheet_name = keys if keys is not None else None
            data = pd.read_excel(self.path, sheet_name=sheet_name, **kwargs)
            if cached or settings.cached:
                cache[self.path] = data
            return data
        else:
            data = cache.get(self.path)
            if keys is None:
                return data
            if isinstance(keys, str):
                return data[keys]
            return {k: v for k, v in data.items() if k in keys}
    else:
        raise RuntimeError('Target does not exist, make sure task is complete')

save

save(data, **kwargs)

Save dict of dataframes as sheets in a single Excel file

Parameters:

Name	Type	Description	Default
data	dict	{sheet_name: dataframe}	required
kwargs		additional arguments to pass to df.to_excel	{}

Returns: filename

Source code in oryxflow/targets/__init__.py

def save(self, data, **kwargs):
    """
    Save dict of dataframes as sheets in a single Excel file

    Args:
        data (dict): {sheet_name: dataframe}
        kwargs: additional arguments to pass to df.to_excel

    Returns: filename

    """
    opts = {**{'index': False}, **kwargs}
    (self.path).parent.mkdir(parents=True, exist_ok=True)
    with pd.ExcelWriter(self.path, engine='openpyxl') as writer:
        for sheet_name, df in data.items():
            df.to_excel(writer, sheet_name=sheet_name, **opts)
    return self.path

PqPandasTarget

Bases: DataTarget

Saves to parquet, loads to pandas dataframe

Source code in oryxflow/targets/__init__.py

class PqPandasTarget(DataTarget):
    """
    Saves to parquet, loads to pandas dataframe

    """
    def load(self, cached=False, **kwargs):
        """
        Load from parquet to pandas dataframe

        Args:
            cached (bool): keep data cached in memory
            **kwargs: arguments to pass to pd.read_parquet

        Returns: pandas dataframe

        """
        return super().load(pd.read_parquet, cached, **kwargs)


    def save(self, df, **kwargs):
        """
        Save dataframe to parquet

        Args:
            df (obj): pandas dataframe
            **kwargs (dict): additional arguments to pass to df.to_parquet

        Returns: filename

        """
        opts = {**{'compression':'gzip','engine':'pyarrow'},**kwargs}
        return super().save(df, 'to_parquet', **opts)

load

load(cached=False, **kwargs)

Load from parquet to pandas dataframe

Parameters:

Name	Type	Description	Default
cached	bool	keep data cached in memory	False
**kwargs		arguments to pass to pd.read_parquet	{}

Returns: pandas dataframe

Source code in oryxflow/targets/__init__.py

def load(self, cached=False, **kwargs):
    """
    Load from parquet to pandas dataframe

    Args:
        cached (bool): keep data cached in memory
        **kwargs: arguments to pass to pd.read_parquet

    Returns: pandas dataframe

    """
    return super().load(pd.read_parquet, cached, **kwargs)

save

save(df, **kwargs)

Save dataframe to parquet

Parameters:

Name	Type	Description	Default
df	obj	pandas dataframe	required
**kwargs	dict	additional arguments to pass to df.to_parquet	{}

Returns: filename

Source code in oryxflow/targets/__init__.py

def save(self, df, **kwargs):
    """
    Save dataframe to parquet

    Args:
        df (obj): pandas dataframe
        **kwargs (dict): additional arguments to pass to df.to_parquet

    Returns: filename

    """
    opts = {**{'compression':'gzip','engine':'pyarrow'},**kwargs}
    return super().save(df, 'to_parquet', **opts)

JsonTarget

Bases: DataTarget

Saves to json, loads to dict

Source code in oryxflow/targets/__init__.py

class JsonTarget(DataTarget):
    """
    Saves to json, loads to dict

    """
    def load(self, cached=False, **kwargs):
        """
        Load from json to dict

        Args:
            cached (bool): keep data cached in memory
            **kwargs: arguments to pass to json.load

        Returns: dict

        """
        def read_json(path, **opts):
            with path.open('r') as fhandle:
                df = json.load(fhandle)
            return df['data']
        return super().load(read_json, cached, **kwargs)


    def save(self, dict_, **kwargs):
        """
        Save dict to json

        Args:
            dict_ (dict): python dict
            **kwargs (dict): additional arguments to pass to json.dump

        Returns: filename

        """
        def write_json(path, _dict_, **opts):
            with path.open('w') as fhandle:
                json.dump(_dict_, fhandle, **opts)
        opts = {**{'indent':4},**kwargs}
        write_json(self.path, {'data':dict_}, **opts)
        return self.path

load

load(cached=False, **kwargs)

Load from json to dict

Parameters:

Name	Type	Description	Default
cached	bool	keep data cached in memory	False
**kwargs		arguments to pass to json.load	{}

Returns: dict

Source code in oryxflow/targets/__init__.py

def load(self, cached=False, **kwargs):
    """
    Load from json to dict

    Args:
        cached (bool): keep data cached in memory
        **kwargs: arguments to pass to json.load

    Returns: dict

    """
    def read_json(path, **opts):
        with path.open('r') as fhandle:
            df = json.load(fhandle)
        return df['data']
    return super().load(read_json, cached, **kwargs)

save

save(dict_, **kwargs)

Save dict to json

Parameters:

Name	Type	Description	Default
dict_	dict	python dict	required
**kwargs	dict	additional arguments to pass to json.dump	{}

Returns: filename

Source code in oryxflow/targets/__init__.py

def save(self, dict_, **kwargs):
    """
    Save dict to json

    Args:
        dict_ (dict): python dict
        **kwargs (dict): additional arguments to pass to json.dump

    Returns: filename

    """
    def write_json(path, _dict_, **opts):
        with path.open('w') as fhandle:
            json.dump(_dict_, fhandle, **opts)
    opts = {**{'indent':4},**kwargs}
    write_json(self.path, {'data':dict_}, **opts)
    return self.path

MarkdownTarget

Bases: DataTarget

Saves to markdown (.md) and HTML (.html), loads markdown string

Source code in oryxflow/targets/__init__.py

class MarkdownTarget(DataTarget):
    """
    Saves to markdown (.md) and HTML (.html), loads markdown string

    """
    def load(self, cached=False, **kwargs):
        """
        Load from markdown file to string

        Args:
            cached (bool): keep data cached in memory
            **kwargs: arguments to pass to read function

        Returns: markdown string

        """
        def read_md(path, **opts):
            with path.open('r', encoding='utf-8') as fhandle:
                return fhandle.read()
        return super().load(read_md, cached, **kwargs)

    def save(self, md_string, **kwargs):
        """
        Save markdown string to .md and .html files

        Args:
            md_string (str): markdown string
            **kwargs (dict): additional arguments to pass to markdown.markdown

        Returns: filename

        """
        (self.path).parent.mkdir(parents=True, exist_ok=True)
        with self.path.open('w', encoding='utf-8') as fhandle:
            fhandle.write(md_string)
        html_path = self.path.with_suffix('.html')
        html_body = markdown.markdown(md_string, extensions=['tables'], **kwargs)
        html_string = (
            '<!DOCTYPE html>\n'
            '<html>\n'
            '<head>\n'
            '<meta charset="utf-8">\n'
            '<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/github-markdown-css/5.8.1/github-markdown.css">\n'
            '</head>\n'
            '<body>\n'
            '<article class="markdown-body">\n'
            f'{html_body}\n'
            '</article>\n'
            '</body>\n'
            '</html>\n'
        )
        with html_path.open('w', encoding='utf-8') as fhandle:
            fhandle.write(html_string)
        return self.path

    def invalidate(self):
        html_path = self.path.with_suffix('.html')
        if html_path.exists():
            html_path.unlink()
        if self.exists():
            self.path.unlink()
        return not self.exists()

load

load(cached=False, **kwargs)

Load from markdown file to string

Parameters:

Name	Type	Description	Default
cached	bool	keep data cached in memory	False
**kwargs		arguments to pass to read function	{}

Returns: markdown string

Source code in oryxflow/targets/__init__.py

def load(self, cached=False, **kwargs):
    """
    Load from markdown file to string

    Args:
        cached (bool): keep data cached in memory
        **kwargs: arguments to pass to read function

    Returns: markdown string

    """
    def read_md(path, **opts):
        with path.open('r', encoding='utf-8') as fhandle:
            return fhandle.read()
    return super().load(read_md, cached, **kwargs)

save

save(md_string, **kwargs)

Save markdown string to .md and .html files

Parameters:

Name	Type	Description	Default
md_string	str	markdown string	required
**kwargs	dict	additional arguments to pass to markdown.markdown	{}

Returns: filename

Source code in oryxflow/targets/__init__.py

def save(self, md_string, **kwargs):
    """
    Save markdown string to .md and .html files

    Args:
        md_string (str): markdown string
        **kwargs (dict): additional arguments to pass to markdown.markdown

    Returns: filename

    """
    (self.path).parent.mkdir(parents=True, exist_ok=True)
    with self.path.open('w', encoding='utf-8') as fhandle:
        fhandle.write(md_string)
    html_path = self.path.with_suffix('.html')
    html_body = markdown.markdown(md_string, extensions=['tables'], **kwargs)
    html_string = (
        '<!DOCTYPE html>\n'
        '<html>\n'
        '<head>\n'
        '<meta charset="utf-8">\n'
        '<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/github-markdown-css/5.8.1/github-markdown.css">\n'
        '</head>\n'
        '<body>\n'
        '<article class="markdown-body">\n'
        f'{html_body}\n'
        '</article>\n'
        '</body>\n'
        '</html>\n'
    )
    with html_path.open('w', encoding='utf-8') as fhandle:
        fhandle.write(html_string)
    return self.path

PickleTarget

Bases: DataTarget

Saves to pickle, loads to python obj

Source code in oryxflow/targets/__init__.py

class PickleTarget(DataTarget):
    """
    Saves to pickle, loads to python obj

    """
    def load(self, cached=False, **kwargs):
        """
        Load from pickle to obj

        Args:
            cached (bool): keep data cached in memory
            **kwargs: arguments to pass to pickle.load

        Returns: dict

        """
        def funload(x):
            with x.open("rb" ) as fhandle:
                data = pickle.load(fhandle)
            return data
        return super().load(funload, cached, **kwargs)


    def save(self, obj, **kwargs):
        """
        Save obj to pickle

        Args:
            obj (obj): python object
            **kwargs (dict): additional arguments to pass to pickle.dump

        Returns: filename

        """
        with self.path.open("wb") as fhandle:
            pickle.dump(obj, fhandle, **kwargs)
        return self.path

load

load(cached=False, **kwargs)

Load from pickle to obj

Parameters:

Name	Type	Description	Default
cached	bool	keep data cached in memory	False
**kwargs		arguments to pass to pickle.load	{}

Returns: dict

Source code in oryxflow/targets/__init__.py

def load(self, cached=False, **kwargs):
    """
    Load from pickle to obj

    Args:
        cached (bool): keep data cached in memory
        **kwargs: arguments to pass to pickle.load

    Returns: dict

    """
    def funload(x):
        with x.open("rb" ) as fhandle:
            data = pickle.load(fhandle)
        return data
    return super().load(funload, cached, **kwargs)

save

save(obj, **kwargs)

Save obj to pickle

Parameters:

Name	Type	Description	Default
obj	obj	python object	required
**kwargs	dict	additional arguments to pass to pickle.dump	{}

Returns: filename

Source code in oryxflow/targets/__init__.py

def save(self, obj, **kwargs):
    """
    Save obj to pickle

    Args:
        obj (obj): python object
        **kwargs (dict): additional arguments to pass to pickle.dump

    Returns: filename

    """
    with self.path.open("wb") as fhandle:
        pickle.dump(obj, fhandle, **kwargs)
    return self.path