AsyncTerraformCommand¶

Import AsyncTerraformCommand from the package root:

from libterraform import AsyncTerraformCommand

AsyncTerraformCommand provides asyncio-compatible access to TerraformCommand. It mirrors the synchronous command methods and runs the blocking Terraform call away from the event-loop thread, so callers can await Terraform operations without blocking the event loop.

Execution Model¶

By default AsyncTerraformCommand uses TerraformCommand's process backend. The Terraform CLI call runs in a controlled worker process, so Terraform's process-wide state does not leak into the event-loop process. Use backend="thread" only when you explicitly want the current-process backend. Pass a TerraformPool as pool when you want to reuse worker processes or run independent Terraform operations in parallel.

If a coroutine is cancelled, the awaiting task is cancelled and the active backend is asked to stop the Terraform run. With the default process backend the worker process is interrupted. With backend="thread", the worker thread is not terminated directly; AsyncTerraformCommand sends a cooperative cancellation request to Terraform's shutdown channel and then re-raises asyncio.CancelledError. Terraform or a provider may still take some time to return from its own shutdown path.

Usage¶

from libterraform import AsyncTerraformCommand

cli = AsyncTerraformCommand("path/to/terraform/module")

await cli.init(check=True)
plan = await cli.plan(check=True)

AsyncTerraformCommand.run() accepts the same command arguments as TerraformCommand.run():

retcode, stdout, stderr = await AsyncTerraformCommand.run("version")

Pass an executor when you need to integrate the awaitable wrapper with an application-owned thread pool. This controls where the blocking Python wrapper waits; it does not switch Terraform execution to the thread backend:

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=1) as executor:
    cli = AsyncTerraformCommand("path/to/terraform/module", executor=executor)
    validation = await cli.validate(check=True)

To opt in to the current-process backend, pass backend="thread":

cli = AsyncTerraformCommand("path/to/terraform/module", backend="thread")
validation = await cli.validate(check=True)

Pass a TerraformPool as pool to await commands that run in worker processes, giving true parallel Terraform execution. AsyncTerraformCommand.run() accepts pool as well. A pool starts worker processes, so this program must run under an if __name__ == "__main__": guard; see Parallel Execution for a complete, runnable setup.

from libterraform import AsyncTerraformCommand, TerraformPool

with TerraformPool(max_workers=4) as pool:
    network = AsyncTerraformCommand("modules/network", pool=pool)
    app = AsyncTerraformCommand("modules/app", pool=pool)
    results = await asyncio.gather(
        network.apply(auto_approve=True),
        app.apply(auto_approve=True),
    )

Cancellation requests are scoped to the Terraform run started by the coroutine:

task = asyncio.create_task(cli.apply(auto_approve=True))
task.cancel()

This asks Terraform to stop through the active backend. With the default process backend the worker process is interrupted; with backend="thread" it is not a direct termination of the worker thread; with a pool backend the request is delivered to the worker process running the command.

Every public TerraformCommand method is mirrored as an awaitable coroutine (await async_cli.plan(...)). The methods documented below are defined directly on AsyncTerraformCommand.

run(
    cmd: CmdType,
    args: Optional[Sequence[str]] = None,
    options: Optional[dict] = None,
    chdir=None,
    check: bool = False,
    json=False,
    executor: Optional[Executor] = None,
    pool: Optional[TerraformPool] = None,
    backend: str = "process",
) -> Tuple[int, str, str]

stream(
    cmd: CmdType,
    args: Optional[Sequence[str]] = None,
    options: Optional[dict] = None,
    chdir=None,
    json: bool = True,
    check: bool = False,
)

plan_stream(
    json: bool = True, check: bool = False, **options
)

apply_stream(
    json: bool = True,
    check: bool = False,
    auto_approve: bool = True,
    input: bool = False,
    **options,
)