Add output command to Terraform.

Author: tobyclemson

src/infrablocks/invoke_terraform/invoke_executor.py

@@ -1,16 +1,27 @@
-from typing import Iterable
+from typing import IO, Iterable
 
 from invoke.context import Context
 
 import infrablocks.invoke_terraform.terraform as tf
-from infrablocks.invoke_terraform.terraform.terraform import Environment
+from infrablocks.invoke_terraform.terraform.terraform import (
+    Environment,
+)
 
 
 class InvokeExecutor(tf.Executor):
     def __init__(self, context: Context):
         self._context = context
 
     def execute(
-        self, command: Iterable[str], env: Environment | None = None
+        self,
+        command: Iterable[str],
+        environment: Environment | None = None,
+        stdout: IO[str] | None = None,
+        stderr: IO[str] | None = None,
     ) -> None:
-        self._context.run(" ".join(command), env=(env or {}))
+        self._context.run(
+            " ".join(command),
+            env=(environment if environment is not None else {}),
+            out_stream=stdout,
+            err_stream=stderr,
+        )

src/infrablocks/invoke_terraform/terraform/__init__.py

@@ -1,10 +1,12 @@
 from .terraform import BackendConfig, ConfigurationValue, Variables
+from .terraform import Environment as Environment
 from .terraform import Executor as Executor
 from .terraform import Terraform as Terraform
 
 __all__ = [
     "BackendConfig",
     "ConfigurationValue",
+    "Environment",
     "Executor",
     "Terraform",
     "Variables",

src/infrablocks/invoke_terraform/terraform/terraform.py

@@ -1,5 +1,7 @@
 import json
 from collections.abc import Mapping, Sequence
+from tempfile import TemporaryFile
+from typing import IO, Literal
 
 type ConfigurationValue = (
     bool
@@ -13,11 +15,37 @@
 type Variables = Mapping[str, ConfigurationValue]
 type BackendConfig = str | Mapping[str, ConfigurationValue]
 type Environment = Mapping[str, str]
+type StreamName = Literal["stdout"] | Literal["stderr"]
+type StreamNames = set[StreamName]
+
+
+class Result:
+    def __init__(self, stdout: IO[str] | None, stderr: IO[str] | None):
+        self.stdout = stdout
+        self.stderr = stderr
 
 
 class Executor:
-    def execute(self, command: Sequence[str], env: Environment | None) -> None:
-        raise Exception("NotImplementedException")
+    def execute(
+        self,
+        command: Sequence[str],
+        environment: Environment | None = None,
+        stdout: IO[str] | None = None,
+        stderr: IO[str] | None = None,
+    ) -> None:
+        raise NotImplementedError
+
+
+def _captures(capture: StreamNames | None, stream: StreamName) -> bool:
+    return capture is not None and stream in capture
+
+
+def _capture_stream(
+    capture: StreamNames | None, stream: StreamName
+) -> IO[str] | None:
+    if _captures(capture, stream):
+        return TemporaryFile(mode="w+t")
+    return None
 
 
 class Terraform:
@@ -41,7 +69,7 @@ def init(
         if reconfigure:
             command = command + ["-reconfigure"]
 
-        self._executor.execute(command, env=environment)
+        self._executor.execute(command, environment=environment)
 
     def plan(
         self,
@@ -52,7 +80,7 @@ def plan(
         base_command = self._build_base_command(chdir)
         command = base_command + ["plan"] + self._build_vars(vars)
 
-        self._executor.execute(command, env=environment)
+        self._executor.execute(command, environment=environment)
 
     def apply(
         self,
@@ -70,7 +98,7 @@ def apply(
             + self._build_vars(vars)
         )
 
-        self._executor.execute(command, env=environment)
+        self._executor.execute(command, environment=environment)
 
     def select_workspace(
         self,
@@ -87,7 +115,43 @@ def select_workspace(
 
         command = command + [workspace]
 
-        self._executor.execute(command, env=environment)
+        self._executor.execute(command, environment=environment)
+
+    def output(
+        self,
+        chdir: str | None = None,
+        name: str | None = None,
+        raw: bool = False,
+        json: bool = False,
+        environment: Environment | None = None,
+        capture: StreamNames | None = None,
+    ) -> Result:
+        base_command = self._build_base_command(chdir)
+        command = base_command + ["output"]
+
+        if raw:
+            command = command + ["-raw"]
+        if json:
+            command = command + ["-json"]
+        if name is not None:
+            command = command + [name]
+
+        stdout = _capture_stream(capture, "stdout")
+        stderr = _capture_stream(capture, "stderr")
+
+        self._executor.execute(
+            command,
+            environment=environment,
+            stdout=stdout,
+            stderr=stderr,
+        )
+
+        if stdout is not None:
+            stdout.seek(0)
+        if stderr is not None:
+            stderr.seek(0)
+
+        return Result(stdout, stderr)
 
     @staticmethod
     def _build_base_command(chdir: str | None) -> list[str]: