Repository: Quantum-Accelerators/quacc
Pull request #1432 (2023-12-28 19:04:14)

Allow for complete flexibility on decorators (and thereby executors)
In preparation for #1404.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Quantum-Accelerators/quacc/pull/1432?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Quantum-Accelerators) Report
Attention: `10 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`6e07ffc`)](https://app.codecov.io/gh/Quantum-Accelerators/quacc/commit/6e07ffca7a830311cbc9d6ea40e648381502bb71?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Quantum-Accelerators) 98.98% compared to head [(`38d9310`)](https://app.codecov.io/gh/Quantum-Accelerators/quacc/pull/1432?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Quantum-Accelerators) 98.66%.

| [Files](https://app.codecov.io/gh/Quantum-Accelerators/quacc/pull/1432?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Quantum-Accelerators) | Patch % | Lines |
|---|---|---|
| [src/quacc/wflow\_tools/customizers.py](https://app.codecov.io/gh/Quantum-Accelerators/quacc/pull/1432?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Quantum-Accelerators#diff-c3JjL3F1YWNjL3dmbG93X3Rvb2xzL2N1c3RvbWl6ZXJzLnB5) | 75.60% | [10 Missing :warning: ](https://app.codecov.io/gh/Quantum-Accelerators/quacc/pull/1432?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Quantum-Accelerators) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main    #1432      +/-   ##
==========================================
- Coverage   98.98%   98.66%   -0.32%     
==========================================
  Files          76       77       +1     
  Lines        2958     3003      +45     
==========================================
+ Hits         2928     2963      +35     
- Misses         30       40      +10     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/Quantum-Accelerators/quacc/pull/1432?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Quantum-Accelerators).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Quantum-Accelerators).


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2941f020..d586a67b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,11 +4,15 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [0.4.7]
+## [0.5.0]
+
+### Added
+
+- Created a `quacc.wflow_tools.customizers` module to aid in customizing pre-made recipes.
 
 ### Changed
 
-- `quacc.recipes.vasp.mp.mp_relax_flow` now has modified kwargs that are no longer mutable to prevent subtle bugs.
+- Overhauled how sub-jobs in recipes can be dynamically modified by the user to increase flexibility.
 - Changed default `fmax` for `quacc.recipes.mlp.core.relax_job` to 0.05.
 - Renamed the `quacc.utils.dicts.merge_dicts` function to `._recursive_dict_pair_merge` to better reflect its functionality.
 - Renamed the `quacc.utils.dicts.merge_several_dicts` function to `.recursive_dict_merge` to better reflect its functionality.
diff --git a/docs/user/basics/wflow_syntax.md b/docs/user/basics/wflow_syntax.md
index 184381dd..7d28fafc 100644
--- a/docs/user/basics/wflow_syntax.md
+++ b/docs/user/basics/wflow_syntax.md
@@ -254,7 +254,7 @@ graph LR
 
 ## Stripping the Decorator from a Job
 
-If you ever want to strip the decorator from a pre-decorated `#!Python @job` for any reason, you can call the `.__wrapped__` attribute. This returns the original function.
+If you ever want to strip the decorator from a pre-decorated `#!Python @job` (e.g. to test out a calculation locally without changing your quacc settings), you can call the `.__wrapped__` attribute. This returns the original function.
 
 ```python
 from quacc import job
diff --git a/docs/user/recipes/modifications.md b/docs/user/recipes/modifications.md
new file mode 100644
index 00000000..3b1b3be3
--- /dev/null
+++ b/docs/user/recipes/modifications.md
@@ -0,0 +1,74 @@
+# Modifying Recipes
+
+Oftentimes, you will want to modify a recipe to suit your specific needs. This can be done in a number of ways, which we will explore in this section.
+
+## Modifying a Job in a Dynamic Workflow
+
+```mermaid
+graph LR
+  A[Input] --> B(Make Slabs)
+  B --> C(Slab Relax) --> G(Slab Static) --> K[Output]
+  B --> D(Slab Relax) --> H(Slab Static) --> K[Output]
+  B --> E(Slab Relax) --> I(Slab Static) --> K[Output]
+  B --> F(Slab Relax) --> J(Slab Static) --> K[Output];
+```
+
+In this example, we will run a pre-made workflow that generates a set of slabs from a bulk structure and then runs a structure relaxation and static calculation on each slab. We will specifically highlight an example where we want to override the default parameters of one step in the recipe, in this case to tighten the force tolerance for the slab relaxation step.
+
+!!! Tip
+
+    Unsure what arguments a given function takes? Check out the [API documentation](https://quantum-accelerators.github.io/quacc/reference/quacc/recipes/emt/slabs.html).
+
+```python
+from ase.build import bulk
+from quacc.recipes.emt.core import relax_job
+from quacc.recipes.emt.slabs import bulk_to_slabs_flow
+
+# Define the Atoms object
+atoms = bulk("Cu")
+
+# Define the workflow
+result = bulk_to_slabs_flow(
+    atoms, parameters={"relax_job": {"opt_params": {"fmax": 1e-4}}}  # (1)!
+)
+
+# Print the result
+print(result)
+```
+
+1. See the documentation for [quacc.wflow_tools.customizers.customize_funcs][] for additional details.
+
+Here, we have used the `parameters` keyword argument to override the default parameters of the `relax_job` function called within `bulk_to_slabs_flow`. This is a convenient way to modify the behavior of a workflow without having to modify the source code itself.
+
+## Modifying all the Jobs in a Dynamic Workflow
+
+What if you wanted to modify a common parameters of all the jobs in the aforementioned recipe? Rather than enumerate each one, you can use the magic keyword `"all"`:
+
+```python
+from ase.build import bulk
+from quacc.recipes.emt.core import relax_job
+from quacc.recipes.emt.slabs import bulk_to_slabs_flow
+
+# Define the Atoms object
+atoms = bulk("Cu")
+
+# Define the workflow
+result = bulk_to_slabs_flow(
+    atoms, parameters={"all": {"asap_cutoff": True}}  # (1)!
+)
+
+# Print the result
+print(result)
+```
+
+1. See the documentation for [quacc.wflow_tools.customizers.customize_funcs][] for additional details.
+
+!!! Note
+
+    If you try to modify a parameter that doesn't exist in all the corresponding jobs, you will get an error.
+
+## Concluding Comments
+
+At this point, you now have the basic idea of how quacc recipes work!
+
+If you don't care about using a workflow engine, feel free to write simple quacc-based Python scripts and submit them as-is using your favorite computing machine and scheduler. However, if you are looking to efficiently scale up and monitor large numbers of workflows, continue reading!
diff --git a/docs/user/recipes/recipes_intro.md b/docs/user/recipes/recipes_intro.md
index 67fb85bf..56481cdb 100644
--- a/docs/user/recipes/recipes_intro.md
+++ b/docs/user/recipes/recipes_intro.md
@@ -271,49 +271,3 @@ print(result2)
                 'version': '2.1.0'},
     'volume': 11.761470249999999}
     ```
-
-### Modifying Jobs in a Dynamic Workflow
-
-```mermaid
-graph LR
-  A[Input] --> B(Make Slabs)
-  B --> C(Slab Relax) --> G(Slab Static) --> K[Output]
-  B --> D(Slab Relax) --> H(Slab Static) --> K[Output]
-  B --> E(Slab Relax) --> I(Slab Static) --> K[Output]
-  B --> F(Slab Relax) --> J(Slab Static) --> K[Output];
-```
-
-In this example, we will run a pre-made workflow that generates a set of slabs from a bulk structure and then runs a structure relaxation and static calculation on each slab. We will specifically highlight an example where we want to override the default parameters of one step in the recipe, in this case to tighten the force tolerance for the slab relaxation.
-
-!!! Tip
-
-    Unsure what arguments a given function takes? Check out the [API documentation](https://quantum-accelerators.github.io/quacc/reference/quacc/recipes/emt/slabs.html).
-
-```python
-from functools import partial
-from ase.build import bulk
-from quacc.recipes.emt.core import relax_job
-from quacc.recipes.emt.slabs import bulk_to_slabs_flow
-
-# Define the Atoms object
-atoms = bulk("Cu")
-
-# Define the workflow
-custom_relax_job = partial(relax_job, opt_params={"fmax": 1e-4})  # (1)!
-result = bulk_to_slabs_flow(atoms, custom_relax_job=custom_relax_job)
-
-# Print the result
-print(result)
-```
-
-1. We have used a [partial function](https://docs.python.org/3/library/functools.html#functools.partial) here, which is a way to create a new function with specific arguments already applied. In other words, `#!Python opt_params={"fmax": 1e-4}` will be set as a keyword argument in the `relax_job` function by default. The same could be achieved, albeit more verbosely, as follows:
-   ```python
-   def custom_relax_job(*args, **kwargs):
-       return relax_job(*args, opt_params={"fmax": 1e-4}, **kwargs)
-   ```
-
-## Concluding Comments
-
-At this point, you now have the basic idea of how quacc recipes work!
-
-If you don't care about using a workflow engine, feel free to write simple quacc-based Python scripts and submit them as-is using your favorite computing machine and scheduler. However, if you are looking to efficiently scale up and monitor large numbers of workflows, continue reading!
diff --git a/docs/user/wflow_engine/wflow_engines3.md b/docs/user/wflow_engine/wflow_engines3.md
new file mode 100644
index 00000000..75144e66
--- /dev/null
+++ b/docs/user/wflow_engine/wflow_engines3.md
@@ -0,0 +1,44 @@
+# Modifying Decorators
+
+Occasionally, you may want to modify the decorators of select jobs in a workflow, such as to modify the executor of a given function. This can be done in an analagous way as in the [Modifying Recipes](../recipes/modifications.md) section except here we use the `decorators` keyword argument instead of the `parameters` keyword argument.
+
+## Modifying a Job in a Dynamic Workflow
+
+```mermaid
+graph LR
+  A[Input] --> B(Make Slabs)
+  B --> C(Slab Relax) --> G(Slab Static) --> K[Output]
+  B --> D(Slab Relax) --> H(Slab Static) --> K[Output]
+  B --> E(Slab Relax) --> I(Slab Static) --> K[Output]
+  B --> F(Slab Relax) --> J(Slab Static) --> K[Output];
+```
+
+In this example, we will run a pre-made workflow that generates a set of slabs from a bulk structure and then runs a structure relaxation and static calculation on each slab. We will specifically highlight an example where we want to modify the applied decorator of only the static calculation.
+
+!!! Tip
+
+    Unsure what arguments a given function takes? Check out the [API documentation](https://quantum-accelerators.github.io/quacc/reference/quacc/recipes/emt/slabs.html).
+
+```python
+from ase.build import bulk
+from quacc import job
+from quacc.recipes.emt.core import relax_job
+from quacc.recipes.emt.slabs import bulk_to_slabs_flow
+
+# Define the Atoms object
+atoms = bulk("Cu")
+
+# Define the workflow
+result = bulk_to_slabs_flow(
+    atoms, decorators={"static_job": job(MyKeywordArg=NewValue)}  # (1)!
+)
+
+# Print the result
+print(result)
+```
+
+1. See the documentation for [quacc.wflow_tools.customizers.customize_funcs][] for additional details.
+
+!!! Note
+
+    The `MyKeywordArg` and `NewValue` arguments are just placeholders. You will need to replace them with the appropriate values for your workflow engine and use case.
diff --git a/mkdocs.yml b/mkdocs.yml
index 5d5128c4..4b4015f5 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -15,6 +15,7 @@ nav:
       - Recipes:
           - user/recipes/recipes_intro.md
           - user/recipes/recipes_list.md
+          - user/recipes/modifications.md
       - Workflow Basics:
           - user/basics/wflow_overview.md
           - user/basics/wflow_decorators.md
@@ -22,6 +23,7 @@ nav:
       - Using a Workflow Engine:
           - user/wflow_engine/wflow_engines1.md
           - user/wflow_engine/wflow_engines2.md
+          - user/wflow_engine/wflow_engines3.md
           - user/wflow_engine/executors1.md
           - user/wflow_engine/executors2.md
       - Global Settings:
diff --git a/pyproject.toml b/pyproject.toml
index c37db306..be32a98c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "quacc"
 description="A platform to enable high-throughput, database-driven quantum chemistry and computational materials science"
-version = "0.4.7"
+version = "0.5.0"
 readme = "README.md"
 license = { text = "BSD-3" }
 authors = [{ name = "Andrew S. Rosen", email = "asrosen@princeton.edu" }]
diff --git a/src/quacc/calculators/espresso/espresso.py b/src/quacc/calculators/espresso/espresso.py
index daf0e301..0eb47068 100644
--- a/src/quacc/calculators/espresso/espresso.py
+++ b/src/quacc/calculators/espresso/espresso.py
@@ -260,18 +260,23 @@ class Espresso(Espresso_):
             calc_preset = load_yaml_calc(
                 SETTINGS.ESPRESSO_PRESET_DIR / f"{self.preset}"
             )
-            ecutwfc, ecutrho, pseudopotentials = get_pseudopotential_info(
-                calc_preset, self.input_atoms
-            )
-            calc_preset.pop("pseudopotentials", None)
-            self._user_calc_params = recursive_dict_merge(
-                calc_preset,
-                {
-                    "input_data": {"system": {"ecutwfc": ecutwfc, "ecutrho": ecutrho}},
-                    "pseudopotentials": pseudopotentials,
-                },
-                self.kwargs,
-            )
+            if "pseudopotentials" in calc_preset:
+                ecutwfc, ecutrho, pseudopotentials = get_pseudopotential_info(
+                    calc_preset["pseudopotentials"], self.input_atoms
+                )
+                calc_preset.pop("pseudopotentials", None)
+                self._user_calc_params = recursive_dict_merge(
+                    calc_preset,
+                    {
+                        "input_data": {
+                            "system": {"ecutwfc": ecutwfc, "ecutrho": ecutrho}
+                        },
+                        "pseudopotentials": pseudopotentials,
+                    },
+                    self.kwargs,
+                )
+            else:
+                self._user_calc_params = recursive_dict_merge(calc_preset, self.kwargs)
         else:
             self._user_calc_params = self.kwargs
 
diff --git a/src/quacc/calculators/espresso/utils.py b/src/quacc/calculators/espresso/utils.py
index afcd0a54..61895b0d 100644
--- a/src/quacc/calculators/espresso/utils.py
+++ b/src/quacc/calculators/espresso/utils.py
@@ -9,7 +9,7 @@ if TYPE_CHECKING:
 
 
 def get_pseudopotential_info(
-    config: dict[str, Any], atoms: Atoms
+    pp_dict: dict[str, Any], atoms: Atoms
 ) -> tuple[float, float, dict[str, str]]:
     """
     Function that parses the pseudopotentials and cutoffs from a preset file.
@@ -18,8 +18,8 @@ def get_pseudopotential_info(
 
     Parameters
     ----------
-    config
-        The config dictionary return by load_yaml_calc
+    pp_dict
+        The "pseudopotential" parameters returned by load_yaml_calc
     atoms
         The atoms object to get the elements from
 
@@ -33,16 +33,13 @@ def get_pseudopotential_info(
         The pseudopotentials dictinoary, e.g. {"O": "O.pbe-n-kjpaw_psl.0.1.UPF"}
     """
 
-    if "pseudopotentials" in config:
-        pp_dict = config["pseudopotentials"]
-        unique_elements = list(set(atoms.get_chemical_symbols()))
-        ecutwfc, ecutrho = 0, 0
-        pseudopotentials = {}
-        for element in unique_elements:
-            if pp_dict[element]["cutoff_wfc"] > ecutwfc:
-                ecutwfc = pp_dict[element]["cutoff_wfc"]
-            if pp_dict[element]["cutoff_rho"] > ecutrho:
-                ecutrho = pp_dict[element]["cutoff_rho"]
-            pseudopotentials[element] = pp_dict[element]["filename"]
-
+    unique_elements = list(set(atoms.get_chemical_symbols()))
+    ecutwfc, ecutrho = 0, 0
+    pseudopotentials = {}
+    for element in unique_elements:
+        if pp_dict[element]["cutoff_wfc"] > ecutwfc:
+            ecutwfc = pp_dict[element]["cutoff_wfc"]
+        if pp_dict[element]["cutoff_rho"] > ecutrho:
+            ecutrho = pp_dict[element]["cutoff_rho"]
+        pseudopotentials[element] = pp_dict[element]["filename"]
     return ecutwfc, ecutrho, pseudopotentials
diff --git a/src/quacc/recipes/emt/defects.py b/src/quacc/recipes/emt/defects.py
index 28fbb5b3..6a0f2326 100644
--- a/src/quacc/recipes/emt/defects.py
+++ b/src/quacc/recipes/emt/defects.py
@@ -9,9 +9,10 @@ from quacc import flow
 from quacc.recipes.common.defects import bulk_to_defects_subflow
 from quacc.recipes.emt.core import relax_job, static_job
 from quacc.utils.dicts import recursive_dict_merge
+from quacc.wflow_tools.customizers import customize_funcs
 
 if TYPE_CHECKING:
-    from typing import Any
+    from typing import Any, Callable
 
     from ase.atoms import Atoms
     from pymatgen.analysis.defects.generators import (
@@ -22,7 +23,6 @@ if TYPE_CHECKING:
         VoronoiInterstitialGenerator,
     )
 
-    from quacc import Job
     from quacc.schemas._aliases.ase import OptSchema, RunSchema
 
 
@@ -38,19 +38,19 @@ def bulk_to_defects_flow(
         | VoronoiInterstitialGenerator
     ) = VacancyGenerator,
     defect_charge: int = 0,
-    custom_relax_job: Job | None = None,
-    custom_static_job: Job | None = None,
     run_static: bool = True,
     make_defects_kwargs: dict[str, Any] | None = None,
+    decorators: dict[str, Callable | None] | None = None,
+    parameters: dict[str, Any] | None = None,
 ) -> list[RunSchema | OptSchema]:
     """
     Workflow consisting of:
 
     1. Defect generation
 
-    2. Defect relaxations
+    2. Defect relaxations ("relax_job")
 
-    3. Defect statics (optional)
+    3. Optional defect statics ("static_job")
 
     Parameters
     ----------
@@ -60,13 +60,17 @@ def bulk_to_defects_flow(
         Defect generator
     defect_charge
         Charge state of the defect
-    custom_relax_job
-        Relaxation job, which defaults to [quacc.recipes.emt.core.relax_job][].
-    custom_static_job
-        Static job, which defaults to [quacc.recipes.emt.core.static_job][].
+    run_static
+        Whether to run static calculations.
     make_defects_kwargs
         Keyword arguments to pass to
         [quacc.atoms.defects.make_defects_from_bulk][]
+    decorators
+        Custom decorators to apply to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
+    parameters
+        Custom parameters to pass to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
 
     Returns
     -------
@@ -77,12 +81,15 @@ def bulk_to_defects_flow(
     make_defects_kwargs = recursive_dict_merge(
         make_defects_kwargs, {"defect_gen": defect_gen, "defect_charge": defect_charge}
     )
-    defect_relax_job = relax_job if custom_relax_job is None else custom_relax_job
-    defect_static_job = static_job if custom_static_job is None else custom_static_job
+    relax_job_, static_job_ = customize_funcs(
+        {"relax_job": relax_job, "static_job": static_job},
+        decorators=decorators,
+        parameters=parameters,
+    )
 
     return bulk_to_defects_subflow(
         atoms,
-        defect_relax_job,
-        static_job=defect_static_job if run_static else None,
+        relax_job_,
+        static_job=static_job_ if run_static else None,
         make_defects_kwargs=make_defects_kwargs,
     )
diff --git a/src/quacc/recipes/emt/phonons.py b/src/quacc/recipes/emt/phonons.py
index 28c79104..b53c0781 100644
--- a/src/quacc/recipes/emt/phonons.py
+++ b/src/quacc/recipes/emt/phonons.py
@@ -6,12 +6,14 @@ from typing import TYPE_CHECKING
 from quacc import flow
 from quacc.recipes.common.phonons import phonon_flow as phonon_flow_
 from quacc.recipes.emt.core import static_job
+from quacc.wflow_tools.customizers import customize_funcs
 
 if TYPE_CHECKING:
+    from typing import Any, Callable
+
     from ase.atoms import Atoms
     from numpy.typing import ArrayLike
 
-    from quacc import Job
     from quacc.schemas._aliases.phonons import PhononSchema
 
 
@@ -24,10 +26,17 @@ def phonon_flow(
     t_step: float = 10,
     t_min: float = 0,
     t_max: float = 1000,
-    custom_static_job: Job | None = None,
+    decorators: dict[str, Callable | None] | None = None,
+    parameters: dict[str, Any] | None = None,
 ) -> PhononSchema:
     """
-    Carry out a phonon calculation.
+    Carry out a phonon workflow, consisting of:
+
+    1. Generation of supercells.
+
+    2. Static calculations on supercells ("static_job").
+
+    3. Calculation of thermodynamic properties.
 
     Parameters
     ----------
@@ -45,20 +54,25 @@ def phonon_flow(
         Min temperature (K).
     t_max
         Max temperature (K).
-    custom_static_job
-        The static job for the force calculations, which defaults
-        to [quacc.recipes.emt.core.static_job][]
+    decorators
+        Custom decorators to apply to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
+    parameters
+        Custom parameters to pass to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
 
     Returns
     -------
     PhononSchema
         Dictionary of results from [quacc.schemas.phonons.summarize_phonopy][]
     """
-    phonon_static_job = static_job if custom_static_job is None else custom_static_job
+    static_job_ = customize_funcs(
+        {"static_job": static_job}, decorators=decorators, parameters=parameters
+    )[0]
 
     return phonon_flow_(
         atoms,
-        phonon_static_job,
+        static_job_,
         supercell_matrix=supercell_matrix,
         atom_disp=atom_disp,
         t_step=t_step,
diff --git a/src/quacc/recipes/emt/slabs.py b/src/quacc/recipes/emt/slabs.py
index 823a6697..63ab07c0 100644
--- a/src/quacc/recipes/emt/slabs.py
+++ b/src/quacc/recipes/emt/slabs.py
@@ -6,46 +6,47 @@ from typing import TYPE_CHECKING
 from quacc import flow
 from quacc.recipes.common.slabs import bulk_to_slabs_subflow
 from quacc.recipes.emt.core import relax_job, static_job
+from quacc.wflow_tools.customizers import customize_funcs
 
 if TYPE_CHECKING:
-    from typing import Any
+    from typing import Any, Callable
 
     from ase.atoms import Atoms
 
-    from quacc import Job
     from quacc.schemas._aliases.ase import OptSchema, RunSchema
 
 
 @flow
 def bulk_to_slabs_flow(
     atoms: Atoms,
-    custom_relax_job: Job | None = None,
-    custom_static_job: Job | None = None,
     run_static: bool = True,
     make_slabs_kwargs: dict[str, Any] | None = None,
+    decorators: dict[str, Callable | None] | None = None,
+    parameters: dict[str, Any] | None = None,
 ) -> list[RunSchema | OptSchema]:
     """
     Workflow consisting of:
 
     1. Slab generation
 
-    2. Slab relaxations
+    2. Slab relaxations ("relax_job")
 
-    3. Slab statics (optional)
+    3. Optional slab statics ("static_job")
 
     Parameters
     ----------
     atoms
         Atoms object
-    custom_relax_job
-        The relaxation job, which defaults to [quacc.recipes.emt.core.relax_job][].
-    custom_static_job
-        The static job, which defaults to [quacc.recipes.emt.core.static_job][].
     run_static
         Whether to run static calculations.
     make_slabs_kwargs
-        Additional keyword arguments to pass to
-        [quacc.atoms.slabs.make_slabs_from_bulk][]
+        Additional keyword arguments to pass to [quacc.atoms.slabs.make_slabs_from_bulk][]
+    decorators
+        Custom decorators to apply to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
+    parameters
+        Custom parameters to pass to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
 
     Returns
     -------
@@ -53,12 +54,15 @@ def bulk_to_slabs_flow(
         [RunSchema][quacc.schemas.ase.summarize_run] or
         [OptSchema][quacc.schemas.ase.summarize_opt_run] for each slab.
     """
-    slab_relax_job = relax_job if custom_relax_job is None else custom_relax_job
-    slab_static_job = static_job if custom_static_job is None else custom_static_job
+    relax_job_, static_job_ = customize_funcs(
+        {"relax_job": relax_job, "static_job": static_job},
+        decorators=decorators,
+        parameters=parameters,
+    )
 
     return bulk_to_slabs_subflow(
         atoms,
-        slab_relax_job,
-        static_job=slab_static_job if run_static else None,
+        relax_job_,
+        static_job=static_job_ if run_static else None,
         make_slabs_kwargs=make_slabs_kwargs,
     )
diff --git a/src/quacc/recipes/tblite/phonons.py b/src/quacc/recipes/tblite/phonons.py
index b64b1a7d..baf4411a 100644
--- a/src/quacc/recipes/tblite/phonons.py
+++ b/src/quacc/recipes/tblite/phonons.py
@@ -6,12 +6,14 @@ from typing import TYPE_CHECKING
 from quacc import flow
 from quacc.recipes.common.phonons import phonon_flow as common_phonon_flow
 from quacc.recipes.tblite.core import static_job
+from quacc.wflow_tools.customizers import customize_funcs
 
 if TYPE_CHECKING:
+    from typing import Any, Callable
+
     from ase.atoms import Atoms
     from numpy.typing import ArrayLike
 
-    from quacc import Job
     from quacc.schemas._aliases.phonons import PhononSchema
 
 
@@ -24,10 +26,17 @@ def phonon_flow(
     t_step: float = 10,
     t_min: float = 0,
     t_max: float = 1000,
-    custom_static_job: Job | None = None,
+    decorators: dict[str, Callable | None] | None = None,
+    parameters: dict[str, Any] | None = None,
 ) -> PhononSchema:
     """
-    Carry out a phonon calculation.
+    Carry out a phonon workflow, consisting of:
+
+    1. Generation of supercells.
+
+    2. Static calculations on supercells ("static_job").
+
+    3. Calculation of thermodynamic properties.
 
     Parameters
     ----------
@@ -45,18 +54,25 @@ def phonon_flow(
         Min temperature (K).
     t_max
         Max temperature (K).
-    custom_static_job
-        Static job, which defaults to [quacc.recipes.tblite.core.static_job][]
+    decorators
+        Custom decorators to apply to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
+    parameters
+        Custom parameters to pass to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
 
     Returns
     -------
     PhononSchema
         Dictionary of results from [quacc.schemas.phonons.summarize_phonopy][]
     """
+    static_job_ = customize_funcs(
+        {"static_job": static_job}, decorators=decorators, parameters=parameters
+    )[0]
 
     return common_phonon_flow(
         atoms,
-        static_job if custom_static_job is None else custom_static_job,
+        static_job_,
         supercell_matrix=supercell_matrix,
         atom_disp=atom_disp,
         t_step=t_step,
diff --git a/src/quacc/recipes/vasp/mp.py b/src/quacc/recipes/vasp/mp.py
index a9e6d158..e6be9980 100644
--- a/src/quacc/recipes/vasp/mp.py
+++ b/src/quacc/recipes/vasp/mp.py
@@ -22,11 +22,13 @@ import numpy as np
 
 from quacc import flow, job
 from quacc.recipes.vasp._base import base_fn
+from quacc.wflow_tools.customizers import customize_funcs
 
 if TYPE_CHECKING:
+    from typing import Any, Callable
+
     from ase.atoms import Atoms
 
-    from quacc import Job
     from quacc.schemas._aliases.vasp import MPRelaxFlowSchema, VaspSchema
 
 
@@ -127,40 +129,43 @@ def mp_relax_job(
 @flow
 def mp_relax_flow(
     atoms: Atoms,
-    custom_prerelax_job: Job | None = None,
-    custom_relax_job: Job | None = None,
+    decorators: dict[str, Callable | None] | None = None,
+    parameters: dict[str, Any] | None = None,
 ) -> MPRelaxFlowSchema:
     """
     Workflow consisting of:
 
-    1. MP-compatible pre-relax
+    1. MP-compatible pre-relax ("mp_prerelax_job")
 
-    2. MP-compatible relax
+    2. MP-compatible relax ("mp_relax_job")
 
     Parameters
     ----------
     atoms
         Atoms object for the structure.
-    custom_prerelax_job
-        Pre-relaxation job, which defaults to [quacc.recipes.vasp.mp.mp_prerelax_job][].
-    custom_relax_job
-        Relaxation job, which defaults to [quacc.recipes.vasp.mp.mp_relax_job][].
+    decorators
+        Custom decorators to apply to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
+    parameters
+        Custom parameters to pass to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
 
     Returns
     -------
     MPRelaxFlowSchema
         Dictionary of results
     """
+    mp_prerelax_job_, mp_relax_job_ = customize_funcs(
+        {"mp_prerelax_job": mp_prerelax_job, "mp_relax_job": mp_relax_job},
+        decorators=decorators,
+        parameters=parameters,
+    )
 
     # Run the prerelax
-    prerelax_job = (
-        mp_prerelax_job if custom_prerelax_job is None else custom_prerelax_job
-    )
-    prerelax_results = prerelax_job(atoms)
+    prerelax_results = mp_prerelax_job_(atoms)
 
     # Run the relax
-    relax_job = mp_relax_job if custom_relax_job is None else custom_relax_job
-    relax_results = relax_job(
+    relax_results = mp_relax_job_(
         prerelax_results["atoms"],
         bandgap=prerelax_results["output"]["bandgap"],
         copy_files=[
diff --git a/src/quacc/recipes/vasp/slabs.py b/src/quacc/recipes/vasp/slabs.py
index 784a04d8..2eb6344c 100644
--- a/src/quacc/recipes/vasp/slabs.py
+++ b/src/quacc/recipes/vasp/slabs.py
@@ -6,14 +6,14 @@ from typing import TYPE_CHECKING
 from quacc import flow, job
 from quacc.recipes.common.slabs import bulk_to_slabs_subflow, slab_to_ads_subflow
 from quacc.recipes.vasp._base import base_fn
+from quacc.wflow_tools.customizers import customize_funcs
 
 if TYPE_CHECKING:
     from pathlib import Path
-    from typing import Any
+    from typing import Any, Callable
 
     from ase.atoms import Atoms
 
-    from quacc import Job
     from quacc.schemas._aliases.vasp import VaspSchema
 
 
@@ -121,18 +121,18 @@ def relax_job(
 def bulk_to_slabs_flow(
     atoms: Atoms,
     make_slabs_kwargs: dict[str, Any] | None = None,
-    custom_relax_job: Job | None = None,
-    custom_static_job: Job | None = None,
     run_static: bool = True,
+    decorators: dict[str, Callable | None] | None = None,
+    parameters: dict[str, Any] | None = None,
 ) -> list[VaspSchema]:
     """
     Workflow consisting of:
 
     1. Slab generation
 
-    2. Slab relaxations
+    2. Slab relaxations ("relax_job")
 
-    3. Slab statics (optional)
+    3. Optional slab statics ("static_job")
 
     Parameters
     ----------
@@ -140,25 +140,30 @@ def bulk_to_slabs_flow(
         Atoms object
     make_slabs_kwargs
         Additional keyword arguments to pass to [quacc.atoms.slabs.make_slabs_from_bulk][]
-    custom_relax_job
-        Relaxation job, which defaults to [quacc.recipes.vasp.slabs.relax_job][].
-    custom_static_job
-        Static job, which defaults to [quacc.recipes.vasp.slabs.static_job][].
     run_static
         Whether to run static calculations.
+    decorators
+        Custom decorators to apply to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
+    parameters
+        Custom parameters to pass to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
 
     Returns
     -------
     list[VaspSchema]
         List of dictionary results from [quacc.schemas.vasp.vasp_summarize_run][]
     """
-    slab_relax_job = relax_job if custom_relax_job is None else custom_relax_job
-    slab_static_job = static_job if custom_static_job is None else custom_static_job
+    relax_job_, static_job_ = customize_funcs(
+        {"relax_job": relax_job, "static_job": static_job},
+        decorators=decorators,
+        parameters=parameters,
+    )
 
     return bulk_to_slabs_subflow(
         atoms,
-        slab_relax_job,
-        static_job=slab_static_job if run_static else None,
+        relax_job_,
+        static_job=static_job_ if run_static else None,
         make_slabs_kwargs=make_slabs_kwargs,
     )
 
@@ -167,19 +172,19 @@ def bulk_to_slabs_flow(
 def slab_to_ads_flow(
     slab: Atoms,
     adsorbate: Atoms,
-    custom_relax_job: Job | None = None,
-    custom_static_job: Job | None = None,
     run_static: bool = True,
     make_ads_kwargs: dict[str, Any] | None = None,
+    decorators: dict[str, Callable | None] | None = None,
+    parameters: dict[str, Any] | None = None,
 ) -> list[VaspSchema]:
     """
     Workflow consisting of:
 
     1. Slab-adsorbate generation
 
-    2. Slab-adsorbate relaxations
+    2. Slab-adsorbate relaxations ("relax_job")
 
-    3. Slab-adsorbate statics (optional)
+    3. Optional slab-adsorbate statics ("static_job")
 
     Parameters
     ----------
@@ -187,28 +192,32 @@ def slab_to_ads_flow(
         Atoms object for the slab structure.
     adsorbate
         Atoms object for the adsorbate.
-    custom_relax_job
-        Relaxation job, which defaults to [quacc.recipes.vasp.slabs.relax_job][].
-    custom_static_job
-        Static job, which defaults to [quacc.recipes.vasp.slabs.static_job][].
     run_static
         Whether to run static calculations.
     make_ads_kwargs
         Additional keyword arguments to pass to [quacc.atoms.slabs.make_adsorbate_structures][]
+    decorators
+        Custom decorators to apply to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
+    parameters
+        Custom parameters to pass to each Job in the Flow.
+        Refer to [quacc.wflow_tools.customizers.customize_funcs][] for details.
 
     Returns
     -------
     list[VaspSchema]
         List of dictionaries of results from [quacc.schemas.vasp.vasp_summarize_run][]
     """
-
-    slab_relax_job = relax_job if custom_relax_job is None else custom_relax_job
-    slab_static_job = static_job if custom_static_job is None else custom_static_job
+    relax_job_, static_job_ = customize_funcs(
+        {"relax_job": relax_job, "static_job": static_job},
+        decorators=decorators,
+        parameters=parameters,
+    )
 
     return slab_to_ads_subflow(
         slab,
         adsorbate,
-        slab_relax_job,
-        static_job=slab_static_job if run_static else None,
+        relax_job_,
+        static_job=static_job_ if run_static else None,
         make_ads_kwargs=make_ads_kwargs,
     )
diff --git a/src/quacc/wflow_tools/customizers.py b/src/quacc/wflow_tools/customizers.py
new file mode 100644
index 00000000..1e3c0485
--- /dev/null
+++ b/src/quacc/wflow_tools/customizers.py
@@ -0,0 +1,139 @@
+"""Functions to customize workflow steps."""
+from __future__ import annotations
+
+from functools import partial
+from typing import TYPE_CHECKING
+
+from quacc import SETTINGS, job
+
+if TYPE_CHECKING:
+    from typing import Any, Callable
+
+
+def strip_decorator(func: Callable) -> Callable:
+    """
+    Strip the decorators from a function.
+
+    Parameters
+    ----------
+    func
+        The function to strip decorators from.
+
+    Returns
+    -------
+    Callable
+        The function with all decorators removed.
+    """
+
+    if hasattr(func, "__wrapped__"):
+        func = func.__wrapped__
+
+    if SETTINGS.WORKFLOW_ENGINE == "covalent":
+        from covalent._workflow.lattice import Lattice
+
+        if isinstance(func, Lattice):
+            func = func.workflow_function.get_deserialized()
+    return func
+
+
+def redecorate(func: Callable, decorator: Callable | None) -> Callable:
+    """
+    Redecorate pre-decorated functions with custom decorators.
+
+    Parameters
+    ----------
+    func
+        The pre-decorated function.
+    decorator
+        The new decorator to apply. If `None`, the function is stripped of its
+        decorators.
+
+    Returns
+    -------
+    Callable
+        The newly decorated function.
+    """
+    func = strip_decorator(func)
+    return func if decorator is None else decorator(func)
+
+
+def update_parameters(func: Callable, params: dict[str, Any]) -> Callable:
+    """
+    Update the parameters of a function. If the function does not have a given parameter,
+    it is ignored.
+
+    Parameters
+    ----------
+    func
+        The function to update.
+    params
+        The parameters and associated values to update.
+
+    Returns
+    -------
+    Callable
+        The updated function.
+    """
+    if SETTINGS.WORKFLOW_ENGINE == "dask":
+        func = strip_decorator(func)
+        updated_func = job(partial(func, **params))
+    else:
+        updated_func = partial(func, **params)
+
+    return updated_func
+
+
+def customize_funcs(
+    funcs: dict[str, Callable],
+    decorators: dict[str, Callable | None] | None = None,
+    parameters: dict[str, Any] | None = None,
+) -> tuple[Callable]:
+    """
+    Customize a set of functions with decorators and common parameters.
+
+    Parameters
+    ----------
+    funcs
+        The functions to customize, as a dictionary where the keys are unique
+        identifiers for each function and the values are the functions themselves.
+    decorators
+        Custom decorators to apply to each function. The keys of this dictionary correspond
+        to the keys of `funcs`. If the key `"all"` is present, it will be applied to all
+        functions. If a value is `None`, no decorator will be applied that function.
+    parameters
+        Custom parameters to apply to each function. The keys of this dictionary correspond
+        to the keys of `funcs`. If the key `"all"` is present, it will be applied to all
+        functions. If the value is `None`, no custom parameters will be applied to that function.
+
+    Returns
+    -------
+    tuple[Callable]
+        The customized functions, returned in the same order as provided in `funcs`.
+    """
+    decorators = decorators or {}
+    parameters = parameters or {}
+    updated_funcs = []
+
+    if bad_decorator_keys := [k for k in decorators if k not in funcs and k != "all"]:
+        raise ValueError(
+            f"Invalid decorator keys: {bad_decorator_keys}. "
+            f"Valid keys are: {list(funcs.keys())}"
+        )
+    if bad_parameter_keys := [k for k in parameters if k not in funcs and k != "all"]:
+        raise ValueError(
+            f"Invalid parameter keys: {bad_parameter_keys}. "
+            f"Valid keys are: {list(funcs.keys())}"
+        )
+
+    for func_name, func in funcs.items():
+        func_ = func
+        if params := parameters.get("all"):
+            func_ = update_parameters(func_, params)
+        if params := parameters.get(func_name):
+            func_ = update_parameters(func_, params)
+        if "all" in decorators:
+            func_ = redecorate(func_, decorators["all"])
+        if func_name in decorators:
+            func_ = redecorate(func_, decorators[func_name])
+        updated_funcs.append(func_)
+    return tuple(updated_funcs)
diff --git a/src/quacc/wflow_tools/decorators.py b/src/quacc/wflow_tools/decorators.py
index 68386574..dde2adaf 100644
--- a/src/quacc/wflow_tools/decorators.py
+++ b/src/quacc/wflow_tools/decorators.py
@@ -1,7 +1,7 @@
 """Workflow decorators."""
 from __future__ import annotations
 
-import functools
+from functools import partial, wraps
 from typing import TYPE_CHECKING, TypeVar
 
 Job = TypeVar("Job")
@@ -117,7 +117,7 @@ def job(_func: Callable | None = None, **kwargs) -> Job:
     wflow_engine = SETTINGS.WORKFLOW_ENGINE
 
     if _func is None:
-        return functools.partial(job, **kwargs)
+        return partial(job, **kwargs)
 
     if wflow_engine == "covalent":
         import covalent as ct
@@ -261,7 +261,7 @@ def flow(
     from quacc import SETTINGS
 
     if _func is None:
-        return functools.partial(flow, **kwargs)
+        return partial(flow, **kwargs)
 
     wflow_engine = SETTINGS.WORKFLOW_ENGINE
     if wflow_engine == "covalent":
@@ -275,6 +275,9 @@ def flow(
     else:
         decorated = _func
 
+    if wflow_engine != "covalent" and not hasattr(decorated, "__wrapped__"):
+        decorated.__wrapped__ = _func
+
     return decorated
 
 
@@ -444,7 +447,7 @@ def subflow(
         The decorated function.
     """
 
-    @functools.wraps(_func)
+    @wraps(_func)
     def _inner(
         *f_args, decorator_kwargs: dict[str, Any] | None = None, **f_kwargs
     ) -> Any:
@@ -479,13 +482,13 @@ def subflow(
     from quacc import SETTINGS
 
     if _func is None:
-        return functools.partial(subflow, **kwargs)
+        return partial(subflow, **kwargs)
 
     wflow_engine = SETTINGS.WORKFLOW_ENGINE
     if wflow_engine == "covalent":
         import covalent as ct
 
-        decorated = ct.electron(ct.lattice(_func, **kwargs))
+        decorated = ct.electron(ct.lattice(_func), **kwargs)
     elif wflow_engine == "parsl":
         from parsl import join_app
 
@@ -499,4 +502,7 @@ def subflow(
     else:
         decorated = _func
 
+    if wflow_engine != "covalent" and not hasattr(decorated, "__wrapped__"):
+        decorated.__wrapped__ = _func
+
     return decorated


Test patch:
diff --git a/tests/core/recipes/emt_recipes/test_emt_defect_recipes.py b/tests/core/recipes/emt_recipes/test_emt_defect_recipes.py
index 68777006..eddd1441 100644
--- a/tests/core/recipes/emt_recipes/test_emt_defect_recipes.py
+++ b/tests/core/recipes/emt_recipes/test_emt_defect_recipes.py
@@ -1,11 +1,8 @@
-from functools import partial
-
 import pytest
 from ase.build import bulk
 
 pytest.importorskip("pymatgen.analysis.defects")
 
-from quacc.recipes.emt.core import relax_job
 from quacc.recipes.emt.defects import bulk_to_defects_flow
 
 
@@ -13,16 +10,14 @@ def test_bulk_to_defects_flow(tmp_path, monkeypatch):
     monkeypatch.chdir(tmp_path)
     atoms = bulk("Cu")
     output = bulk_to_defects_flow(
-        atoms, custom_relax_job=partial(relax_job, opt_params={"fmax": 5})
+        atoms, parameters={"relax_job": {"opt_params": {"fmax": 5}}}
     )
     assert len(output) == 1
     assert len(output[0]["atoms"]) == 107
 
     atoms = bulk("Cu")
     output = bulk_to_defects_flow(
-        atoms,
-        custom_relax_job=partial(relax_job, opt_params={"fmax": 5}),
-        run_static=False,
+        atoms, parameters={"relax_job": {"opt_params": {"fmax": 5}}}, run_static=False
     )
     assert len(output) == 1
     assert len(output[0]["atoms"]) == 107
diff --git a/tests/core/recipes/emt_recipes/test_emt_recipes.py b/tests/core/recipes/emt_recipes/test_emt_recipes.py
index 11927525..0b01e801 100644
--- a/tests/core/recipes/emt_recipes/test_emt_recipes.py
+++ b/tests/core/recipes/emt_recipes/test_emt_recipes.py
@@ -1,5 +1,3 @@
-from functools import partial
-
 import numpy as np
 import pytest
 from ase.build import bulk, molecule
@@ -96,7 +94,7 @@ def test_slab_dynamic_jobs(tmp_path, monkeypatch):
     outputs = bulk_to_slabs_flow(
         atoms,
         run_static=False,
-        custom_relax_job=partial(relax_job, opt_params={"fmax": 1.0}, asap_cutoff=True),
+        parameters={"relax_job": {"opt_params": {"fmax": 1.0}, "asap_cutoff": True}},
     )
     assert len(outputs) == 4
     assert outputs[0]["nsites"] == 80
@@ -104,3 +102,36 @@ def test_slab_dynamic_jobs(tmp_path, monkeypatch):
     assert outputs[2]["nsites"] == 80
     assert outputs[3]["nsites"] == 64
     assert [output["parameters"]["asap_cutoff"] is True for output in outputs]
+
+
+def test_customizer():
+    atoms = bulk("Cu")
+    results = bulk_to_slabs_flow(
+        atoms, parameters={"static_job": {"asap_cutoff": True}}
+    )
+    for result in results:
+        assert result["parameters"]["asap_cutoff"] is True
+
+
+def test_customizer_v2():
+    atoms = bulk("Cu")
+    results = bulk_to_slabs_flow(atoms, parameters={"relax_job": {"asap_cutoff": True}})
+    for result in results:
+        assert result["parameters"]["asap_cutoff"] is False
+
+
+def test_all_customizers():
+    atoms = bulk("Cu")
+    results = bulk_to_slabs_flow(atoms, parameters={"all": {"asap_cutoff": True}})
+    for result in results:
+        assert result["parameters"]["asap_cutoff"] is True
+
+
+def test_all_customizers_v2():
+    atoms = bulk("Cu")
+    results = bulk_to_slabs_flow(
+        atoms,
+        parameters={"all": {"asap_cutoff": True}, "static_job": {"asap_cutoff": False}},
+    )
+    for result in results:
+        assert result["parameters"]["asap_cutoff"] is False
diff --git a/tests/core/recipes/tblite_recipes/test_tblite_phonons.py b/tests/core/recipes/tblite_recipes/test_tblite_phonons.py
index 0e5515e9..6acce83e 100644
--- a/tests/core/recipes/tblite_recipes/test_tblite_phonons.py
+++ b/tests/core/recipes/tblite_recipes/test_tblite_phonons.py
@@ -1,9 +1,6 @@
-from functools import partial
-
 import pytest
 from ase.build import bulk
 
-from quacc.recipes.tblite.core import static_job as static_job_
 from quacc.recipes.tblite.phonons import phonon_flow
 
 pytest.importorskip("tblite.ase")
@@ -13,8 +10,6 @@ pytest.importorskip("phonopy")
 def test_phonon_flow(tmp_path, monkeypatch):
     monkeypatch.chdir(tmp_path)
     atoms = bulk("Cu")
-    output = phonon_flow(
-        atoms, custom_static_job=partial(static_job_, method="GFN1-xTB")
-    )
+    output = phonon_flow(atoms, parameters={"static_job": {"method": "GFN1-xTB"}})
     assert output["results"]["force_constants"].shape == (8, 8, 3, 3)
     assert len(output["results"]["thermal_properties"]["temperatures"]) == 101
diff --git a/tests/core/recipes/vasp_recipes/mocked/test_vasp_recipes.py b/tests/core/recipes/vasp_recipes/mocked/test_vasp_recipes.py
index 0eaac567..cdd6a0d8 100644
--- a/tests/core/recipes/vasp_recipes/mocked/test_vasp_recipes.py
+++ b/tests/core/recipes/vasp_recipes/mocked/test_vasp_recipes.py
@@ -1,5 +1,3 @@
-from functools import partial
-
 import pytest
 from ase.build import bulk, molecule
 
@@ -219,7 +217,7 @@ def test_slab_dynamic_jobs(tmp_path, monkeypatch):
 
     outputs = bulk_to_slabs_flow(
         atoms,
-        custom_relax_job=partial(slab_relax_job, preset="SlabSet", nelmin=6),
+        parameters={"relax_job": {"preset": "SlabSet", "nelmin": 6}},
         run_static=False,
     )
     assert len(outputs) == 4
@@ -232,7 +230,7 @@ def test_slab_dynamic_jobs(tmp_path, monkeypatch):
     assert [output["parameters"]["encut"] == 450 for output in outputs]
 
     outputs = bulk_to_slabs_flow(
-        atoms, custom_relax_job=partial(slab_relax_job, preset="SlabSet", nelmin=6)
+        atoms, parameters={"relax_job": {"preset": "SlabSet", "nelmin": 6}}
     )
     assert len(outputs) == 4
     assert outputs[0]["nsites"] == 45
@@ -259,7 +257,7 @@ def test_slab_dynamic_jobs(tmp_path, monkeypatch):
     outputs = slab_to_ads_flow(
         atoms,
         adsorbate,
-        custom_relax_job=partial(slab_relax_job, preset="SlabSet", nelmin=6),
+        parameters={"relax_job": {"preset": "SlabSet", "nelmin": 6}},
         run_static=False,
     )
 
@@ -269,9 +267,7 @@ def test_slab_dynamic_jobs(tmp_path, monkeypatch):
     assert [output["parameters"]["encut"] == 450 for output in outputs]
 
     outputs = slab_to_ads_flow(
-        atoms,
-        adsorbate,
-        custom_static_job=partial(slab_static_job, preset="SlabSet", nelmin=6),
+        atoms, adsorbate, parameters={"relax_job": {"preset": "SlabSet", "nelmin": 6}}
     )
 
     assert [output["nsites"] == 82 for output in outputs]
diff --git a/tests/core/wflow/test_customizers.py b/tests/core/wflow/test_customizers.py
new file mode 100644
index 00000000..10b5e3ae
--- /dev/null
+++ b/tests/core/wflow/test_customizers.py
@@ -0,0 +1,32 @@
+import pytest
+
+from quacc.wflow_tools.customizers import customize_funcs
+
+
+def test_basic_customizers():
+    def add(a, b=1, c=2):
+        return a + b + c
+
+    def mult(a, b=1, c=2, d=2):
+        return a * b * c * d
+
+    add_, mult_ = customize_funcs(
+        {"add": add, "mult": mult}, parameters={"add": {"b": 2}}
+    )
+    assert add_(1) == 5
+    assert mult_(1) == 4
+
+    add_, mult_ = customize_funcs(
+        {"add": add, "mult": mult}, parameters={"add": {"b": 2}, "mult": {"b": 2}}
+    )
+    assert add_(1) == 5
+    assert mult_(1) == 8
+
+    add_, mult_ = customize_funcs(
+        {"add": add, "mult": mult}, parameters={"all": {"b": 2}}
+    )
+    assert add_(1) == 5
+    assert mult_(1) == 8
+
+    with pytest.raises(ValueError):
+        customize_funcs({"add": add, "mult": mult}, parameters={"bad": {"b": 2}})
diff --git a/tests/covalent/test_covalent_recipes.py b/tests/covalent/test_covalent_recipes.py
index efc2fb41..00b5c1fb 100644
--- a/tests/covalent/test_covalent_recipes.py
+++ b/tests/covalent/test_covalent_recipes.py
@@ -1,5 +1,3 @@
-from functools import partial
-
 import pytest
 from ase.build import bulk
 
@@ -19,9 +17,7 @@ def test_covalent_functools(tmp_path, monkeypatch):
     monkeypatch.chdir(tmp_path)
     atoms = bulk("Cu")
     dispatch_id = ct.dispatch(bulk_to_slabs_flow)(
-        atoms,
-        custom_relax_job=partial(relax_job, opt_params={"fmax": 0.1}),
-        run_static=False,
+        atoms, parameters={"relax_job": {"opt_params": {"fmax": 0.1}}}, run_static=False
     )
     output = ct.get_result(dispatch_id, wait=True)
     assert output.status == "COMPLETED"
diff --git a/tests/covalent/test_customizers.py b/tests/covalent/test_customizers.py
new file mode 100644
index 00000000..31f4e7e6
--- /dev/null
+++ b/tests/covalent/test_customizers.py
@@ -0,0 +1,85 @@
+import pytest
+
+from quacc import SETTINGS, flow, job, subflow
+from quacc.wflow_tools.customizers import customize_funcs
+
+ct = pytest.importorskip("covalent")
+pytestmark = pytest.mark.skipif(
+    SETTINGS.WORKFLOW_ENGINE != "covalent",
+    reason="This test requires the Covalent workflow engine",
+)
+
+
+def test_customize_jobs():
+    @job
+    def add(a, b):
+        return a + b
+
+    @job
+    def mult(a, b):
+        return a * b
+
+    add_, mult_ = customize_funcs(
+        {"add": add, "mult": mult}, decorators={"add": job(executor="local")}
+    )
+    assert add_.electron_object.executor == "local"
+    assert mult_.electron_object.executor is None
+
+    add_, mult_ = customize_funcs(
+        {"add": add, "mult": mult},
+        decorators={"add": job(executor="local"), "mult": job(executor="local")},
+    )
+    assert add_.electron_object.executor == "local"
+    assert mult_.electron_object.executor == "local"
+
+    add_, mult_ = customize_funcs(
+        {"add": add, "mult": mult}, decorators={"all": job(executor="local")}
+    )
+    assert add_.electron_object.executor == "local"
+    assert mult_.electron_object.executor == "local"
+
+    add_, mult_ = customize_funcs(
+        {"add": add, "mult": mult},
+        decorators={"all": job(executor="local"), "mult": job(executor="dask")},
+    )
+    assert add_.electron_object.executor == "local"
+    assert mult_.electron_object.executor == "dask"
+
+
+def test_customize_flows():
+    @job
+    def add(a, b):
+        return a + b
+
+    @flow
+    def workflow():
+        return add(1, 2)
+
+    workflow_ = customize_funcs(
+        {"workflow": workflow}, decorators={"workflow": flow(executor="local")}
+    )[0]
+    assert workflow_.metadata["executor"] == "local"
+
+
+def test_customize_subflows():
+    @job
+    def add(a, b):
+        return a + b
+
+    @subflow
+    def subworkflow():
+        return add(1, 2)
+
+    subworkflow_ = customize_funcs(
+        {"workflow": subworkflow}, decorators={"workflow": subflow(executor="local")}
+    )[0]
+    assert subworkflow_.electron_object.executor == "local"
+
+
+def test_bad_customizers():
+    @job
+    def add(a, b):
+        return a + b
+
+    with pytest.raises(ValueError):
+        customize_funcs({"add": add}, decorators={"bad": job(executor="local")})
diff --git a/tests/dask/test_dask_recipes.py b/tests/dask/test_dask_recipes.py
index a058ac97..3302301f 100644
--- a/tests/dask/test_dask_recipes.py
+++ b/tests/dask/test_dask_recipes.py
@@ -1,5 +1,3 @@
-from functools import partial
-
 import pytest
 from ase.build import bulk
 
@@ -20,16 +18,10 @@ client = default_client()
 
 
 def test_dask_functools(tmp_path, monkeypatch):
-    from dask import delayed as delayed_
-
     monkeypatch.chdir(tmp_path)
     atoms = bulk("Cu")
     delayed = bulk_to_slabs_flow(
-        atoms,
-        custom_relax_job=delayed_(
-            partial(relax_job.__wrapped__, opt_params={"fmax": 0.1})
-        ),
-        run_static=False,
+        atoms, parameters={"relax_job": {"opt_params": {"fmax": 0.1}}}, run_static=False
     )
     result = client.gather(client.compute(delayed))
     assert len(result) == 4
diff --git a/tests/parsl/test_parsl_recipes.py b/tests/parsl/test_parsl_recipes.py
index 6323a36f..a76c9a4f 100644
--- a/tests/parsl/test_parsl_recipes.py
+++ b/tests/parsl/test_parsl_recipes.py
@@ -1,5 +1,4 @@
 import contextlib
-from functools import partial
 
 import pytest
 from ase.build import bulk
@@ -29,9 +28,7 @@ def test_parsl_functools(tmp_path, monkeypatch):
     monkeypatch.chdir(tmp_path)
     atoms = bulk("Cu")
     result = bulk_to_slabs_flow(
-        atoms,
-        custom_relax_job=partial(relax_job, opt_params={"fmax": 0.1}),
-        run_static=False,
+        atoms, parameters={"relax_job": {"opt_params": {"fmax": 0.1}}}, run_static=False
     ).result()
     assert len(result) == 4
     assert "atoms" in result[-1]
diff --git a/tests/redun/test_redun_recipes.py b/tests/redun/test_redun_recipes.py
index 3675ece7..8f0eb1eb 100644
--- a/tests/redun/test_redun_recipes.py
+++ b/tests/redun/test_redun_recipes.py
@@ -1,5 +1,3 @@
-from functools import partial
-
 import pytest
 from ase.build import bulk
 
@@ -17,7 +15,6 @@ def scheduler():
     return redun.Scheduler()
 
 
-from quacc.recipes.emt.core import relax_job  # skipcq: PYL-C0412
 from quacc.recipes.emt.slabs import bulk_to_slabs_flow  # skipcq: PYL-C0412
 
 
@@ -27,7 +24,7 @@ def test_redun_functools(tmp_path, monkeypatch, scheduler):
     result = scheduler.run(
         bulk_to_slabs_flow(
             atoms,
-            custom_relax_job=partial(relax_job, opt_params={"fmax": 0.1}),
+            parameters={"relax_job": {"opt_params": {"fmax": 0.1}}},
             run_static=False,
         )
     )