update readme

Author: Fire-Hound

README.md

@@ -120,11 +120,68 @@ print(scores)
 
 ```
 
+
 ## Example Library
 
 * [Minimal example](https://github.com/d6t/d6tflow/blob/master/docs/example-minimal.py)
 * [Rapid Prototyping for Quantitative Investing with d6tflow](https://github.com/d6tdev/d6tflow-binder-interactive/blob/master/example-trading.ipynb) 
 * Chain together functions into a workflow and get the power of d6tflow with only little change in code. **[Jupyter notebook example](https://github.com/d6t/d6tflow/blob/master/docs/example-functional.ipynb)**
+Alternatively, chain together functions into a workflow and get the power of d6tflow with only little change in code. **[Jupyter notebook example](https://github.com/d6t/d6tflow/blob/master/docs/example-functional.ipynb)**
+
+## Example: Functional Workflow
+``` python
+import pandas as pd
+
+import d6tflow
+from d6tflow.functional import Workflow
+
+flow = Workflow()
+
+
+@flow.task(d6tflow.tasks.TaskPqPandas)
+def Task1(task):
+    df = pd.DataFrame({'a': range(3)})
+    task.save(df)
+
+
+@flow.task(d6tflow.tasks.TaskPqPandas)
+def Task2(task):
+    df = pd.DataFrame({'b': range(3)})
+    task.save(df)
+
+
+@flow.task(d6tflow.tasks.TaskPqPandas)
+@flow.params(multiplier=d6tflow.IntParameter(default=2))
+@flow.requires({'input1': Task1, 'input2': Task2})
+def Task3(task):
+    df1 = task.input()['input1'].load()  # quickly load input data
+    df2 = task.input()['input2'].load()  # quickly load input data
+    df = df1.join(df2, lsuffix='1', rsuffix='2')
+    df['c'] = df['a'] * task.multiplier  # use task parameter
+    task.save(df)
+
+
+flow.run(Task3)
+flow.outputLoad(Task3)
+'''
+   a  b  c
+0  0  0  0
+1  1  1  2
+2  2  2  4
+'''
+
+# You can rerun the flow with different parameters
+flow.run(Task3, params={'multiplier': 3})
+flow.outputLoad(Task3)
+
+'''
+   a  b  c
+0  0  0  0
+1  1  1  3
+2  2  2  6
+'''
+```
+
 
 ## Documentation
 

d6tflow/functional.py

@@ -2,6 +2,7 @@
 import d6tflow
 from functools import wraps
 import pathlib
+import d6tcollect
 
 
 class Workflow:
@@ -39,6 +40,7 @@ def wrapper(*args, **kwargs):
             return wrapper
         self.common_decorator = common_decorator
 
+    @d6tcollect._collectClass
     def task(self, task_type: d6tflow.tasks.TaskData):
         """
             Flow step decorator.
@@ -62,6 +64,7 @@ def task(self, task_type: d6tflow.tasks.TaskData):
         self.object_count += 1
         return self.common_decorator
 
+    @d6tcollect._collectClass
     def requires(self, *args, **kwargs):
         """
             Flow requires decorator.
@@ -94,6 +97,7 @@ def requires(self, *args, **kwargs):
 
         return self.common_decorator
 
+    @d6tcollect._collectClass
     def params(self, **params):
         """
             Flow parameters decorator.
@@ -111,6 +115,7 @@ def params(self, **params):
             setattr(self.steps[self.current_step], param, params[param])
         return self.common_decorator
 
+    @d6tcollect._collectClass
     def persists(self, to_persist: list):
         """
             Flow persists decorator.
@@ -138,6 +143,7 @@ def preview(self, func_to_preview, params: dict):
         else:
             d6tflow.preview(self.steps[name]())
 
+    @d6tcollect._collectClass
     def run(self, funcs_to_run, params: dict = None, multi_params: dict = None, *args, **kwargs):
         """
             Runs flow steps locally. See luigi.build for additional details
@@ -294,7 +300,7 @@ def reset(self, func_to_reset, params=None, *args, **kwargs):
                 ]
 
     def resetAll(self, *args, **kwargs):
-        """Resets all functions that are attached to the workflow object that have run atleast once."""
+        """Resets all functions that are attached to the workflow object that have run at least once."""
         for name in self.steps:
             self.reset(name, params=None, *args, **kwargs)