Implement set of "runtime APIs"

The file at `tests/test_cli/deploy/runtime_apis.py` explains these APIs
and when they should be used. Due to the way pyinfra executes deploy code
these constructs are needed in certain cases to give results as expected
by checking facts/op changes at runtime.

pyinfra/api/arguments.py

@@ -169,7 +169,7 @@ class MetaArguments(TypedDict):
     name: str
     _ignore_errors: bool
     _continue_on_error: bool
-    _if: Callable[[], bool]
+    _if: list[Callable[[], bool]]
 
 
 meta_argument_meta: dict[str, ArgumentMeta] = {
@@ -190,8 +190,8 @@ class MetaArguments(TypedDict):
         default=lambda _: False,
     ),
     "_if": ArgumentMeta(
-        "Only run this operation if this function returns True",
-        default=lambda _: None,
+        "Only run this operation if these functions returns True",
+        default=lambda _: [],
     ),
 }
 
@@ -299,7 +299,7 @@ def pop_global_arguments(
     if context.ctx_config.isset():
         config = context.config
 
-    meta_kwargs = host.current_deploy_kwargs or {}
+    meta_kwargs: dict[str, Any] = cast(dict[str, Any], host.current_deploy_kwargs) or {}
 
     arguments: dict[str, Any] = {}
     found_keys: list[str] = []

pyinfra/api/arguments_typed.py

@@ -51,7 +51,7 @@ def __call__(
         name: Optional[str] = None,
         _ignore_errors: bool = False,
         _continue_on_error: bool = False,
-        _if: Optional[Callable[[], bool]] = None,
+        _if: Optional[list[Callable[[], bool]]] = None,
         #
         # ExecutionArguments
         #

pyinfra/api/host.py

@@ -1,7 +1,19 @@
 from __future__ import annotations
 
 from contextlib import contextmanager
-from typing import TYPE_CHECKING, Any, Callable, Generator, Optional, Type, TypeVar, Union, overload
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Generator,
+    Optional,
+    Type,
+    TypeVar,
+    Union,
+    Unpack,
+    cast,
+    overload,
+)
 from uuid import uuid4
 
 import click
@@ -216,8 +228,25 @@ def noop(self, description):
         handler = logger.info if self.state.print_noop_info else logger.debug
         handler("{0}noop: {1}".format(self.print_prefix, description))
 
+    def when(self, condition: Callable[[], bool]):
+        return self.deploy(
+            "",
+            cast("AllArguments", {"_if": [condition]}),
+            {},
+            in_deploy=False,
+        )
+
+    def arguments(self, **arguments: Unpack["AllArguments"]):
+        return self.deploy("", arguments, {}, in_deploy=False)
+
     @contextmanager
-    def deploy(self, name: str, kwargs, data, in_deploy: bool = True):
+    def deploy(
+        self,
+        name: str,
+        kwargs: Optional["AllArguments"],
+        data: Optional[dict],
+        in_deploy: bool = True,
+    ):
         """
         Wraps a group of operations as a deploy, this should not be used
         directly, instead use ``pyinfra.api.deploy.deploy``.
@@ -234,6 +263,13 @@ def deploy(self, name: str, kwargs, data, in_deploy: bool = True):
         old_deploy_data = self.current_deploy_data
         self.in_deploy = in_deploy
 
+        # Combine any old _ifs with the new ones
+        if old_deploy_kwargs and kwargs:
+            old_ifs = old_deploy_kwargs["_if"]
+            new_ifs = kwargs["_if"]
+            if old_ifs and new_ifs:
+                kwargs["_if"] = old_ifs + new_ifs
+
         # Set the new values
         self.current_deploy_name = name
         self.current_deploy_kwargs = kwargs

pyinfra/api/operation.py

@@ -42,10 +42,10 @@ class OperationMeta:
 
     _combined_output_lines = None
     _commands: Optional[list[Any]] = None
-    _maybe_is_change: bool = False
+    _maybe_is_change: Optional[bool] = False
     _success: Optional[bool] = None
 
-    def __init__(self, hash, is_change=False):
+    def __init__(self, hash, is_change: Optional[bool]):
         self._hash = hash
         self._maybe_is_change = is_change
 
@@ -57,7 +57,7 @@ def __repr__(self) -> str:
         if self._commands is not None:
             return (
                 "OperationMeta(executed=True, "
-                f"changed={self.did_change()}, hash={self._hash}, commands={len(self._commands)})"
+                f"success={self.did_succeed}, hash={self._hash}, commands={len(self._commands)})"
             )
         return (
             "OperationMeta(executed=False, "
@@ -84,10 +84,17 @@ def _raise_if_not_complete(self) -> None:
         if not self.is_complete():
             raise RuntimeError("Cannot evaluate operation result before execution")
 
-    def did_change(self) -> bool:
-        self._raise_if_not_complete()
+    def _did_change(self) -> bool:
         return bool(self._success and len(self._commands or []) > 0)
 
+    @property
+    def did_change(self):
+        return context.host.when(self._did_change)
+
+    @property
+    def did_not_change(self):
+        return context.host.when(lambda: not self._did_change())
+
     def did_succeed(self) -> bool:
         self._raise_if_not_complete()
         return self._success is True
@@ -96,6 +103,21 @@ def did_error(self) -> bool:
         self._raise_if_not_complete()
         return self._success is False
 
+    # TODO: deprecated, remove in v4
+    @property
+    def changed(self) -> bool:
+        if self.is_complete():
+            return self._did_change()
+
+        if self._maybe_is_change is not None:
+            return self._maybe_is_change
+
+        op_data = context.state.get_op_data_for_host(context.host, self._hash)
+        cmd_gen = op_data.command_generator
+        for _ in cmd_gen():
+            return True
+        return False
+
     # Output lines
     def _get_lines(self, types=("stdout", "stderr")):
         self._raise_if_not_complete()
@@ -118,14 +140,6 @@ def stdout(self) -> str:
     def stderr(self) -> str:
         return "\n".join(self.stderr_lines)
 
-    # TODO: deprecated, remove in v4
-    @property
-    def changed(self) -> int:
-        if not self.is_complete():
-            logger.warning("Checking changed before execution can give unexpected results")
-            return self._maybe_is_change
-        return self.did_change()
-
 
 def add_op(state: State, op_func, *args, **kwargs):
     """
@@ -230,7 +244,7 @@ def decorated_func(*args: P.args, **kwargs: P.kwargs) -> OperationMeta:
                     break
 
             if has_run:
-                return OperationMeta(op_hash)
+                return OperationMeta(op_hash, is_change=False)
 
         # "Run" operation - here we make a generator that will yield out actual commands to execute
         # and, if we're diff-ing, we then iterate the generator now to determine if any changes
@@ -239,8 +253,8 @@ def decorated_func(*args: P.args, **kwargs: P.kwargs) -> OperationMeta:
         def command_generator() -> Iterator[PyinfraCommand]:
             # Check global _if_ argument function and do nothing if returns False
             if state.is_executing:
-                _if = global_arguments.get("_if")
-                if _if and _if() is False:
+                _ifs = global_arguments.get("_if")
+                if _ifs and not all(_if() for _if in _ifs):
                     return
 
             host.in_op = _set_in_op
@@ -257,9 +271,10 @@ def command_generator() -> Iterator[PyinfraCommand]:
                 host.current_op_hash = None
                 host.current_op_global_arguments = None
 
-        op_is_change = False
+        op_is_change = None
         if state.should_check_for_changes():
-            for command in command_generator():
+            op_is_change = False
+            for _ in command_generator():
                 op_is_change = True
                 break
         else:

pyinfra/api/state.py

@@ -344,10 +344,19 @@ def get_meta_for_host(self, host: "Host") -> StateHostMeta:
     def get_results_for_host(self, host: "Host") -> StateHostResults:
         return self.results[host]
 
-    def get_op_data_for_host(self, host: "Host", op_hash: str):
+    def get_op_data_for_host(
+        self,
+        host: "Host",
+        op_hash: str,
+    ) -> StateOperationHostData:
         return self.ops[host][op_hash]
 
-    def set_op_data_for_host(self, host: "Host", op_hash: str, op_data: StateOperationHostData):
+    def set_op_data_for_host(
+        self,
+        host: "Host",
+        op_hash: str,
+        op_data: StateOperationHostData,
+    ):
         self.ops[host][op_hash] = op_data
 
     def activate_host(self, host: "Host"):

tests/test_cli/deploy/runtime_apis.py

@@ -0,0 +1,90 @@
+from pyinfra import host
+from pyinfra.facts.files import File, Sha256File
+from pyinfra.operations import files, python, server
+
+TEST_FILENAME = "/opt/testfile"
+TEST_LINE = "test line"
+
+
+# Facts
+# Facts are gathered during the operation run which happens after the deploy
+# code is completely executed, so facts altered by operations will have the
+# state prior to any operations.
+
+files.line(
+    path=TEST_FILENAME,
+    line=TEST_LINE,
+)
+
+# Instead of:
+if host.get_fact(File, path=TEST_FILENAME):
+    # This will only be executed if the file existed on the remote machine before
+    # the deploy started.
+    server.shell(commands=["echo if sees file"])
+
+# Use the host.when context to add operations that will run if a condition,
+# evaluated at runtime, is met.
+with host.when(lambda: bool(host.get_fact(File, path=TEST_FILENAME))):
+    # This will be executed if the file exists on the remote machine at this point
+    # in the deploy.
+    server.shell(commands=["echo when sees file"])
+
+
+# This fact will be None because the file doesn't exist on the remote machine
+# before the deploy starts.
+
+# Instead of:
+sha256 = host.get_fact(Sha256File, path=TEST_FILENAME)
+print("Wrong sha256:", sha256)
+
+
+# Use a callback function to get the updated fact data
+def check_host_roles():
+    # This will have the correct SHA of the file as callback functions are
+    # evaluated during the deploy.
+    sha256 = host.get_fact(Sha256File, path=TEST_FILENAME)
+    print("Correct sha256:", sha256)
+
+    # Call nested operations here using sha256..
+
+
+python.call(function=check_host_roles)
+
+
+# Operation changes
+# Operations are run after deploy code are completely executed, so operations
+# that overlap making changes to the remote system will both report they will
+# make changes.
+
+# For demonstration we just duplicate the first line op so it's always no change
+second_line_op = files.line(
+    path=TEST_FILENAME,
+    line=TEST_LINE,
+)
+
+# Instead of:
+if second_line_op.changed:
+    # This will be executed if the line/file didn't exist on the remote machine
+    # before the deploy started.
+    server.shell(commands=["echo if sees second op"])
+
+# Use the op.did_change context
+with second_line_op.did_change:
+    # This will be executed if the second line operation made any changes, in
+    # this deploy that's never. This will appear as a conditional op in the CLI.
+    server.shell(commands=["echo with sees second op"])
+
+
+# Nested
+
+with host.when(lambda: False):
+    with host.when(lambda: True):
+        # This should never be executed
+        server.shell(commands=["echo this message should not be printed"])
+    server.shell(commands=["echo this message should not be printed"])
+
+with host.when(lambda: True):
+    with host.when(lambda: False):
+        # This should never be executed
+        server.shell(commands=["echo this message should not be printed"])
+    server.shell(commands=["echo this message should be printed"])