Add context manager to load workflow resources

.github/workflows/test-suite.yaml

@@ -94,7 +94,7 @@ jobs:
           GLOBUS_COMPUTE_CLIENT_ID: ${{ secrets.GC_CLIENT_ID }}
           GLOBUS_COMPUTE_CLIENT_SECRET: ${{ secrets.GC_CLIENT_SECRET }}
         run: |
-          docker exec -e GLOBUS_COMPUTE_CLIENT_ID="$GLOBUS_COMPUTE_CLIENT_ID" -e GLOBUS_COMPUTE_CLIENT_SECRET="$GLOBUS_COMPUTE_CLIENT_SECRET" frontend bash -l -c "cd work; source .venv-py${{ matrix.python-version }}/bin/activate; pytest -s -v --assert=plain --cov=src/chiltepin --cov-report=term-missing --cov-report=markdown:coverage.md --cov-report=xml --cov-report=html --cov-fail-under=0 --config=tests/configs/docker.yaml"
+          docker exec -e GLOBUS_COMPUTE_CLIENT_ID="$GLOBUS_COMPUTE_CLIENT_ID" -e GLOBUS_COMPUTE_CLIENT_SECRET="$GLOBUS_COMPUTE_CLIENT_SECRET" frontend bash -l -c "cd work; source .venv-py${{ matrix.python-version }}/bin/activate; pytest -s -v --cov=src/chiltepin --cov-report=term-missing --cov-report=markdown:coverage.md --cov-report=xml --cov-report=html --cov-fail-under=0 --config=tests/configs/docker.yaml"
       -
         name: Copy coverage reports from container
         if: always()
@@ -233,7 +233,7 @@ jobs:
           GLOBUS_COMPUTE_CLIENT_ID: ${{ secrets.GC_CLIENT_ID }}
           GLOBUS_COMPUTE_CLIENT_SECRET: ${{ secrets.GC_CLIENT_SECRET }}
         run: |
-          docker exec -e GLOBUS_COMPUTE_CLIENT_ID="$GLOBUS_COMPUTE_CLIENT_ID" -e GLOBUS_COMPUTE_CLIENT_SECRET="$GLOBUS_COMPUTE_CLIENT_SECRET" frontend bash -l -c "cd work; source .venv-py${{ env.LATEST_PYTHON_VERSION }}/bin/activate; pytest -s -v --assert=plain --config=tests/configs/docker.yaml"
+          docker exec -e GLOBUS_COMPUTE_CLIENT_ID="$GLOBUS_COMPUTE_CLIENT_ID" -e GLOBUS_COMPUTE_CLIENT_SECRET="$GLOBUS_COMPUTE_CLIENT_SECRET" frontend bash -l -c "cd work; source .venv-py${{ env.LATEST_PYTHON_VERSION }}/bin/activate; pytest -s -v --config=tests/configs/docker.yaml"
       -
         name: Debug session
         if: ${{ failure() }}

docs/api.rst

@@ -11,6 +11,13 @@ Configuration Module
    :members:
    :show-inheritance:
 
+Workflow Module
+---------------
+
+.. automodule:: chiltepin.workflow
+   :members:
+   :show-inheritance:
+
 Endpoint Management Module
 ---------------------------
 

docs/configuration.rst

@@ -17,6 +17,45 @@ Resources can be:
 The configuration file defines named resources, where each resource represents a pool
 of nodes and/or cores on which tasks can be executed.
 
+Default "local" Resource
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+Chiltepin automatically provides a default resource named **"local"** that is always
+available, even if you don't define any resources in your configuration file. This
+default resource:
+
+- Uses the current machine (localhost provider)
+- Has 1 core per worker and 1 maximum worker
+- Starts with 0 blocks and can scale up to 1 block
+- Is useful for testing and light computational tasks
+
+You can use this default resource without any configuration:
+
+.. code-block:: python
+
+   from chiltepin import run_workflow
+   from chiltepin.tasks import python_task
+
+   @python_task
+   def my_task():
+       return "Hello from local!"
+
+   # Works even with an empty config
+   with run_workflow({}):
+       result = my_task(executor=["local"]).result()
+
+You can override the default "local" resource by defining your own resource with
+that name in your configuration file. This is useful if you need different settings
+for local execution:
+
+.. code-block:: yaml
+
+   local:
+     provider: "localhost"
+     max_blocks: 2
+     max_workers_per_node: 4
+     cores_per_worker: 2
+
 Basic Structure
 ---------------
 
@@ -415,26 +454,70 @@ Loading Configurations
 Parse and Load
 ^^^^^^^^^^^^^^
 
+**Loading from a file:**
+
 .. code-block:: python
 
-   import chiltepin.configure
-   import parsl
+   from chiltepin import run_workflow
 
-   # Parse YAML configuration file
-   config_dict = chiltepin.configure.parse_file("my_config.yaml")
-
-   # Create Parsl configuration
-   parsl_config = chiltepin.configure.load(
-       config_dict,
+   # Load configuration from YAML file and run workflow
+   with run_workflow(
+       "my_config.yaml",
        include=["compute", "mpi"],  # Only load specific resources
        run_dir="./runinfo"           # Directory for Parsl runtime files
-   )
+   ):
+       # Run your tasks here
+       result = my_task(executor=["compute"]).result()
 
-   # Initialize Parsl with configuration
-   parsl.load(parsl_config)
+**Loading from a dict:**
+
+.. code-block:: python
+
+   from chiltepin import run_workflow
+
+   # Define configuration as a dictionary
+   config_dict = {
+       "compute": {
+           "provider": "slurm",
+           "account": "my-account",
+           "partition": "compute",
+           "nodes_per_block": 1,
+           "cores_per_node": 40,
+           "walltime": "01:00:00",
+       },
+       "mpi": {
+           "provider": "slurm",
+           "account": "my-account",
+           "partition": "compute",
+           "nodes_per_block": 2,
+           "launcher": "mpi",
+           "walltime": "02:00:00",
+       }
+   }
+
+   # Load configuration from dict and run workflow
+   with run_workflow(
+       config_dict,
+       include=["compute", "mpi"],  # Only load specific resources
+       run_dir="./runinfo"
+   ):
+       # Run your tasks here
+       result = my_task(executor=["compute"]).result()
 
 The ``include`` parameter lets you selectively load only specific resources from your
-configuration file. If omitted, all resources are loaded.
+configuration. If omitted, all resources are loaded.
+
+.. note::
+   The default **"local"** resource is always available, regardless of the ``include``
+   parameter. You do not need to add "local" to the include list to use it. This ensures
+   you always have a fallback resource available for tasks.
+
+   .. code-block:: python
+
+      # "local" is available even though not in include list
+      with run_workflow("config.yaml", include=["compute"]):
+          local_result = my_task(executor=["local"]).result()  # Works!
+          compute_result = my_task(executor=["compute"]).result()  # Works!
 
 Configuration Best Practices
 -----------------------------

docs/container.rst

@@ -66,13 +66,13 @@ After installation, you can run the test suite using the Docker-specific configu
 
 .. code-block:: console
 
-   (container) $ pytest --assert=plain --config=tests/configs/docker.yaml
+   (container) $ pytest --config=tests/configs/docker.yaml
 
 For more verbose output:
 
 .. code-block:: console
 
-   (container) $ pytest -s -vvv --assert=plain --config=tests/configs/docker.yaml
+   (container) $ pytest -s -vvv --config=tests/configs/docker.yaml
 
 Adjusting Core Count
 --------------------

docs/data.rst

@@ -45,7 +45,7 @@ Basic Usage
        dst_ep="hpc-scratch",
        src_path="/Users/me/data/input.dat",
        dst_path="/scratch/project/input.dat",
-       executor="local"
+       executor=["local"]
    )
 
    # Wait for transfer to complete
@@ -115,7 +115,7 @@ Transfer entire directories recursively:
        src_path="/Users/me/project/data/",
        dst_path="/scratch/project/data/",
        recursive=True,
-       executor="local"
+       executor=["local"]
    )
 
 Endpoint Names vs UUIDs
@@ -159,7 +159,7 @@ Basic Usage
    delete_future = delete_task(
        src_ep="hpc-scratch",
        src_path="/scratch/project/temp.dat",
-       executor="local"
+       executor=["local"]
    )
 
    # Wait for deletion to complete
@@ -219,7 +219,7 @@ Delete entire directories:
        src_ep="hpc-scratch",
        src_path="/scratch/project/temp_data/",
        recursive=True,
-       executor="local"
+       executor=["local"]
    )
 
 .. warning::
@@ -238,8 +238,7 @@ A common pattern is to stage data, process it, then clean up:
 
 .. code-block:: python
 
-   import parsl
-   import chiltepin.configure
+   from chiltepin import run_workflow
    from chiltepin.tasks import python_task
    from chiltepin.data import transfer_task, delete_task
 
@@ -251,32 +250,29 @@ A common pattern is to stage data, process it, then clean up:
        result = df.mean().to_dict()
        return result
 
-   # Load configuration and start Parsl
-   config_dict = chiltepin.configure.parse_file("config.yaml")
-   parsl_config = chiltepin.configure.load(config_dict)
-
-   with parsl.load(parsl_config):
+   # Load configuration and start workflow
+   with run_workflow("config.yaml"):
        # Stage data to compute resource
        stage_in = transfer_task(
            src_ep="my-laptop",
            dst_ep="hpc-scratch",
            src_path="/Users/me/data/dataset.csv",
            dst_path="/scratch/project/dataset.csv",
-           executor="local"
+           executor=["local"]
        )
 
        # Process the staged data (waits for transfer via inputs)
        analysis = analyze_data(
            "/scratch/project/dataset.csv",
-           executor="compute",
+           executor=["compute"],
            inputs=[stage_in]  # Non-blocking dependency
        )
 
        # Clean up staged data (waits for processing via inputs)
        cleanup = delete_task(
            src_ep="hpc-scratch",
            src_path="/scratch/project/dataset.csv",
-           executor="local",
+           executor=["local"],
            inputs=[analysis]  # Non-blocking dependency
        )
 
@@ -306,7 +302,7 @@ Transfer multiple files in parallel:
            dst_ep="hpc-scratch",
            src_path=f"/Users/me/data/{filename}",
            dst_path=f"/scratch/project/{filename}",
-           executor="local"
+           executor=["local"]
        )
        transfers.append(future)
 
@@ -340,8 +336,8 @@ To run a transfer or deletion after multiple tasks complete, pass them via the
        return True
 
    # Generate files in parallel
-   config_ready = generate_config(executor="compute")
-   input_ready = generate_input(executor="compute")
+   config_ready = generate_config(executor=["compute"])
+   input_ready = generate_input(executor=["compute"])
 
    # Wait for both files before transferring (non-blocking dependency)
    transfer = transfer_task(
@@ -350,7 +346,7 @@ To run a transfer or deletion after multiple tasks complete, pass them via the
        src_path="/scratch/",
        dst_path="/results/",
        recursive=True,
-       executor="local",
+       executor=["local"],
        inputs=[config_ready, input_ready]  # Multiple dependencies
    )
 
@@ -386,21 +382,21 @@ dependencies:
        dst_ep="hpc-scratch",
        src_path="/data/raw.csv",
        dst_path="/scratch/raw.csv",
-       executor="local"
+       executor=["local"]
    )
 
    # Preprocess (waits for transfer via inputs)
    preprocess_future = preprocess(
        "/scratch/raw.csv",
        "/scratch/clean.csv",
-       executor="compute",
+       executor=["compute"],
        inputs=[stage_raw]  # Wait for transfer non-blocking
    )
 
    # Analyze (waits for preprocess completing and passing output path)
    analysis_future = analyze(
        preprocess_future,  # Parsl waits for this future and passes the output path
-       executor="compute"
+       executor=["compute"]
    )
 
    # Stage results back (waits for analysis via inputs)
@@ -409,7 +405,7 @@ dependencies:
        dst_ep="my-laptop",
        src_path="/scratch/clean.csv",
        dst_path="/data/cleaned_output.csv",
-       executor="local",
+       executor=["local"],
        inputs=[analysis_future]  # Wait for analysis non-blocking
    )
 
@@ -418,7 +414,7 @@ dependencies:
        src_ep="hpc-scratch",
        src_path="/scratch/",
        recursive=True,
-       executor="local",
+       executor=["local"],
        inputs=[stage_out]  # Wait for stage out non-blocking
    )