API reference

Generated from the source docstrings via mkdocstrings.

Composition

provisio.plan.Plan dataclass

An explicit, ordered sequence of steps.

validate

validate()

Statically check that every requires is produced by an earlier step.

Ignores skip (that is a runtime concern); this catches a malformed plan before any command runs.

preview

preview(skip=())

Return the ordered preview lines, marking skipped steps.

dry_run

dry_run(ctx, *, skip=())

Preview the plan without executing anything.

Reports each step to ctx.reporter (the human preview) and writes an auditable record to the audit log — dry-run plus would-run / would-skip per step — so a preview leaves a compliance trace just like a real apply. No commands run and no credentials are needed.

fingerprint

fingerprint()

A hash over the ordered step fingerprints (the plan's structure).

dependency_edges

dependency_edges()

(producer_key, consumer_key) edges via OutputKeys — the DAG data.

execute

execute(ctx, *, skip=())

Run the plan against ctx and return a RunResult.

Raises MissingOutputError if a running step needs an output that a skipped step would have produced, StepFailedError for unexpected step errors, or the original ProvisioError for domain failures.

provisio.plan.step

step(key, title, *, requires=(), produces=())

Decorator turning an action(ctx) function into a FunctionStep.

requires/produces declare the output data-flow: they are validated up front by Plan and are the source of the (deferred) dependency graph.

provisio.plan.command_step

command_step(key, title, tool, *args)

Wrap a single CLI command as a step — the L1 adoption on-ramp.

No idempotency: it just runs tool args (audited, previewable, ordered). Add an exists= probe via ensure later to reach L2.

provisio.plan.RunResult dataclass

The programmatic result of executing a plan (the library entry point).

provisio.context.ExecutionContext dataclass

Collaborators + produced outputs, handed to every step.

Defaults are deliberately sensible so the minimal path stays short (see the simplicity budget): a silent NullReporter and an empty tool registry.

tool

tool(name)

Return the registered CLI tool, or raise a clear error if absent.

set

set(key, value)

Record the value produced for key.

get

get(key)

Return the value produced for key, or raise MissingOutputError.

has

has(key)

True if a value has been produced for key.

snapshot

snapshot()

A copy of the produced outputs, keyed by name (for RunResult).

provisio.context.OutputKey dataclass

A typed handle for a value one step produces and another consumes.

Declare each key once (e.g. SWA_URL: OutputKey[str] = OutputKey("swa_url")) and pass the key — not a bare string — to ctx.set/ctx.get. The type parameter flows through get so consumers get the right static type.

Mark secret=True for sensitive values (e.g. deploy tokens): they are hashed, never stored raw, when a run is persisted to state.

Primitives

provisio.primitives.ensure

ensure(ctx, *, describe, exists, create)

Create a resource only if it is not already there (idempotent).

Parameters:

Name	Type	Description	Default
ctx	ExecutionContext	the execution context (for reporting).	required
describe	str	a human phrase naming the resource, e.g. "resource group 'rg'".	required
exists	Callable[[], bool]	a probe returning True when the resource already exists. Must not raise on "absent" — use CliTool.exists(...), which probes with check=False.	required
create	Callable[[], object]	the action that creates the resource when exists() is False.	required

provisio.primitives.update_if_confirmed

update_if_confirmed(ctx, *, describe, apply, prompt=None)

Apply a configuration change, asking for confirmation first when asked to.

Parameters:

Name	Type	Description	Default
describe	str	human phrase naming what is being configured.	required
apply	Callable[[], object]	the action that performs the update.	required
prompt	str | None	when provided, ctx.confirm is asked and the update is skipped if declined. Pass None (the step's choice, e.g. when the resource does not yet exist) to apply without asking.	None

provisio.primitives.poll_until

poll_until(ctx, *, describe, read_state, done, failed=(), timeout=300, interval=15, sleep=time.sleep)

Poll read_state until it reports a terminal state.

Returns the final state once it is in done. Raises ProvisioError if the state enters failed or the timeout elapses. sleep is injected so tests run instantly.

Command layer

provisio.command.CommandResult dataclass

The immutable outcome of running one external command.

Attributes:

Name	Type	Description
args	tuple[str, ...]	the argv that was run, e.g. ("az", "group", "show", ...).
returncode	int	the process exit code (0 conventionally means success).
stdout	str	captured standard output.
stderr	str	captured standard error.

ok property

ok

True when the command exited successfully (return code 0).

json

json()

Parse stdout as JSON.

The caller is responsible for having asked the underlying CLI for JSON output; this is a convenience, not a guarantee.

provisio.command.CommandRunner

Bases: Protocol

Runs an argv and returns a CommandResult. The only door to the outside.

Implementations share nothing but this contract — the real runner shells out to a subprocess; the test fake returns canned results. Everything that would touch a cloud goes through here, which is what makes the framework testable.

run

run(args, *, check=True)

Run args and return the result.

Parameters:

Name	Type	Description	Default
args	Sequence[str]	the argv to run, e.g. ["az", "group", "show", "--name", "rg"].	required
check	bool	when True a non-zero exit raises CommandFailedError; when False the failing CommandResult is returned (used by existence probes).	True

provisio.command.SubprocessCommandRunner

The real CommandRunner: resolve argv[0] on PATH, run it, capture output.

This is the only place subprocess lives. It performs no cloud-specific logic — it just runs whatever argv it is given (az ..., gh ..., aws ...) and returns a CommandResult.

provisio.cli_tool.CliTool

A thin, callable wrapper around one CLI binary.

Parameters:

Name	Type	Description	Default
binary	str	the executable name, e.g. "az".	required
runner	CommandRunner	the CommandRunner all invocations go through.	required
json_flags	Sequence[str]	flags appended by json() to request JSON output.	()
tsv_flags	Sequence[str]	flags appended by tsv() to request tab/line output.	()

__call__

__call__(*args, check=True)

Run <binary> <args> and return stdout, stripped.

capture

capture(*args, check=True)

Run <binary> <args> and return the full CommandResult.

exists

exists(*args)

Run a probe (check=False) and return True iff it exited 0.

This is the building block of idempotency: it never raises on a non-zero exit, so a "does this resource exist?" query is just a failed probe.

json

json(*args, check=True)

Run with the configured JSON flags appended and parse stdout as JSON.

tsv

tsv(*args, check=True)

Run with the configured TSV flags appended and return stdout, stripped.

Observability

provisio.reporting.Reporter

Bases: Protocol

Receives human-facing progress events.

step marks the start of the n-th of total steps; the rest are one-line status messages under the current step.

provisio.logging.configure_audit_log

configure_audit_log(*, destination='stdout', fmt='text', redactor=None, level=logging.INFO)

Attach a single audit handler to the provisio logger and return it.

Idempotent: re-calling replaces provisio's own handler rather than stacking handlers, so repeated configuration (e.g. across CLI invocations or tests) does not duplicate output.

Parameters:

Name	Type	Description	Default
destination	str | PathLike[str]	"stdout", "stderr", or a file path.	'stdout'
fmt	Literal['text', 'json']	"text" (human) or "json" (structured, for log ingestion).	'text'
redactor	Redactor | None	masks secret values in every line when provided.	None
level	int	minimum level to emit (INFO captures the audit trail).	INFO

provisio.logging.Redactor

Masks known secret values in any string before it is logged.

Fed from the settings schema's secret=True fields (wired later); here it is just an ordered set of literal secrets and a mask token. Empty/blank secrets are ignored so an unset secret never blanks out unrelated text.

provisio.logging.get_logger

get_logger(name=_ROOT_LOGGER_NAME)

Return a provisio audit logger. Reusable by consumers for their own code.

Pass a dotted child name (e.g. "provisio.command" or "provisio.myapp") to log under the provisio namespace.

Settings & application

provisio.settings.Setting dataclass

One configurable option, declared once.

Parameters:

Name	Type	Description	Default
name	str	the resolved attribute name (e.g. "resource_group").	required
env	str | None	environment variable to read as a fallback (e.g. "AZURE_RESOURCE_GROUP").	None
default	str | None	value used when nothing else is provided (None => required).	None
secret	bool	if True, prompted hidden and masked in the audit log.	False
help	str	human description (shown in prompts and generated CLI help).	''
flag	str | None	CLI flag override; defaults to --<name-with-dashes>.	None

provisio.settings.resolve_settings

resolve_settings(specs, *, cli, env, prompter=None)

Resolve each Setting to a value.

Precedence: explicit CLI value → environment variable → (interactive prompt | default). A required setting (no default) that is not supplied raises ProvisioError in non-interactive mode.

Parameters:

Name	Type	Description	Default
specs	Sequence[Setting]	the settings schema.	required
cli	Mapping[str, Any]	mapping of name -> value from the CLI (missing => not supplied).	required
env	Mapping[str, str]	environment mapping (usually os.environ).	required
prompter	Prompter | None	when provided, missing values are prompted (interactive mode); when None, the run is non-interactive (CI).	None

provisio.application.InfraApplication dataclass

Everything needed to generate an app's CLI and run its plan.

provisio.application.CliToolSpec dataclass

Declares a vendor CLI the app drives (e.g. az, gh).

build_cli verifies binary_name is on PATH before running, then builds a CliTool registered under name on the context.

State & diff

provisio.state.State dataclass

The last-applied declaration snapshot.

provisio.state.StateStore

Bases: Protocol

Loads/saves the last-applied State. Pluggable (local file default).

provisio.state.FileStateStore

Persists state as human-readable JSON on the local filesystem.

provisio.diff.diff

diff(previous, plan, settings)

Compare plan + settings against previous state.

provisio.diff.Diff dataclass

The result of comparing a definition against persisted state.

is_empty property

is_empty

True when nothing changed (CI can skip re-provisioning).

Testing utilities

provisio.testing.FakeCommandRunner

A CommandRunner that records argv and returns canned results.

Register expectations with stub(*prefix, stdout=..., returncode=...): a call whose argv starts with a registered prefix returns that result (with its args set to the actual argv). Unmatched calls default to success with empty stdout. Inspect what ran via .calls and .issued(*prefix).

Honours check exactly like the real runner, so idempotency tests can assert both the skip branch (probe returns success) and the create branch.

stub

stub(*prefix, stdout='', returncode=0, stderr='')

Register a canned result for any call whose argv starts with prefix.

Returns self so stubs can be chained.

issued

issued(*prefix)

True if any recorded call's argv starts with prefix.

provisio.testing.RecordingReporter

A Reporter that records events as (level, message) tuples.

Tests assert on .events (order + content) or .messages(level). For a step event the message is the step title.

messages

messages(level)

All messages recorded at level, in order.