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_PROFILEor 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_bucketfails on the second run. put_bucket_versioningis naturally idempotent - safe to call every time.- Return structured results (
createdvsupdated) 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
okwithchanged=0- that is idempotency in action. - Prefer modules (
file,copy,apt) over rawshellunless 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
ENVIRONMENTvariable or CLI flag, never hardcodeprod.
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
deleteon 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 synthoutput 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
ruffandpytest. - 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+.