Skip to content

Writing and Managing Tasks

What are tasks?

A task is one step of your analysis — load the data, clean it, train the model — packaged so oryxflow can manage it for you. Instead of a loose function plus a hand-written line to read its input file and another to write its output, you declare what a task depends on and what it produces, and the engine handles the rest: it runs upstream steps first so nothing is computed from stale input, hands each task its inputs already loaded, and reuses anything already computed rather than redoing it. Tasks are the main object you will be interacting with. They let you:

  • define input dependency tasks — so you declare the pipeline order once instead of re-wiring it every run

  • process data

  • load input data from upstream tasks — already loaded, no manual file paths

  • save output data for downstream tasks — so results are cached and reused, not recomputed

  • run tasks — the engine runs only what's missing

  • load output data — fetch any result by referencing the task that made it

You write your own tasks by inheriting from one of the predefined oryxflow task formats, for example pandas dataframes saved to parquet. Picking the parent class is how you choose the output format (parquet, CSV, pickle, in-memory, ...) without writing any save/load code yourself — see Targets.

class YourTask(oryxflow.tasks.TaskPqPandas):

Additional details on how to write tasks is below. To run tasks see Running Workflows.

Define Upstream Dependency Tasks

You can define input dependencies by using a @oryxflow.requires decorator which takes input tasks. You can have no, one or multiple input tasks. This may be required when the decorator shortcut does not work.

Tip

The Claude Code plugin writes this wiring for you - ask it to "add a task that takes <Upstream>'s output" and it emits the task class with the correct @oryxflow.requires decorator.

# no dependency
class TaskSingleInput(oryxflow.tasks.TaskPqPandas):
    #[...]

# single dependency
@oryxflow.requires(TaskSingleOutput)
class TaskSingleInput(oryxflow.tasks.TaskPqPandas):
    #[...]

# multiple dependencies
@oryxflow.requires({'input1':TaskSingleOutput1, 'input2':TaskSingleOutput2})
class TaskMultipleInput(oryxflow.tasks.TaskPqPandas):
    #[...]

# one dependency per item in a list -- one TaskPerRegion per region
@oryxflow.requires_each(TaskPerRegion, region=cfg.REGIONS)
class TaskCombineRegions(oryxflow.tasks.TaskPqPandas):
    #[...]

# a fan-out plus a shared input: decorators stack, in any order
@oryxflow.requires({'input':TaskSingleOutput})
@oryxflow.requires_each(TaskPerRegion, region=cfg.REGIONS)
class TaskReport(oryxflow.tasks.TaskPqPandas):
    #[...]

The last two are how you avoid writing out one dependency per region by hand. @oryxflow.requires_each declares one dependency per value, so adding a region to the list adds a task; stack it with @oryxflow.requires when the combining task also needs something shared that isn't fanned out — the table the per-region work was built from, a baseline to compare against. Both are covered in Dynamic Workflow Generation, which also has a decision test for whether a given loop wants this, a WorkflowMulti, or nothing at all.

Process Data

You process data by writing a run() function. This function will take input data, process it and save output data.

class YourTask(oryxflow.tasks.TaskPqPandas):

    def run(self):
        # load input data
        # process data
        # save data

Load Input Data

Input data from upstream dependency tasks can be easily loaded in run()

# no dependency
class TaskNoInput(oryxflow.tasks.TaskPqPandas):

    def run(self):
        data = pd.read_csv(oryxflow.settings.dirpath/'file.csv') # data/file.csv

# single dependency, single output
@oryxflow.requires(TaskSingleOutput)
class TaskSingleInput(oryxflow.tasks.TaskPqPandas):
    def run(self):
        data = self.inputLoad()

# single dependency, multiple outputs
@oryxflow.requires(TaskMultipleOutput)
class TaskSingleInput(oryxflow.tasks.TaskPqPandas):
    def run(self):
        data1, data2 = self.inputLoad()  # load all outputs
        # or load just one specific output by its persists name
        data1 = self.inputLoad(keys='output1')
        # equivalent lower-level spelling
        data1 = self.input()['output1'].load()

# multiple dependencies, single output
# prefer the named-dict form: you select deps by meaningful name, not by position
@oryxflow.requires({'input1':TaskSingleOutput1, 'input2':TaskSingleOutput2})
class TaskMultipleInput(oryxflow.tasks.TaskPqPandas):
    def run(self):
        data1 = self.inputLoad()['input1']
        data2 = self.inputLoad()['input2']
        # or
        data1 = self.inputLoad(task='input1')
        data2 = self.inputLoad(task='input2')

# multiple dependencies, multiple outputs
@oryxflow.requires({'input1':TaskMultipleOutput1, 'input2':TaskMultipleOutput2})
class TaskMultipleInput(oryxflow.tasks.TaskPqPandas):
    def run(self):
        data = self.inputLoad(as_dict=True)
        data1a = data['input1']['output1']
        data1b = data['input1']['output2']
        data2a = data['input2']['output1']
        data2b = data['input2']['output2']
        # or
        data1a, data1b = self.inputLoad()["input1"]
        data2a, data2b = self.inputLoad()["input2"]
        # or
        data1a, data1b = self.inputLoad(task='input1')
        data2a, data2b = self.inputLoad(task='input2')

# multiple dependencies (positional, without a dictionary), multiple outputs
# works, but the named-dict form above is preferred — here deps are selected by
# integer position (0, 1) instead of by name
@oryxflow.requires(TaskMultipleOutput1, TaskMultipleOutput2)
class TaskMultipleInput(oryxflow.tasks.TaskPqPandas):
    def run(self):
        data = self.inputLoad(as_dict=True)
        data1a = data[0]['output1']
        data1b = data[0]['output2']
        data2a = data[1]['output1']
        data2b = data[1]['output2']
        # or
        data1a, data1b = self.inputLoad()[0]
        data2a, data2b = self.inputLoad()[1]
        # or
        data1a, data1b = self.inputLoad(task=0)
        data2a, data2b = self.inputLoad(task=1)

# one dependency per item, plus a shared input
@oryxflow.requires({'input':TaskSingleOutput})
@oryxflow.requires_each(TaskPerRegion, region=cfg.REGIONS)
class TaskReport(oryxflow.tasks.TaskPqPandas):
    def run(self):
        deps = self.inputLoad(flatten=False)   # keeps the two kinds apart
        shared = deps['input']
        for region, data in deps['TaskPerRegion'].items():
            pass
        # or get just the per-region ones
        by_region = self.inputLoad(task='TaskPerRegion')
        # or stack them into one frame, tagged by region
        df = self.inputLoadConcat(task='TaskPerRegion')

Load External Files

You probably want to load external data which is not the output of a task. There are a few options.

class TaskExternalData(oryxflow.tasks.TaskPqPandas):

    def run(self):

        import pandas as pd
        # read from oryxflow data folder
        data = pd.read_parquet(oryxflow.settings.dirpath/'file.pq')

        # totally manual
        data = pd.read_parquet('/some/folder/file.pq')

        # multiple files
        data = pd.concat([pd.read_csv(f) for f in glob.glob('*.csv')], ignore_index=True)

For more advanced options see Sharing Workflows and Outputs

Dynamic Inputs

When the number of inputs comes from a list — one per region, per file, per model — you write the list once instead of writing out the dependencies. @oryxflow.requires_each(TaskLoadFile, file=glob.glob('data-raw/*.csv')) gives you one task per file, each cached on its own, so dropping a new CSV in the folder reads only that one.

See Dynamic Workflow Generation, and start with its decision test if you're not sure whether a loop you already have wants this at all.

Save Output Data

Saving output data is quick and convenient. You can save a single or multiple outputs.

# quick save one output
class TaskSingleOutput(oryxflow.tasks.TaskPqPandas):

    def run(self):
        self.save(data_output)

# save more than one output
class TaskMultipleOutput(oryxflow.tasks.TaskPqPandas):
    persists=['output1','output2'] # declare what you will save

    def run(self):
        self.save({'output1':data1, 'output2':data2}) # needs to match persists

persist (singular) is a backwards-compatible alias for persists; prefer persists.

When you have multiple outputs and don't declare persists you will get raise ValueError('Save dictionary needs to consistent with Task.persist')

What can I save?

Anything — you are not limited to dataframes. self.save() accepts whatever the task's parent class handles, so if your result is a dict, save it as JSON; if it's a trained model, save it as pickle. There's no need to force a non-tabular result into a dataframe.

# a dict saved as JSON, and loaded back as a dict
class TaskConfig(oryxflow.tasks.TaskJson):

    def run(self):
        self.save({'features': ['x1', 'x2'], 'nrows': 100})

# any python object saved as pickle
class TaskModel(oryxflow.tasks.TaskPickle):

    def run(self):
        self.save(model)

See Targets for the full list of formats and which class saves what.

Where Is Output Data Saved?

Output data by default is saved in data/, you can check with

oryxflow.settings.dirpath # folder where workflow output is saved
TaskTrain().output().path # file where task output is saved

You can change where data is saved using oryxflow.set_dir('data/'). See advanced options for Sharing Workflows and Outputs Global Data Path can be also changed by including the path parameter to the Workflow.

Changing Task Output Formats

See Targets

Running tasks

See Running Workflows

Load Output Data

Once a workflow is run and the task is complete, you can easily load its output data by referencing the task.

data = flow.outputLoad() # load default task output
data = flow.outputLoad(as_dict=True) # useful for multi output
data2 = flow.outputLoad(TaskMultipleOutput, as_dict=True) # load another task output
data2['data1']
data2['data2']

Before you load output data you need to run the workflow. See run the workflow. If a task has not been run, it will show

raise RuntimeError('Target does not exist, make sure task is complete')
RuntimeError: Target does not exist, make sure task is complete

Which load method: output().load() vs outputLoad() vs outputLoadConcat()

There are three ways to get at a task's output, lowest- to highest-level. Prefer the highest one that fits — most code should just use outputLoad().

  • task.output().load() — lowest level. output() returns the target object (or a dict of targets for a multi-persists task), and .load() reads it. Reach for this only when you want the target itself — its .path, or a deliberate .load(). You index the target yourself, e.g. task.output()['train'].load().
  • task.outputLoad() / flow.outputLoad() — the default; use this to fetch results. Returns the data directly: a single object for a single-persists task, a list (or a dict with as_dict=True) for multiple outputs, and a {flow: data} dict for WorkflowMulti. It also checks the task is complete for you.
  • flow.outputLoadConcat() — narrow and opt-in; WorkflowMulti only. Row-stacks every flow's output into one DataFrame, tagging each flow's rows with its params. Use it only when every flow's output is a schema-compatible DataFrame you want combined (the iterate-and-aggregate case). It is a separate method on purpose: it collapses the per-flow {flow: data} dict into a single frame — a different operation from outputLoad, and one that is meaningless for non-DataFrame outputs like models. See advtasksdyn.

The same three tiers exist on the input side inside run(): self.input().load() (the target), self.inputLoad() (the data — the default), and self.inputLoadConcat() (stack a task's dependencies into one frame).

Loading Output Data with Parameters

If you are using parameters this is how you load outputs. Make sure you run the task with that parameter first.

params = {'default_params':{}, 'use_params':{'preprocess':True}}
flow = oryxflow.WorkflowMulti(TaskSingleOutput, params)
data = flow.outputLoad() # load default task output
data['default_params']
data['use_params']

# multi output
data2 = flow.outputLoad(TaskMultipleOutput, as_dict=True) # load another task output
data2['default_params']['data1']
data2['default_params']['data2']
data2['use_params']['data1']
data2['use_params']['data2']

Putting it all together

See the full ML example.

To scaffold a real-life project layout, run /oryxflow:init-project in Claude Code.

Advanced: task attribute overrides

`persist`: data items to save, see above `external`: do check dependencies, good for sharing tasks without providing code `code_version`: bump (str or int) when this task's logic changes so it and everything downstream recompute; see Automatic code invalidation `keep_versions`: with code_version set, keep old versions at readable .../<Task>/v<version>/ paths `target_dir`: specify directory `target_ext`: specify extension `save_attrib`: include taskid in filename