Infrastructure Automation Basics

10 examples to get you started with Infra Automation - 7 basic and 3 intermediate.

Prerequisites

uv venv && source .venv/bin/activate
uv pip install "ansible>=10.0" "pulumi>=3.0" "cdktf>=0.20"

• Python 3.14.0 with an isolated virtual environment.
• AWS credentials via AWS_PROFILE or IAM role for cloud examples (optional for local-only runs).

Basic Examples

1. Define Desired State in Code

Infrastructure automation starts by describing what should exist, not how to click through a console.

# desired_state.py
from dataclasses import dataclass
 
@dataclass(frozen=True)
class BucketSpec:
    name: str
    versioning: bool = True
 
DESIRED = BucketSpec(name="app-logs-prod", versioning=True)

• Treat desired state as data structures you can diff, test, and review in PRs.
• Console changes drift away from the repo - code is the source of truth.
• Start small: one resource type before orchestrating an entire VPC.

Related: Configuration Management - templating and environments

2. Idempotent Create-or-Update

Run the same script twice and end in the same state - no duplicate resources.

import boto3
 
def ensure_bucket(s3, name: str) -> None:
    buckets = [b["Name"] for b in s3.list_buckets().get("Buckets", [])]
    if name not in buckets:
        s3.create_bucket(Bucket=name)
    s3.put_bucket_versioning(
        Bucket=name,
        VersioningConfiguration={"Status": "Enabled"},
    )
 
session = boto3.Session()
ensure_bucket(session.client("s3"), "demo-idempotent-bucket")

• Check existence before create - bare create_bucket fails on the second run.
• put_bucket_versioning is naturally idempotent - safe to call every time.
• Return structured results (created vs updated) for logging and audits.

Related: Provisioning Cloud Resources - repeatable provisioning

3. Ansible Ad-Hoc Idempotency

Ansible modules converge toward declared state automatically.

ansible localhost -m ansible.builtin.file -a "path=/tmp/iac-demo state=directory mode=0755"
ansible localhost -m ansible.builtin.file -a "path=/tmp/iac-demo state=directory mode=0755"

• Second run reports ok with changed=0 - that is idempotency in action.
• Prefer modules (file, copy, apt) over raw shell unless no module exists.
• Use -C (check mode) to preview changes without applying them.

Related: Ansible - playbooks, inventories, and roles

4. Version Infrastructure Definitions

Store infra code in git with the same rigor as application code.

git init infra-demo && cd infra-demo
mkdir -p stacks/prod
echo 'name = "prod"' > stacks/prod/Pulumi.yaml
git add . && git commit -m "chore: initial prod stack"

• Every apply should map to a commit SHA for rollback and blame.
• Tag releases (infra-v2026.07.09) when promoting to production.
• Never edit live cloud resources without a matching code change.

Related: Pulumi (Python IaC) - real Python for cloud resources

5. Separate Environments

Keep dev, staging, and prod configs distinct but structurally identical.

from dataclasses import dataclass
 
@dataclass(frozen=True)
class EnvConfig:
    name: str
    bucket_suffix: str
    enable_deletion_protection: bool
 
ENVS = {
    "dev": EnvConfig("dev", "-dev", False),
    "prod": EnvConfig("prod", "-prod", True),
}

• Parameterize names and flags - do not fork entire stacks per environment.
• Prod-only safeguards (deletion_protection=True) belong in config, not comments.
• Load env config from ENVIRONMENT variable or CLI flag, never hardcode prod.

Related: Configuration Management - secrets and templating

6. Plan Before Apply

Preview changes before mutating shared infrastructure.

cd stacks/prod
pulumi preview   # or: terraform plan

• Plans show creates, updates, and deletes - review destructive changes explicitly.
• Block applies when plan output includes unexpected delete on stateful resources.
• Save plan artifacts in CI logs for incident investigation.

Related: Testing Infrastructure Code - dry-runs and policy checks

7. Structured Automation Logging

Log what changed, who ran it, and the outcome - not just success/failure.

import json
import logging
 
logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("iac")
 
def apply(spec: dict) -> dict:
    result = {"resource": spec["name"], "action": "updated", "changed": True}
    log.info(json.dumps({"event": "iac_apply", **result}))
    return result
 
apply({"name": "app-logs-prod"})

• JSON logs parse cleanly in CloudWatch, Loki, or ELK.
• Include correlation IDs when automation is triggered from CI pipelines.
• Never log secrets, tokens, or full Terraform state blobs.

Related: Infrastructure Automation Best Practices - auditable infra rules

Intermediate Examples

8. Drift Detection with boto3

Compare live AWS state against your declared spec.

import boto3
 
def detect_versioning_drift(bucket: str, expected: str = "Enabled") -> bool:
    s3 = boto3.client("s3")
    resp = s3.get_bucket_versioning(Bucket=bucket)
    actual = resp.get("Status", "Suspended")
    return actual != expected
 
if detect_versioning_drift("app-logs-prod"):
    raise SystemExit("drift detected: versioning mismatch")

• Run drift checks on a schedule, not only at deploy time.
• Emit metrics or alerts when drift is found - silent drift becomes outage debt.
• Pair detection with automated reconcile jobs where safe.

Related: Provisioning Cloud Resources - reviewable infrastructure

9. Wrap Terraform with Python (CDKTF)

Generate Terraform JSON from typed Python constructs.

from cdktf import App, TerraformStack
from constructs import Construct
 
class MiniStack(TerraformStack):
    def __init__(self, scope: Construct, ns: str):
        super().__init__(scope, ns)
        # Add providers and resources here - cdktf synth writes terraform JSON
 
app = App()
MiniStack(app, "mini")
app.synth()

• CDKTF gives you loops, functions, and tests over HCL templates.
• cdktf synth output is still plain Terraform - ops teams can inspect it.
• Pin provider versions in generated config for reproducible plans.

Related: Terraform with Python - CDKTF and wrapping Terraform

10. Policy Check Before Apply

Reject non-compliant plans in CI before they touch cloud accounts.

def validate_tags(tags: dict) -> list[str]:
    required = {"Environment", "Owner", "CostCenter"}
    missing = required - set(tags)
    return [f"missing tag: {t}" for t in sorted(missing)]
 
errors = validate_tags({"Environment": "prod"})
assert errors, "policy should fail without Owner and CostCenter"

• Encode org policies as pure Python functions - fast unit tests, no cloud calls.
• Run policy checks on plan JSON in CI gates alongside ruff and pytest.
• Fail closed: block apply when policy validation returns errors.

Related: Testing Infrastructure Code - linting and policy checks

Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.