API Reference¶

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

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.
        Parameters ``cls`` has bound (an imported task: its exported defaults, or ``bind()``)
        are not carried over -- only ``kwargs`` override them.
        """
        if cls is None:
            cls = self.__class__

        bound = getattr(cls, '_oryxflow_bound', ())
        new_k = {}
        for param_name, param_class in cls.get_params():
            if param_name in kwargs:
                new_k[param_name] = kwargs[param_name]
            elif param_name not in bound and 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):
                try:
                    values = values(self)
                except AttributeError as e:
                    missing = getattr(e, 'name', None)
                    if (missing in grid and getattr(e, 'obj', None) is self
                            and not hasattr(self, missing)):
                        # @requires_each keeps fanned parameters off this task by design (it
                        # would carry one branch's value), so a grid computed from this task's
                        # own value of that parameter -- "the same loader, previous quarter"
                        # -- can't read it
                        raise TypeError(
                            "{0}: the {1!r} values for {2} are computed from this task's own "
                            "'{1}', but a task that fans out over '{1}' with @requires_each "
                            "does not have that parameter. To depend on {2} at another value of "
                            "your own '{1}' (e.g. the previous period): replace the decorators "
                            "with @oryxflow.inherits(<the same tasks>) so {0} keeps '{1}', and "
                            "write requires() by hand -- {{'current': self.clone(<Task>), "
                            "'prior': self.clone({2}, {1}=prior(self.{1}))}}, or "
                            "self.requires_grid({2}, {1}=[...]) for several values.".format(
                                self.task_family, missing, cls.__name__)) from e
                    raise
            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 = _resolve_code_version(self)
        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

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

    def exists(self):
        raise NotImplementedError

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

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,
        })

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})

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

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 output -- ...'
    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")

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)

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()