connecting workflow logic to new api syntax

Author: tclose

benchmark.py

@@ -0,0 +1,38 @@
+import asyncio
+import time
+
+
+def sync_function(x):
+    return x * 2
+
+
+async def async_function(x):
+    return x * 2
+
+
+def benchmark_sync():
+    start_time = time.time()
+    for _ in range(1000000):
+        sync_function(10)
+    end_time = time.time()
+    return end_time - start_time
+
+
+async def benchmark_async():
+    start_time = time.time()
+    for _ in range(1000000):
+        await async_function(10)
+    end_time = time.time()
+    return end_time - start_time
+
+
+def main():
+    sync_time = benchmark_sync()
+    print(f"Sync function time: {sync_time:.6f} seconds")
+
+    async_time = asyncio.run(benchmark_async())
+    print(f"Async function time: {async_time:.6f} seconds")
+
+
+if __name__ == "__main__":
+    main()

example.py

@@ -0,0 +1,34 @@
+import asyncio
+
+
+def is_coroutine_function(func):
+    return asyncio.iscoroutinefunction(func)
+
+
+async def call_function(func, *args, **kwargs):
+    if is_coroutine_function(func):
+        return await func(*args, **kwargs)
+    else:
+        return func(*args, **kwargs)
+
+
+# Example usage
+async def async_function(x):
+    await asyncio.sleep(1)
+    return x * 2
+
+
+def sync_function(x):
+    return x * 2
+
+
+async def main():
+    result1 = await call_function(async_function, 10)
+    result2 = await call_function(sync_function, 10)
+    print(result1)  # Output: 20
+    print(result2)  # Output: 20
+
+
+# To run the example
+if __name__ == "__main__":
+    asyncio.run(main())

new-docs/source/index.rst

@@ -3,40 +3,41 @@
 Pydra
 =====
 
-Pydra is a lightweight Python dataflow engine for scientific analysis.
-Although designed as a successor to Nipype_, Pydra is supports analytics in any domain.
-Pydra helps build reproducible, scalable, reusable workflows that link processing tasks
-implemented in Python or shell commands to be executed on distributed compute platforms.
-
-The power of Pydra lies in ease of constructing workflows containing complex
-multiparameter map-reduce operations in Python code (see :ref:`Design philosophy` for
-the rationale behind its design).
+Pydra is a lightweight dataflow engine written in Python. Although designed to succeed
+Nipype_ in order to address the needs of the neuroimaging community, Pydra can be used
+for analytics in any scientific domain. Pydra facilitates the design of reproducible,
+scalable and robust workflows that can link diverse processing tasks implemented as
+shell commands or Python functions.
 
 **Key features:**
 
-* Combine diverse tasks (`Python functions <./tutorial/3-python.html>`__ or `shell commands <./tutorial/4-shell.html>`__) into coherent `workflows <./tutorial/5-workflow.html>`__
-* Concurrent execution on `choice of computing platform (e.g. workstation, SLURM, SGE, Dask, etc...) <./tutorial/2-advanced-execution.html#Workers>`__
+* Combine diverse tasks (`Python functions <./tutorial/3-python.html>`__ or `shell commands <./tutorial/4-shell.html>`__) into coherent, robust `workflows <./tutorial/5-workflow.html>`__
 * Dynamic workflow construction using Python code (see :ref:`Dynamic construction`)
+* Concurrent execution on `choice of computing platform (e.g. workstation, SLURM, SGE, Dask, etc...) <./tutorial/2-advanced-execution.html#Workers>`__
 * Map-reduce-like semantics (see :ref:`Splitting and combining`)
 * Global caching to reduce recomputation (see :ref:`Caches and hashes`)
 * Tasks can be executed in separate software environments, e.g. containers (see :ref:`Software environments`)
 * Strong type-checking, including file types, before execution (see :ref:`Typing and file-formats`)
 
+See :ref:`Design philosophy` for more details on the rationale behind Pydra's design.
+
 
 Installation
 ------------
 
-Pydra itself is a pure-Python package, which has only a handful of dependencies,
-therefore, it is straightforward to install via pip for Python >= 3.11
+Pydra is implemented purely in Python and has a small number of dependencies
+It is easy to install via pip for Python >= 3.11 (preferably within a
+`virtual environment`_):
 
 .. code-block:: bash
 
    $ pip install pydra
 
-Pre-designed tasks are available under the `pydra.tasks.*` package namespace. These tasks
-are implemented within separate packages that are typically specific to a given
-shell-command toolkit such as FSL_, AFNI_ or ANTs_, or a collection of related
-tasks/workflows (e.g. `niworkflows`_). Pip can be used to install these packages as well:
+Pre-designed tasks are available under the `pydra.tasks.*` namespace. These tasks
+are typically implemented within separate packages that are specific to a given
+shell-command toolkit, such as FSL_ (*pydra-fsl*), AFNI_ (*pydra-afni*) or
+ANTs_ (*pydra-ants*), or a collection of related tasks/workflows, such as Niworkflows
+(*pydra-niworkflows*). Pip can be used to install these extension packages as well:
 
 .. code-block:: bash
 
@@ -50,13 +51,15 @@ to run them (see :ref:`Software environments`).
 Tutorials and notebooks
 -----------------------
 
-The following tutorials provide a step-by-step guide to using Pydra.
-They can be read in any order, but it is recommended to start with :ref:`Getting started`.
-The tutorials are implemented as Jupyter notebooks, which can be downloaded and run locally
+The following tutorials provide a step-by-step guide to using Pydra. They can be
+studied in any order, but it is recommended to start with :ref:`Getting started` and
+step through the list from there.
+
+The tutorials are written in Jupyter notebooks, which can be downloaded and run locally
 or run online using the |Binder| button within each tutorial.
 
 If you decide to download the notebooks and run locally, be sure to install the necessary
-dependencies with
+dependencies (ideally within a  `virtual environment`_):
 
 .. code-block:: bash
 
@@ -74,7 +77,8 @@ Learn how to execute existing tasks (including workflows) on different systems
 Design
 ~~~~~~
 
-Learn how to design your own tasks
+Learn how to design your own tasks, wrapped shell commands or Python functions, or
+workflows,
 
 * :ref:`Python-tasks`
 * :ref:`Shell-tasks`
@@ -171,5 +175,6 @@ See the full reference documentation for Pydra
 .. _AFNI: https://afni.nimh.nih.gov/
 .. _niworkflows: https://niworkflows.readthedocs.io/en/latest/
 .. _Nipype: https://nipype.readthedocs.io/en/latest/
+.. _virtual environment: https://docs.python.org/3/library/venv.html
 .. |Binder| image:: https://mybinder.org/badge_logo.svg
    :target: https://mybinder.org/v2/gh/nipype/pydra/develop

pydra/design/base.py

@@ -27,7 +27,7 @@
 
 if ty.TYPE_CHECKING:
     from pydra.engine.specs import TaskDef, TaskOutputs
-    from pydra.engine.core import Task
+
 
 __all__ = [
     "Field",
@@ -352,7 +352,6 @@ def get_fields(klass, field_type, auto_attribs, helps) -> dict[str, Field]:
 def make_task_def(
     spec_type: type["TaskDef"],
     out_type: type["TaskOutputs"],
-    task_type: type["Task"],
     inputs: dict[str, Arg],
     outputs: dict[str, Out],
     klass: type | None = None,
@@ -418,15 +417,12 @@ def make_task_def(
             name=name,
             bases=bases,
             kwds={},
-            exec_body=lambda ns: ns.update(
-                {"Task": task_type, "Outputs": outputs_klass}
-            ),
+            exec_body=lambda ns: ns.update({"Outputs": outputs_klass}),
         )
     else:
         # Ensure that the class has it's own annotations dict so we can modify it without
         # messing up other classes
         klass.__annotations__ = copy(klass.__annotations__)
-        klass.Task = task_type
         klass.Outputs = outputs_klass
     # Now that we have saved the attributes in lists to be
     for arg in inputs.values():

pydra/design/boutiques.py

@@ -6,7 +6,6 @@
 from functools import reduce
 from fileformats.generic import File
 from pydra.engine.specs import ShellDef
-from pydra.engine.task import BoshTask
 from .base import make_task_def
 from . import shell
 
@@ -115,7 +114,6 @@ def define(
     )
     return make_task_def(
         spec_type=ShellDef,
-        task_type=BoshTask,
         out_type=out,
         arg_type=arg,
         inputs=inputs,

pydra/design/python.py

@@ -121,7 +121,6 @@ def define(
     PythonDef
         The task definition class for the Python function
     """
-    from pydra.engine.task import PythonTask
     from pydra.engine.specs import PythonDef, PythonOutputs
 
     def make(wrapped: ty.Callable | type) -> PythonDef:
@@ -162,7 +161,6 @@ def make(wrapped: ty.Callable | type) -> PythonDef:
         interface = make_task_def(
             PythonDef,
             PythonOutputs,
-            PythonTask,
             parsed_inputs,
             parsed_outputs,
             name=name,

pydra/design/shell.py

@@ -287,7 +287,6 @@ def define(
     ShellDef
         The interface for the shell command
     """
-    from pydra.engine.task import ShellTask
     from pydra.engine.specs import ShellDef, ShellOutputs
 
     def make(
@@ -376,7 +375,6 @@ def make(
         interface = make_task_def(
             ShellDef,
             ShellOutputs,
-            ShellTask,
             parsed_inputs,
             parsed_outputs,
             name=class_name,

pydra/design/workflow.py

@@ -14,7 +14,7 @@
 )
 
 if ty.TYPE_CHECKING:
-    from pydra.engine.workflow.base import Workflow
+    from pydra.engine.core import Workflow
     from pydra.engine.specs import TaskDef, TaskOutputs, WorkflowDef
 
 
@@ -128,7 +128,6 @@ def define(
     TaskDef
         The interface for the function or class.
     """
-    from pydra.engine.core import WorkflowTask
     from pydra.engine.specs import TaskDef, WorkflowDef, WorkflowOutputs
 
     if lazy is None:
@@ -174,7 +173,6 @@ def make(wrapped: ty.Callable | type) -> TaskDef:
         interface = make_task_def(
             WorkflowDef,
             WorkflowOutputs,
-            WorkflowTask,
             parsed_inputs,
             parsed_outputs,
             name=name,
@@ -208,12 +206,12 @@ def this() -> "Workflow":
 OutputsType = ty.TypeVar("OutputsType", bound="TaskOutputs")
 
 
-def add(task_spec: "TaskDef[OutputsType]", name: str = None) -> OutputsType:
+def add(task_def: "TaskDef[OutputsType]", name: str = None) -> OutputsType:
     """Add a node to the workflow currently being constructed
 
     Parameters
     ----------
-    task_spec : TaskDef
+    task_def : TaskDef
         The definition of the task to add to the workflow as a node
     name : str, optional
         The name of the node, by default it will be the name of the task definition
@@ -224,4 +222,4 @@ def add(task_spec: "TaskDef[OutputsType]", name: str = None) -> OutputsType:
     Outputs
         The outputs definition of the node
     """
-    return this().add(task_spec, name=name)
+    return this().add(task_def, name=name)