oryxflow package

Submodules

Module contents

class oryxflow.FlowExport(tasks=None, flows=None, save=False, path_export='tasks_export.py')

Bases: object

Auto generate task files to quickly share workflows with others.

Parameters:

• tasks (obj) – task or list of tasks to share

• flows (obj) – flow or list of flows to get tasks from.

• save (bool) – save to tasks file

• path_export (str) – filename for tasks to export.

generate()

Generate output files

class oryxflow.FlowImport(path=None, module=None, path_data=None)

Bases: object

Import a specific module from a directory.

Parameters:

• path (str) – path to the dir to import from

• module (str) – the module name to import

• path_data (str) – path to the data file; if not absolute will be appended to path

class oryxflow.Workflow(task=None, params=None, path=None, env=None)

Bases: object

The class is used to orchestrate tasks and define a task pipeline

accept_code(task=None)

Accept an output-equivalent code change for a task and its entire upstream dependency tree (see oryxflow.accept_code()). A bare flow.accept_code() accepts the whole flow: every imported task family that resolves with this flow’s parameters – so one call from a fresh process blesses a multi-final pipeline, not just the configured default task’s subtree.

Parameters:

task (class, list) – task class, or list of task classes, to accept from (default: every task the flow can compute)

Returns: list of task_ids re-stamped

attach_flow(flow=None, flow_name='flow')

complete(task=None, cascade=True)

get_task(task=None)

Get task with the workflow parameters

Parameters:

task (class)

Retuns: An instance of task class with the workflow parameters

output(task=None)

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

Load output from task with the workflow parameters

Parameters:

• task (class) – task class

• 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

outputLoadAll(task=None, keys=None, as_dict=False, cached=False)

Load all output from task with the workflow parameters

Parameters:

• task (class) – task class

• 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

outputLoadMeta(task=None)

outputLoadMetaJson(task=None)

outputPath(task=None)

Ouputs the Path given a task

Parameters:

task (class) – task class

Returns: list or dict of all task paths

preview(tasks=None, indent='', last=True, show_params=True, clip_params=False, print_it=True)

Preview task flows with the workflow parameters

Parameters:

tasks (class, list) – task class or list of tasks class

reset(task=None, confirm=False)

reset_downstream(task, task_downstream=None, confirm=False)

Invalidate all downstream tasks in a flow.

For example, you have 3 dependant tasks. Normally you run Task3 but you’ve changed parameters for Task1. By invalidating Task3 it will check the full DAG and realize Task1 needs to be invalidated and therefore Task2 and Task3 also.

Parameters:

• task (obj, class, list) – task/family — or list of families — to invalidate downstream of. Only the family is used (a class is fine — it is not instantiated), so this works for tasks whose params are internal to the DAG (e.g. a per-country task you can’t name from flow params); a list resets several families at once.

• task_downstream (obj) – terminal downstream task the walk stops at. Defaults to the flow’s default task; must be set (here or as the default task) so it knows where “down” ends.

• confirm (bool) – confirm operation

reset_upstream(task, confirm=False, only=None)

run(tasks=None, forced=None, forced_all=False, forced_all_upstream=False, confirm=False, workers=1, abort=True, execution_summary=None, **kwargs)

Run tasks with the workflow parameters. Runs the DAG sequentially in dependency order.

Parameters:

• tasks (class, list) – task class or list of tasks class

• forced (list) – list of forced tasks

• forced_all (bool) – force all tasks

• forced_all_upstream (bool) – force all tasks including upstream

• confirm (list) – confirm invalidating tasks

• workers (int) – number of workers

• abort (bool) – on errors raise exception

• execution_summary (bool) – log execution summary

• kwargs – keywords to pass to core.build

set_default(task)

Set default task for the workflow object

Parameters:

task (obj)

class oryxflow.WorkflowMulti(task=None, params=None, path=None, env=None)

Bases: object

A multi experiment workflow can be defined with multiple flows and separate parameters for each flow and a default task. It is mandatory to define the flows and parameters for each of the flows.

accept_code(task=None, flow=None)

Accept an output-equivalent code change for a task and its upstream tree (see oryxflow.accept_code()), for one flow or all flows. With no task, each flow accepts its default task plus everything run on it (see Workflow.accept_code()).

Parameters:

• task (class, list) – task class or list of task classes (defaults to the flow’s default task plus everything run on that flow)

• flow (string) – flow name; if not passed, accepts across all flows

Returns: list of task_ids re-stamped (dict of lists when run for all flows)

get_flow(flow)

Get flow by name

Parameters:

flow (string) – The name of the experiment for which the flow is to be run. If nothing is passed, all the flows are run

Retuns: An instance of Workflow

get_task(task=None, flow=None)

Get task with the workflow parameters for a flow

Parameters:

• flow (string) – The name of the experiment for which the flow is to be run. If nothing is passed, all the flows are run

• task (class) – task class

Retuns: An instance of task class with the workflow parameters

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

Load output from task with the workflow parameters for a flow

Parameters:

• flow (string) – The name of the experiment for which the flow is to be run. If nothing is passed, all the flows are run

• task (class) – task class

• 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

outputLoadAll(task=None, flow=None, keys=None, as_dict=False, cached=False)

Load all output from task with the workflow parameters for a flow

Parameters:

• flow (string) – The name of the experiment for which the flow is to be run. If nothing is passed, all the flows are run

• task (class) – task class

• 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

outputLoadConcat(task=None, keys=None, as_dict=False, cached=False, concat_fn=None, tagkeys=None)

Load task output for every flow and concatenate into one DataFrame, tagging each flow’s rows with that flow’s raw params.

outputLoadMeta(task=None, flow=None)

outputLoadMetaJson(task=None, flow=None)

outputPath(task=None, flow=None)

Ouputs the Path given a task

Parameters:

• task (class) – task class

• flow (string) – The name of the experiment for which the flow is to be run. If nothing is passed, all the flows are run

Returns: list or dict of all task paths

preview(tasks=None, flow=None, indent='', last=True, show_params=True, clip_params=False, print_it=True)

Preview task flows with the workflow parameters for a flow

Parameters:

• flow (string) – The name of the experiment for which the flow is to be run. If nothing is passed, all the flows are run

• tasks (class, list) – task class or list of tasks class

reset(task=None, flow=None, confirm=False)

reset_downstream(task=None, task_downstream=None, flow=None, confirm=False)

reset_upstream(task=None, flow=None, confirm=False, only=None)

run(tasks=None, flow=None, forced=None, forced_all=False, forced_all_upstream=False, confirm=False, workers=1, abort=True, execution_summary=None, **kwargs)

Run tasks with the workflow parameters for a flow. Runs the DAG sequentially in dependency order.

Parameters:

• flow (string) – The name of the experiment for which the flow is to be run. If nothing is passed, all the flows are run

• tasks (class, list) – task class or list of tasks class

• forced (list) – list of forced tasks

• forced_all (bool) – force all tasks

• forced_all_upstream (bool) – force all tasks including upstream

• confirm (list) – confirm invalidating tasks

• workers (int) – number of workers

• abort (bool) – on errors raise exception

• execution_summary (bool) – log execution summary

• kwargs – keywords to pass to core.build

set_default(task)

Set default task for the workflow. The default task is set for all the experiments

Parameters:

task (obj)

oryxflow.clone_parent(cls)

class oryxflow.dict_inherits(*tasks_to_inherit)

Bases: object

class oryxflow.dict_requires(*tasks_to_require)

Bases: object

class oryxflow.dotdict

Bases: dict

dot.notation access to dictionary attributes

oryxflow.enable_cloud_storage(protocol, bucket, prefix=None)

Initialize cloud storage Uses https://github.com/orgs/fsspec/repositories

Parameters:

• protocol (str) – fsspec eg gcs, s3, dropbox etc. See https://pypi.org/project/universal-pathlib/

• bucket (str) – bucket name

• prefix (str) – prefix similar to folder

oryxflow.enable_gcs(bucket, prefix=None)

Initialize google cloud storage, for reference see https://cloud.google.com/storage/docs/listing-objects Uses https://gcsfs.readthedocs.io/en/latest/

Parameters:

• bucket (str) – bucket name

• prefix (str) – prefix similar to folder

oryxflow.inherits(*tasks_to_inherit)

oryxflow.invalidate_all(confirm=False)

Invalidate all tasks by deleting all files in data directory

Parameters:

confirm (bool) – confirm operation

oryxflow.invalidate_downstream(task, task_downstream, confirm=False)

Invalidate all downstream tasks in a flow.

For example, you have 3 dependant tasks. Normally you run Task3 but you’ve changed parameters for Task1. By invalidating Task3 it will check the full DAG and realize Task1 needs to be invalidated and therefore Task2 and Task3 also.

Parameters:

• task (obj, class, list) – task/family — or list of families — to invalidate downstream of. Only the family is used (a class is fine), so a list resets several families and everything downstream of each, in one call.

• task_downstream (obj) – downstream task target

• confirm (bool) – confirm operation

oryxflow.invalidate_orphans(confirm=False)

Invalidate all unused task outputs

Parameters:

confirm (bool) – confirm operation

oryxflow.invalidate_upstream(task, confirm=False, only=None)

Invalidate all tasks upstream tasks in a flow.

For example, you have 3 dependant tasks. Normally you run Task3 but you’ve changed parameters for Task1. By invalidating Task3 it will check the full DAG and realize Task1 needs to be invalidated and therefore Task2 and Task3 also.

Parameters:

• task (obj) – task to invalidate. This should be an upstream task for which you want to check upstream dependencies for invalidation conditions

• confirm (bool) – confirm operation

• only (class, list) – if set, only invalidate upstream tasks of these task family/families

oryxflow.preview(tasks, indent='', last=True, show_params=True, clip_params=False, print_it=True)

Preview task flows

Parameters:

tasks (obj, list) – task or list of tasks

oryxflow.requires(*tasks_to_require)

oryxflow.run(tasks, forced=None, forced_all=False, forced_all_upstream=False, confirm=False, workers=1, abort=True, execution_summary=None, main_thread_only=False, **kwargs)

Run tasks locally. Runs the DAG sequentially in dependency order.

Parameters:

• tasks (obj, list) – task or list of tasks

• forced (list) – list of forced tasks

• forced_all (bool) – force all tasks

• forced_all_upstream (bool) – force all tasks including upstream

• confirm (list) – confirm invalidating tasks

• workers (int) – number of workers

• abort (bool) – on errors raise exception

• execution_summary (bool) – log execution summary

• main_thread_only (bool) – if True, only works in main thread of the main interpreter. Default false so it can run in apps and workers.

• kwargs – keywords to pass to core.build

oryxflow.runIt(task, params=None, reset=False)

oryxflow.runIterConcat(task, params, load=True, taskLoad=None, reset=False, concat_fn=None, tagkeys=None)

Run task across a grid of params (one flow per param set) and return the per-flow outputs concatenated into one DataFrame, each flow tagged with its params.

oryxflow.runLoad(task, params=None, load=True, taskLoad=None, reset=False)

oryxflow.set_dir(dir=None)

Initialize oryxflow

Parameters:

dir (str) – data output directory

oryxflow.show(task)

Show task execution status

Parameters:

tasks (obj, list) – task or list of tasks

oryxflow.taskflow_downstream(task, task_downstream, only_complete=False)

Get all downstream outputs for a task

Parameters:

• task (obj) – task

• task_downstream (obj) – downstream target task

oryxflow.taskflow_upstream(task, only_complete=False)

Get all upstream inputs for a task

Parameters:

task (obj) – task

oryxflow.tasks module

class oryxflow.tasks.TaskAggregator(*args, **kwargs)

Bases: Task

Task which yields other tasks

NB: Use this function by implementing run() which should do nothing but yield other tasks

example:

class TaskCollector(oryxflow.tasks.TaskAggregator):
    def run(self):
        yield Task1()
        yield Task2()

complete(cascade=True)

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.

deps()

Flattened list of required tasks.

invalidate(confirm=False)

output()

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

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

reset(confirm=False)

class oryxflow.tasks.TaskCSVGZPandas(*args, **kwargs)

Bases: TaskData

Task which saves to CSV

target_class

alias of CSVGZPandasTarget

target_ext = 'csv.gz'

class oryxflow.tasks.TaskCSVPandas(*args, **kwargs)

Bases: TaskData

Task which saves to CSV

target_class

alias of CSVPandasTarget

target_ext = 'csv'

class oryxflow.tasks.TaskCache(*args, **kwargs)

Bases: TaskData

Task which saves to cache

target_class

alias of CacheTarget

target_ext = 'cache'

class oryxflow.tasks.TaskCachePandas(*args, **kwargs)

Bases: TaskData

Task which saves to cache pandas dataframes

target_class

alias of PdCacheTarget

target_ext = 'cache'

class oryxflow.tasks.TaskData(*args, **kwargs)

Bases: Task

Task which has data as input and output

Parameters:

• 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

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.

classmethod get_param_values(params, args, kwargs)

Resolve parameter values from positional args, kwargs and defaults.

Parameters:

• params – list of (param_name, Parameter)

• args – positional arguments

• kwargs – keyword arguments

Returns:

list of (name, value) tuples, one per parameter

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

Load all or several outputs from task

Parameters:

• keys (list) – list of data to load

• task (str) – if requires multiple tasks load that task ‘input1’ for eg def requires: {‘input1’:Task1(), ‘input2’:Task2()}

• cached (bool) – cache data in memory

• as_dict (bool) – if the inputs were saved as a dictionary. use this to return them as dictionary.

Returns: list or dict of all task output

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

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.

inputLoadMetaJson(key=None)

invalidate(confirm=False)

Reset a task, eg by deleting output file

keep_versions = False

metaLoad(key=None)

metaSave(data)

metadata = None

output()

Output target(s) this task produces

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

Load all or several outputs from task

Parameters:

• 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

outputLoadAllMeta()

outputLoadMeta()

outputLoadMetaJson()

persist = ['data']

reset(confirm=False)

Reset a task, eg by deleting output file

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

Persist data to target

Parameters:

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

saveMeta(data)

saveMetaJson(data)

target_class

alias of DataTarget

target_ext = 'ext'

class oryxflow.tasks.TaskExcelPandas(*args, **kwargs)

Bases: TaskData

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

invalidate(confirm=False)

Reset a task, eg by deleting output file

output()

Output target(s) this task produces

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

Load all or several outputs from task

Parameters:

• 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

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

Persist data to target

Parameters:

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

target_class

alias of ExcelPandasSheetsTarget

target_ext = 'xlsx'

class oryxflow.tasks.TaskExcelPandasSingle(*args, **kwargs)

Bases: TaskData

Task which saves each persist key as a separate Excel file

target_class

alias of ExcelPandasTarget

target_ext = 'xlsx'

class oryxflow.tasks.TaskJson(*args, **kwargs)

Bases: TaskData

Task which saves to json

target_class

alias of JsonTarget

target_ext = 'json'

class oryxflow.tasks.TaskMarkdown(*args, **kwargs)

Bases: TaskData

Task which saves to markdown and HTML

target_class

alias of MarkdownTarget

target_ext = 'md'

class oryxflow.tasks.TaskPickle(*args, **kwargs)

Bases: TaskData

Task which saves to pickle

target_class

alias of PickleTarget

target_ext = 'pkl'

class oryxflow.tasks.TaskPqPandas(*args, **kwargs)

Bases: TaskData

Task which saves to parquet

target_class

alias of PqPandasTarget

target_ext = 'parquet'

oryxflow.targets module

class oryxflow.targets.CSVGZPandasTarget(path=None)

Bases: CSVPandasTarget

Saves to CSV gzip, loads to pandas dataframe

save(df, **kwargs)

Save dataframe to csv gzip

Parameters:

• df (obj) – pandas dataframe

• kwargs – additional arguments to pass to df.to_csv

Returns: filename

class oryxflow.targets.CSVPandasTarget(path=None)

Bases: DataTarget

Saves to CSV, loads to pandas dataframe

load(cached=False, **kwargs)

Load from csv to pandas dataframe

Parameters:

• cached (bool) – keep data cached in memory

• **kwargs – arguments to pass to pd.read_csv

Returns: pandas dataframe

save(df, **kwargs)

Save dataframe to csv

Parameters:

• df (obj) – pandas dataframe

• kwargs – additional arguments to pass to df.to_csv

Returns: filename

class oryxflow.targets.CacheTarget(path=None)

Bases: LocalTarget

Saves to in-memory cache, loads to python object

exists()

invalidate()

load(cached=True)

Load from in-memory cache

Returns: python object

save(df)

Save dataframe to in-memory cache

Parameters:

df (obj) – pandas dataframe

Returns: filename

class oryxflow.targets.DataTarget(path=None)

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.

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

Runs a function to load data from storage into memory

Parameters:

• fun (function) – loading function

• cached (bool) – keep data cached in memory

• **kwargs – arguments to pass to fun

Returns: data object

save(df, fun, **kwargs)

Runs a function to save data from memory into storage

Parameters:

• df (obj) – data to save

• fun (function) – saving function

• **kwargs – arguments to pass to fun

Returns: filename

class oryxflow.targets.ExcelPandasSheetsTarget(path=None)

Bases: _LocalPathTarget

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

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

Load sheets from Excel file

Parameters:

• 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

save(data, **kwargs)

Save dict of dataframes as sheets in a single Excel file

Parameters:

• data (dict) – {sheet_name: dataframe}

• kwargs – additional arguments to pass to df.to_excel

Returns: filename

class oryxflow.targets.ExcelPandasTarget(path=None)

Bases: DataTarget

Saves to Excel, loads to pandas dataframe

load(cached=False, **kwargs)

Load from Excel to pandas dataframe

Parameters:

• cached (bool) – keep data cached in memory

• **kwargs – arguments to pass to pd.read_csv

Returns: pandas dataframe

save(df, **kwargs)

Save dataframe to Excel

Parameters:

• df (obj) – pandas dataframe

• kwargs – additional arguments to pass to df.to_csv

Returns: filename

class oryxflow.targets.JsonTarget(path=None)

Bases: DataTarget

Saves to json, loads to dict

load(cached=False, **kwargs)

Load from json to dict

Parameters:

• cached (bool) – keep data cached in memory

• **kwargs – arguments to pass to json.load

Returns: dict

save(dict_, **kwargs)

Save dict to json

Parameters:

• dict (dict) – python dict

• kwargs – additional arguments to pass to json.dump

Returns: filename

class oryxflow.targets.MarkdownTarget(path=None)

Bases: DataTarget

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

invalidate()

load(cached=False, **kwargs)

Load from markdown file to string

Parameters:

• cached (bool) – keep data cached in memory

• **kwargs – arguments to pass to read function

Returns: markdown string

save(md_string, **kwargs)

Save markdown string to .md and .html files

Parameters:

• md_string (str) – markdown string

• kwargs – additional arguments to pass to markdown.markdown

Returns: filename

class oryxflow.targets.PdCacheTarget(path=None)

Bases: CacheTarget

class oryxflow.targets.PickleTarget(path=None)

Bases: DataTarget

Saves to pickle, loads to python obj

load(cached=False, **kwargs)

Load from pickle to obj

Parameters:

• cached (bool) – keep data cached in memory

• **kwargs – arguments to pass to pickle.load

Returns: dict

save(obj, **kwargs)

Save obj to pickle

Parameters:

• obj (obj) – python object

• kwargs – additional arguments to pass to pickle.dump

Returns: filename

class oryxflow.targets.PqPandasTarget(path=None)

Bases: DataTarget

Saves to parquet, loads to pandas dataframe

load(cached=False, **kwargs)

Load from parquet to pandas dataframe

Parameters:

• cached (bool) – keep data cached in memory

• **kwargs – arguments to pass to pd.read_parquet

Returns: pandas dataframe

save(df, **kwargs)

Save dataframe to parquet

Parameters:

• df (obj) – pandas dataframe

• kwargs – additional arguments to pass to df.to_parquet

Returns: filename

oryxflow.functional module

class oryxflow.functional.Workflow

Bases: object

Functional Flow class that acts as a manager of all flow steps. Defines all the decorators that can be used on flow functions.

add_global_params(**params)

Adds params to flow functions. More like declares the params for further use. :param params: dictionary of param name and param type :type params: dict

Example

flow.add_params({‘multiplier’: oryxflow.IntParameter(default=0)})

delete(func_to_reset, *args, **kwargs)

Possibly dangerous! delete(func) will delete all files in the data/func directory of the given func. Useful if you want to delete all function related outputs. Consider using reset(func, params) to reset a specific func

deleteAll(*args, **kwargs)

Possibly dangerous! Will delete all files in the data/ directory of the functions attached to the workflow object. Useful if you want to delete all outputs even the once previously run. Consider using resetAll() if you want to only reset the functions with params you have run thus far

outputLoad(func_to_run, *args, **kwargs)

Loads all or several outputs from flow step.

Parameters:

• func_to_run – flow step function

• 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

outputLoadAll(func_to_run, *args, **kwargs)

Loads all output from flow task and its parents.

Parameters:

• func_to_run – flow step function

• 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

params(**params)

Flow parameters decorator. Use this to add parameter(s) to a particular flow task. Also see flow.add_global_params() to add parameters globally :param **params: :type **params: keyword arguments of oryxflow parameters

Example

@flow.params(multiplier=oryxflow.IntParameter(default=0))

persists(to_persist: list)

Flow persists decorator. Takes in a list of variables that need to be persisted for the flow step. :param to_persist: :type to_persist: list

Example

@flow.persists([‘a1’, ‘a2’])

preview(func_to_preview, params: dict)

requires(*args, **kwargs)

Flow requires decorator. Defines dependencies between flow steps. Internally calls oryxflow.requires.

Parameters:

func (dict or function or mutiple functions)

Examples

@flow.requires({“foo”: func1, “bar”: func2}) @flow.requires(func1)

reset(func_to_reset, params=None, *args, **kwargs)

Resets a particular function. Use with params to reset function with the given parameters. If params is not used, reset(func) will reset the function with all the parameters run thus far

resetAll(*args, **kwargs)

Resets all functions that are attached to the workflow object that have run at least once.

run(funcs_to_run, params: dict = None, multi_params: dict = None, *args, **kwargs)

Runs flow steps locally. :param funcs_to_run: :type funcs_to_run: function or list of functions :param params: dictionary of paramaters. Keys are param names and values are the values of params. :type params: dict

Examples

flow.run(func, params={‘multiplier’:2})

flow.run([func1, func2], params={‘multiplier’:42})

flow.run(func)

task(task_type: TaskData)

Flow step decorator. Converts the decorated function into a flow step. Should be defined at the top of the decorator stack.

Parameters:

task_type (d6ftflow.tasks.TaskData) – task_type should be a class which inherits from d6ftflow.tasks.TaskData

Example

@flow.step(oryxflow.tasks.TaskCache)