We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Core docs language
https://docs.prefect.io/core/

In a few places the Core docs mention that Cloud is coming soon, and I believe the language should be updated.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/prefect/engine/cache_validators.py`
Content:
```
1 """
2 Cache validators are functions that determine if a task's output cache
3 is still valid, or whether that task should be re-run; they are provided at
4 Task creation via the `cache_validator` keyword argument (for more information
5 on instantiating Tasks see the [Task documentation](../core/task.html)).
6 
7 Task caches are created at Task runtime if and only if the `cache_for` keyword
8 argument is provided to the Task, that specifies how long the output cache will be valid for
9 after its creation.  Cache validators come into play when a cached Task is re-run,
10 and are used to determine whether to re-run the Task or use the cache.
11 
12 Note that _all_ validators take into account cache expiration.
13 
14 A cache validator returns `True` if the cache is still valid, and `False` otherwise.
15 """
16 from typing import Any, Callable, Dict, Iterable
17 
18 import pendulum
19 
20 import prefect
21 
22 
23 def never_use(
24     state: "prefect.engine.state.Cached",
25     inputs: Dict[str, Any],
26     parameters: Dict[str, Any],
27 ) -> bool:
28     """
29     Never uses the cache.
30 
31     Args:
32         - state (State): a `Success` state from the last successful Task run that contains the cache
33         - inputs (dict): a `dict` of inputs that were available on the last
34             successful run of the cached Task
35         - parameters (dict): a `dict` of parameters that were available on the
36             last successful run of the cached Task
37 
38     Returns:
39         - boolean specifying whether or not the cache should be used
40     """
41     return False
42 
43 
44 def duration_only(
45     state: "prefect.engine.state.Cached",
46     inputs: Dict[str, Any],
47     parameters: Dict[str, Any],
48 ) -> bool:
49     """
50     Validates the cache based only on cache expiration.
51 
52     Args:
53         - state (State): a `Success` state from the last successful Task run that contains the cache
54         - inputs (dict): a `dict` of inputs that were available on the last
55             successful run of the cached Task
56         - parameters (dict): a `dict` of parameters that were available on the
57             last successful run of the cached Task
58 
59     Returns:
60         - boolean specifying whether or not the cache should be used
61     """
62     if state.cached_result_expiration is None:
63         return True
64     elif state.cached_result_expiration > pendulum.now("utc"):
65         return True
66     else:
67         return False
68 
69 
70 def all_inputs(
71     state: "prefect.engine.state.Cached",
72     inputs: Dict[str, Any],
73     parameters: Dict[str, Any],
74 ) -> bool:
75     """
76     Validates the cache based on cache expiration _and_ all inputs that were provided
77     on the last successful run.
78 
79     Args:
80         - state (State): a `Success` state from the last successful Task run that contains the cache
81         - inputs (dict): a `dict` of inputs that were available on the last
82             successful run of the cached Task
83         - parameters (dict): a `dict` of parameters that were available on the
84             last successful run of the cached Task
85 
86     Returns:
87         - boolean specifying whether or not the cache should be used
88     """
89     if duration_only(state, inputs, parameters) is False:
90         return False
91     elif {key: res.value for key, res in (state.cached_inputs or {}).items()} == inputs:
92         return True
93     else:
94         return False
95 
96 
97 def all_parameters(
98     state: "prefect.engine.state.Cached",
99     inputs: Dict[str, Any],
100     parameters: Dict[str, Any],
101 ) -> bool:
102     """
103     Validates the cache based on cache expiration _and_ all parameters that were provided
104     on the last successful run.
105 
106     Args:
107         - state (State): a `Success` state from the last successful Task run that contains the cache
108         - inputs (dict): a `dict` of inputs that were available on the last
109             successful run of the cached Task
110         - parameters (dict): a `dict` of parameters that were available on the
111             last successful run of the cached Task
112 
113     Returns:
114         - boolean specifying whether or not the cache should be used
115     """
116     if duration_only(state, inputs, parameters) is False:
117         return False
118     elif state.cached_parameters == parameters:
119         return True
120     else:
121         return False
122 
123 
124 def partial_parameters_only(validate_on: Iterable[str] = None,) -> Callable:
125     """
126     Validates the cache based on cache expiration _and_ a subset of parameters (determined by the
127     `validate_on` keyword) that were provided on the last successful run.
128 
129     Args:
130         - validate_on (list): a `list` of strings specifying the parameter names
131             to validate against
132 
133     Returns:
134         - Callable: the actual validation function specifying whether or not the cache should be used
135 
136     Example:
137     ```python
138     from datetime import timedelta
139     import pendulum
140     from prefect import Flow, Parameter, task
141     from prefect.engine.cache_validators import partial_parameters_only
142 
143     @task(cache_for=timedelta(days=1),
144           cache_validator=partial_parameters_only(validate_on=['nrows']))
145     def daily_db_refresh(nrows, runtime):
146         pass
147 
148     with Flow("My Flow") as f:
149         nrows = Parameter("nrows", default=500)
150         runtime = Parameter("runtime")
151         db_state = daily_db_refresh(nrows, runtime)
152 
153     state1 = f.run(parameters=dict(nrows=1000, runtime=pendulum.now('utc')))
154 
155     ## the second run will use the cache contained within state1.result[db_state]
156     ## even though `runtime` has changed
157     state2 = f.run(parameters=dict(nrows=1000, runtime=pendulum.now('utc')),
158                    task_states={result: state1.result[db_state]})
159     ```
160     """
161 
162     def _partial_parameters_only(
163         state: "prefect.engine.state.Cached",
164         inputs: Dict[str, Any],
165         parameters: Dict[str, Any],
166     ) -> bool:
167         """
168         The actual cache validation function that will be used.
169 
170         Args:
171             - state (State): a `Success` state from the last successful Task run that contains the cache
172             - inputs (dict): a `dict` of inputs that were available on the last
173                 successful run of the cached Task
174             - parameters (dict): a `dict` of parameters that were available on the
175                 last successful run of the cached Task
176 
177         Returns:
178             - boolean specifying whether or not the cache should be used
179         """
180         parameters = parameters or {}
181         if duration_only(state, inputs, parameters) is False:
182             return False
183         elif validate_on is None:
184             return (
185                 True
186             )  # if you dont want to validate on anything, then the cache is valid
187         else:
188             cached = state.cached_parameters or {}
189             partial_provided = {
190                 key: value for key, value in parameters.items() if key in validate_on
191             }
192             partial_needed = {
193                 key: value for key, value in cached.items() if key in validate_on
194             }
195             return partial_provided == partial_needed
196 
197     return _partial_parameters_only
198 
199 
200 def partial_inputs_only(validate_on: Iterable[str] = None,) -> Callable:
201     """
202     Validates the cache based on cache expiration _and_ a subset of inputs (determined by the
203     `validate_on` keyword) that were provided on the last successful run.
204 
205     Args:
206         - validate_on (list): a `list` of strings specifying the input names
207             to validate against
208 
209     Returns:
210         - Callable: the actual validation function specifying whether or not the cache should be used
211 
212     Example:
213     ```python
214     import random
215     from datetime import timedelta
216     from prefect import Flow, task
217     from prefect.engine.cache_validators import partial_inputs_only
218 
219     @task(cache_for=timedelta(days=1),
220           cache_validator=partial_inputs_only(validate_on=['x', 'y']))
221     def add(x, y, as_string=False):
222         if as_string:
223             return '{0} + {1}'.format(x, y)
224         return x + y
225 
226     @task
227     def rand_bool():
228         return random.random() > 0.5
229 
230     with Flow("My Flow") as f:
231         ans = add(1, 2, rand_bool())
232 
233     state1 = f.run()
234     ## the second run will use the cache contained within state1.result[ans]
235     ## even though `rand_bool` might change
236     state2 = f.run(task_states={result: state1.result[ans]})
237     ```
238     """
239 
240     def _partial_inputs_only(
241         state: "prefect.engine.state.Cached",
242         inputs: Dict[str, Any],
243         parameters: Dict[str, Any],
244     ) -> bool:
245         """
246         The actual cache validation function that will be used.
247 
248         Args:
249             - state (State): a `Success` state from the last successful Task run that contains the cache
250             - inputs (dict): a `dict` of inputs that were available on the last
251                 successful run of the cached Task
252             - parameters (dict): a `dict` of parameters that were available on the
253                 last successful run of the cached Task
254 
255         Returns:
256             - boolean specifying whether or not the cache should be used
257         """
258         inputs = inputs or {}
259         if duration_only(state, inputs, parameters) is False:
260             return False
261         elif validate_on is None:
262             return (
263                 True
264             )  # if you dont want to validate on anything, then the cache is valid
265         else:
266             cached = {
267                 key: res.value for key, res in (state.cached_inputs or {}).items()
268             }
269             partial_provided = {
270                 key: value for key, value in inputs.items() if key in validate_on
271             }
272             partial_needed = {
273                 key: value for key, value in cached.items() if key in validate_on
274             }
275             return partial_provided == partial_needed
276 
277     return _partial_inputs_only
278
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/prefect/engine/cache_validators.py b/src/prefect/engine/cache_validators.py
--- a/src/prefect/engine/cache_validators.py
+++ b/src/prefect/engine/cache_validators.py
@@ -181,9 +181,7 @@
         if duration_only(state, inputs, parameters) is False:
             return False
         elif validate_on is None:
-            return (
-                True
-            )  # if you dont want to validate on anything, then the cache is valid
+            return True  # if you dont want to validate on anything, then the cache is valid
         else:
             cached = state.cached_parameters or {}
             partial_provided = {
@@ -259,9 +257,7 @@
         if duration_only(state, inputs, parameters) is False:
             return False
         elif validate_on is None:
-            return (
-                True
-            )  # if you dont want to validate on anything, then the cache is valid
+            return True  # if you dont want to validate on anything, then the cache is valid
         else:
             cached = {
                 key: res.value for key, res in (state.cached_inputs or {}).items()

{"golden_diff": "diff --git a/src/prefect/engine/cache_validators.py b/src/prefect/engine/cache_validators.py\n--- a/src/prefect/engine/cache_validators.py\n+++ b/src/prefect/engine/cache_validators.py\n@@ -181,9 +181,7 @@\n         if duration_only(state, inputs, parameters) is False:\n             return False\n         elif validate_on is None:\n-            return (\n-                True\n-            )  # if you dont want to validate on anything, then the cache is valid\n+            return True  # if you dont want to validate on anything, then the cache is valid\n         else:\n             cached = state.cached_parameters or {}\n             partial_provided = {\n@@ -259,9 +257,7 @@\n         if duration_only(state, inputs, parameters) is False:\n             return False\n         elif validate_on is None:\n-            return (\n-                True\n-            )  # if you dont want to validate on anything, then the cache is valid\n+            return True  # if you dont want to validate on anything, then the cache is valid\n         else:\n             cached = {\n                 key: res.value for key, res in (state.cached_inputs or {}).items()\n", "issue": "Core docs language\nhttps://docs.prefect.io/core/\r\n\r\nIn a few places the Core docs mention that Cloud is coming soon, and I believe the language should be updated.\n", "before_files": [{"content": "\"\"\"\nCache validators are functions that determine if a task's output cache\nis still valid, or whether that task should be re-run; they are provided at\nTask creation via the `cache_validator` keyword argument (for more information\non instantiating Tasks see the [Task documentation](../core/task.html)).\n\nTask caches are created at Task runtime if and only if the `cache_for` keyword\nargument is provided to the Task, that specifies how long the output cache will be valid for\nafter its creation.  Cache validators come into play when a cached Task is re-run,\nand are used to determine whether to re-run the Task or use the cache.\n\nNote that _all_ validators take into account cache expiration.\n\nA cache validator returns `True` if the cache is still valid, and `False` otherwise.\n\"\"\"\nfrom typing import Any, Callable, Dict, Iterable\n\nimport pendulum\n\nimport prefect\n\n\ndef never_use(\n    state: \"prefect.engine.state.Cached\",\n    inputs: Dict[str, Any],\n    parameters: Dict[str, Any],\n) -> bool:\n    \"\"\"\n    Never uses the cache.\n\n    Args:\n        - state (State): a `Success` state from the last successful Task run that contains the cache\n        - inputs (dict): a `dict` of inputs that were available on the last\n            successful run of the cached Task\n        - parameters (dict): a `dict` of parameters that were available on the\n            last successful run of the cached Task\n\n    Returns:\n        - boolean specifying whether or not the cache should be used\n    \"\"\"\n    return False\n\n\ndef duration_only(\n    state: \"prefect.engine.state.Cached\",\n    inputs: Dict[str, Any],\n    parameters: Dict[str, Any],\n) -> bool:\n    \"\"\"\n    Validates the cache based only on cache expiration.\n\n    Args:\n        - state (State): a `Success` state from the last successful Task run that contains the cache\n        - inputs (dict): a `dict` of inputs that were available on the last\n            successful run of the cached Task\n        - parameters (dict): a `dict` of parameters that were available on the\n            last successful run of the cached Task\n\n    Returns:\n        - boolean specifying whether or not the cache should be used\n    \"\"\"\n    if state.cached_result_expiration is None:\n        return True\n    elif state.cached_result_expiration > pendulum.now(\"utc\"):\n        return True\n    else:\n        return False\n\n\ndef all_inputs(\n    state: \"prefect.engine.state.Cached\",\n    inputs: Dict[str, Any],\n    parameters: Dict[str, Any],\n) -> bool:\n    \"\"\"\n    Validates the cache based on cache expiration _and_ all inputs that were provided\n    on the last successful run.\n\n    Args:\n        - state (State): a `Success` state from the last successful Task run that contains the cache\n        - inputs (dict): a `dict` of inputs that were available on the last\n            successful run of the cached Task\n        - parameters (dict): a `dict` of parameters that were available on the\n            last successful run of the cached Task\n\n    Returns:\n        - boolean specifying whether or not the cache should be used\n    \"\"\"\n    if duration_only(state, inputs, parameters) is False:\n        return False\n    elif {key: res.value for key, res in (state.cached_inputs or {}).items()} == inputs:\n        return True\n    else:\n        return False\n\n\ndef all_parameters(\n    state: \"prefect.engine.state.Cached\",\n    inputs: Dict[str, Any],\n    parameters: Dict[str, Any],\n) -> bool:\n    \"\"\"\n    Validates the cache based on cache expiration _and_ all parameters that were provided\n    on the last successful run.\n\n    Args:\n        - state (State): a `Success` state from the last successful Task run that contains the cache\n        - inputs (dict): a `dict` of inputs that were available on the last\n            successful run of the cached Task\n        - parameters (dict): a `dict` of parameters that were available on the\n            last successful run of the cached Task\n\n    Returns:\n        - boolean specifying whether or not the cache should be used\n    \"\"\"\n    if duration_only(state, inputs, parameters) is False:\n        return False\n    elif state.cached_parameters == parameters:\n        return True\n    else:\n        return False\n\n\ndef partial_parameters_only(validate_on: Iterable[str] = None,) -> Callable:\n    \"\"\"\n    Validates the cache based on cache expiration _and_ a subset of parameters (determined by the\n    `validate_on` keyword) that were provided on the last successful run.\n\n    Args:\n        - validate_on (list): a `list` of strings specifying the parameter names\n            to validate against\n\n    Returns:\n        - Callable: the actual validation function specifying whether or not the cache should be used\n\n    Example:\n    ```python\n    from datetime import timedelta\n    import pendulum\n    from prefect import Flow, Parameter, task\n    from prefect.engine.cache_validators import partial_parameters_only\n\n    @task(cache_for=timedelta(days=1),\n          cache_validator=partial_parameters_only(validate_on=['nrows']))\n    def daily_db_refresh(nrows, runtime):\n        pass\n\n    with Flow(\"My Flow\") as f:\n        nrows = Parameter(\"nrows\", default=500)\n        runtime = Parameter(\"runtime\")\n        db_state = daily_db_refresh(nrows, runtime)\n\n    state1 = f.run(parameters=dict(nrows=1000, runtime=pendulum.now('utc')))\n\n    ## the second run will use the cache contained within state1.result[db_state]\n    ## even though `runtime` has changed\n    state2 = f.run(parameters=dict(nrows=1000, runtime=pendulum.now('utc')),\n                   task_states={result: state1.result[db_state]})\n    ```\n    \"\"\"\n\n    def _partial_parameters_only(\n        state: \"prefect.engine.state.Cached\",\n        inputs: Dict[str, Any],\n        parameters: Dict[str, Any],\n    ) -> bool:\n        \"\"\"\n        The actual cache validation function that will be used.\n\n        Args:\n            - state (State): a `Success` state from the last successful Task run that contains the cache\n            - inputs (dict): a `dict` of inputs that were available on the last\n                successful run of the cached Task\n            - parameters (dict): a `dict` of parameters that were available on the\n                last successful run of the cached Task\n\n        Returns:\n            - boolean specifying whether or not the cache should be used\n        \"\"\"\n        parameters = parameters or {}\n        if duration_only(state, inputs, parameters) is False:\n            return False\n        elif validate_on is None:\n            return (\n                True\n            )  # if you dont want to validate on anything, then the cache is valid\n        else:\n            cached = state.cached_parameters or {}\n            partial_provided = {\n                key: value for key, value in parameters.items() if key in validate_on\n            }\n            partial_needed = {\n                key: value for key, value in cached.items() if key in validate_on\n            }\n            return partial_provided == partial_needed\n\n    return _partial_parameters_only\n\n\ndef partial_inputs_only(validate_on: Iterable[str] = None,) -> Callable:\n    \"\"\"\n    Validates the cache based on cache expiration _and_ a subset of inputs (determined by the\n    `validate_on` keyword) that were provided on the last successful run.\n\n    Args:\n        - validate_on (list): a `list` of strings specifying the input names\n            to validate against\n\n    Returns:\n        - Callable: the actual validation function specifying whether or not the cache should be used\n\n    Example:\n    ```python\n    import random\n    from datetime import timedelta\n    from prefect import Flow, task\n    from prefect.engine.cache_validators import partial_inputs_only\n\n    @task(cache_for=timedelta(days=1),\n          cache_validator=partial_inputs_only(validate_on=['x', 'y']))\n    def add(x, y, as_string=False):\n        if as_string:\n            return '{0} + {1}'.format(x, y)\n        return x + y\n\n    @task\n    def rand_bool():\n        return random.random() > 0.5\n\n    with Flow(\"My Flow\") as f:\n        ans = add(1, 2, rand_bool())\n\n    state1 = f.run()\n    ## the second run will use the cache contained within state1.result[ans]\n    ## even though `rand_bool` might change\n    state2 = f.run(task_states={result: state1.result[ans]})\n    ```\n    \"\"\"\n\n    def _partial_inputs_only(\n        state: \"prefect.engine.state.Cached\",\n        inputs: Dict[str, Any],\n        parameters: Dict[str, Any],\n    ) -> bool:\n        \"\"\"\n        The actual cache validation function that will be used.\n\n        Args:\n            - state (State): a `Success` state from the last successful Task run that contains the cache\n            - inputs (dict): a `dict` of inputs that were available on the last\n                successful run of the cached Task\n            - parameters (dict): a `dict` of parameters that were available on the\n                last successful run of the cached Task\n\n        Returns:\n            - boolean specifying whether or not the cache should be used\n        \"\"\"\n        inputs = inputs or {}\n        if duration_only(state, inputs, parameters) is False:\n            return False\n        elif validate_on is None:\n            return (\n                True\n            )  # if you dont want to validate on anything, then the cache is valid\n        else:\n            cached = {\n                key: res.value for key, res in (state.cached_inputs or {}).items()\n            }\n            partial_provided = {\n                key: value for key, value in inputs.items() if key in validate_on\n            }\n            partial_needed = {\n                key: value for key, value in cached.items() if key in validate_on\n            }\n            return partial_provided == partial_needed\n\n    return _partial_inputs_only\n", "path": "src/prefect/engine/cache_validators.py"}], "after_files": [{"content": "\"\"\"\nCache validators are functions that determine if a task's output cache\nis still valid, or whether that task should be re-run; they are provided at\nTask creation via the `cache_validator` keyword argument (for more information\non instantiating Tasks see the [Task documentation](../core/task.html)).\n\nTask caches are created at Task runtime if and only if the `cache_for` keyword\nargument is provided to the Task, that specifies how long the output cache will be valid for\nafter its creation.  Cache validators come into play when a cached Task is re-run,\nand are used to determine whether to re-run the Task or use the cache.\n\nNote that _all_ validators take into account cache expiration.\n\nA cache validator returns `True` if the cache is still valid, and `False` otherwise.\n\"\"\"\nfrom typing import Any, Callable, Dict, Iterable\n\nimport pendulum\n\nimport prefect\n\n\ndef never_use(\n    state: \"prefect.engine.state.Cached\",\n    inputs: Dict[str, Any],\n    parameters: Dict[str, Any],\n) -> bool:\n    \"\"\"\n    Never uses the cache.\n\n    Args:\n        - state (State): a `Success` state from the last successful Task run that contains the cache\n        - inputs (dict): a `dict` of inputs that were available on the last\n            successful run of the cached Task\n        - parameters (dict): a `dict` of parameters that were available on the\n            last successful run of the cached Task\n\n    Returns:\n        - boolean specifying whether or not the cache should be used\n    \"\"\"\n    return False\n\n\ndef duration_only(\n    state: \"prefect.engine.state.Cached\",\n    inputs: Dict[str, Any],\n    parameters: Dict[str, Any],\n) -> bool:\n    \"\"\"\n    Validates the cache based only on cache expiration.\n\n    Args:\n        - state (State): a `Success` state from the last successful Task run that contains the cache\n        - inputs (dict): a `dict` of inputs that were available on the last\n            successful run of the cached Task\n        - parameters (dict): a `dict` of parameters that were available on the\n            last successful run of the cached Task\n\n    Returns:\n        - boolean specifying whether or not the cache should be used\n    \"\"\"\n    if state.cached_result_expiration is None:\n        return True\n    elif state.cached_result_expiration > pendulum.now(\"utc\"):\n        return True\n    else:\n        return False\n\n\ndef all_inputs(\n    state: \"prefect.engine.state.Cached\",\n    inputs: Dict[str, Any],\n    parameters: Dict[str, Any],\n) -> bool:\n    \"\"\"\n    Validates the cache based on cache expiration _and_ all inputs that were provided\n    on the last successful run.\n\n    Args:\n        - state (State): a `Success` state from the last successful Task run that contains the cache\n        - inputs (dict): a `dict` of inputs that were available on the last\n            successful run of the cached Task\n        - parameters (dict): a `dict` of parameters that were available on the\n            last successful run of the cached Task\n\n    Returns:\n        - boolean specifying whether or not the cache should be used\n    \"\"\"\n    if duration_only(state, inputs, parameters) is False:\n        return False\n    elif {key: res.value for key, res in (state.cached_inputs or {}).items()} == inputs:\n        return True\n    else:\n        return False\n\n\ndef all_parameters(\n    state: \"prefect.engine.state.Cached\",\n    inputs: Dict[str, Any],\n    parameters: Dict[str, Any],\n) -> bool:\n    \"\"\"\n    Validates the cache based on cache expiration _and_ all parameters that were provided\n    on the last successful run.\n\n    Args:\n        - state (State): a `Success` state from the last successful Task run that contains the cache\n        - inputs (dict): a `dict` of inputs that were available on the last\n            successful run of the cached Task\n        - parameters (dict): a `dict` of parameters that were available on the\n            last successful run of the cached Task\n\n    Returns:\n        - boolean specifying whether or not the cache should be used\n    \"\"\"\n    if duration_only(state, inputs, parameters) is False:\n        return False\n    elif state.cached_parameters == parameters:\n        return True\n    else:\n        return False\n\n\ndef partial_parameters_only(validate_on: Iterable[str] = None,) -> Callable:\n    \"\"\"\n    Validates the cache based on cache expiration _and_ a subset of parameters (determined by the\n    `validate_on` keyword) that were provided on the last successful run.\n\n    Args:\n        - validate_on (list): a `list` of strings specifying the parameter names\n            to validate against\n\n    Returns:\n        - Callable: the actual validation function specifying whether or not the cache should be used\n\n    Example:\n    ```python\n    from datetime import timedelta\n    import pendulum\n    from prefect import Flow, Parameter, task\n    from prefect.engine.cache_validators import partial_parameters_only\n\n    @task(cache_for=timedelta(days=1),\n          cache_validator=partial_parameters_only(validate_on=['nrows']))\n    def daily_db_refresh(nrows, runtime):\n        pass\n\n    with Flow(\"My Flow\") as f:\n        nrows = Parameter(\"nrows\", default=500)\n        runtime = Parameter(\"runtime\")\n        db_state = daily_db_refresh(nrows, runtime)\n\n    state1 = f.run(parameters=dict(nrows=1000, runtime=pendulum.now('utc')))\n\n    ## the second run will use the cache contained within state1.result[db_state]\n    ## even though `runtime` has changed\n    state2 = f.run(parameters=dict(nrows=1000, runtime=pendulum.now('utc')),\n                   task_states={result: state1.result[db_state]})\n    ```\n    \"\"\"\n\n    def _partial_parameters_only(\n        state: \"prefect.engine.state.Cached\",\n        inputs: Dict[str, Any],\n        parameters: Dict[str, Any],\n    ) -> bool:\n        \"\"\"\n        The actual cache validation function that will be used.\n\n        Args:\n            - state (State): a `Success` state from the last successful Task run that contains the cache\n            - inputs (dict): a `dict` of inputs that were available on the last\n                successful run of the cached Task\n            - parameters (dict): a `dict` of parameters that were available on the\n                last successful run of the cached Task\n\n        Returns:\n            - boolean specifying whether or not the cache should be used\n        \"\"\"\n        parameters = parameters or {}\n        if duration_only(state, inputs, parameters) is False:\n            return False\n        elif validate_on is None:\n            return True  # if you dont want to validate on anything, then the cache is valid\n        else:\n            cached = state.cached_parameters or {}\n            partial_provided = {\n                key: value for key, value in parameters.items() if key in validate_on\n            }\n            partial_needed = {\n                key: value for key, value in cached.items() if key in validate_on\n            }\n            return partial_provided == partial_needed\n\n    return _partial_parameters_only\n\n\ndef partial_inputs_only(validate_on: Iterable[str] = None,) -> Callable:\n    \"\"\"\n    Validates the cache based on cache expiration _and_ a subset of inputs (determined by the\n    `validate_on` keyword) that were provided on the last successful run.\n\n    Args:\n        - validate_on (list): a `list` of strings specifying the input names\n            to validate against\n\n    Returns:\n        - Callable: the actual validation function specifying whether or not the cache should be used\n\n    Example:\n    ```python\n    import random\n    from datetime import timedelta\n    from prefect import Flow, task\n    from prefect.engine.cache_validators import partial_inputs_only\n\n    @task(cache_for=timedelta(days=1),\n          cache_validator=partial_inputs_only(validate_on=['x', 'y']))\n    def add(x, y, as_string=False):\n        if as_string:\n            return '{0} + {1}'.format(x, y)\n        return x + y\n\n    @task\n    def rand_bool():\n        return random.random() > 0.5\n\n    with Flow(\"My Flow\") as f:\n        ans = add(1, 2, rand_bool())\n\n    state1 = f.run()\n    ## the second run will use the cache contained within state1.result[ans]\n    ## even though `rand_bool` might change\n    state2 = f.run(task_states={result: state1.result[ans]})\n    ```\n    \"\"\"\n\n    def _partial_inputs_only(\n        state: \"prefect.engine.state.Cached\",\n        inputs: Dict[str, Any],\n        parameters: Dict[str, Any],\n    ) -> bool:\n        \"\"\"\n        The actual cache validation function that will be used.\n\n        Args:\n            - state (State): a `Success` state from the last successful Task run that contains the cache\n            - inputs (dict): a `dict` of inputs that were available on the last\n                successful run of the cached Task\n            - parameters (dict): a `dict` of parameters that were available on the\n                last successful run of the cached Task\n\n        Returns:\n            - boolean specifying whether or not the cache should be used\n        \"\"\"\n        inputs = inputs or {}\n        if duration_only(state, inputs, parameters) is False:\n            return False\n        elif validate_on is None:\n            return True  # if you dont want to validate on anything, then the cache is valid\n        else:\n            cached = {\n                key: res.value for key, res in (state.cached_inputs or {}).items()\n            }\n            partial_provided = {\n                key: value for key, value in inputs.items() if key in validate_on\n            }\n            partial_needed = {\n                key: value for key, value in cached.items() if key in validate_on\n            }\n            return partial_provided == partial_needed\n\n    return _partial_inputs_only\n", "path": "src/prefect/engine/cache_validators.py"}]}